glusterd: Add cksum file (quota.cksum) for quota.conf file

.. and use quota checksum and version to validate one's own quota store
config

cli: cleanup quota-list-all implementation

Also,
change the format in which we store the directory quota configurations.
We store the list of gfids as 16 byte unsigned chars, in binary mode.

Original-author: Krishnan Parthasarathi <kparthas@redhat.com>

glusterd: Store quota checksum and version in quota.cksum

Quota version is incremented AND quota checksum is computed everytime
quota.conf is modified.

The checksum and versions are also retrieved
from store into memory whenever glusterd is restarted.

glusterd: Unlink quota.conf and quota.cksum on quota disable

Also destroy volinfo->quota_conf_shandle and reset it to NULL,
and reset volinfo->quota_conf_version to 0 in memory.

Change-Id: Ie71da3a75bc80e1ffddf4f2e38a99a48ad4de164
Signed-off-by: Krutika Dhananjay <kdhananj@redhat.com>

6 files changed, 514 insertions, 96 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index 24bd41ad..e67208cf 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -580,6 +580,8 @@ glusterd_quota_disable (glusterd_volinfo_t *volinfo, char **op_errstr)
         if (ret)
                 goto out;
 
+        (void) glusterd_clean_up_quota_store (volinfo);
+
         ret = 0;
 out:
         if (ret && op_errstr && !*op_errstr)
@@ -671,6 +673,47 @@ out:
         return ret;
 }
 
+#define QUOTA_CONF_HEADER                                                \
+        "GlusterFS Quota conf | version: v%d.%d\n"
+
+static int
+glusterd_store_quota_conf_skip_header (xlator_t *this, int fd)
+{
+        char buf[PATH_MAX] = {0,};
+
+        snprintf (buf, sizeof(buf)-1, QUOTA_CONF_HEADER, 1, 1);
+        return gf_skip_header_section (fd, strlen (buf));
+}
+
+static int
+glusterd_store_quota_conf_stamp_header (xlator_t *this, int fd)
+{
+        char buf[PATH_MAX]  = {0,};
+        int  buf_len        = 0;
+        ssize_t  ret        = -1;
+        ssize_t  written    = 0;
+
+        snprintf (buf, sizeof(buf)-1, QUOTA_CONF_HEADER, 1, 1);
+        buf_len = strlen (buf);
+        for (written = 0; written != buf_len; written += ret) {
+                ret = write (fd, buf + written, buf_len - written);
+                if (ret == -1) {
+                        goto out;
+                }
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+static int
+glusterd_update_quota_conf_version (glusterd_volinfo_t *volinfo)
+{
+        volinfo->quota_conf_version++;
+        return 0;
+}
+
 static int
 glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
                              char *gfid_str, int opcode, char **op_errstr)
@@ -679,13 +722,13 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
         int                count                 = 0;
         xlator_t          *this                  = NULL;
         glusterd_conf_t   *conf                  = NULL;
-        char               buf[256]              = {0,};
+        unsigned char      buf[16]               = {0,};
         int                fd                    = -1;
-        FILE              *conf_filep            = NULL;
-        FILE              *tmp_filep             = NULL;
+        int                conf_fd               = -1;
+        size_t             entry_sz              = 16;
         uuid_t             gfid                  = {0,};
-        uuid_t             gfid_iter             = {0,};
         gf_boolean_t       found                 = _gf_false;
+        gf_boolean_t       modified              = _gf_false;
 
 
         this = THIS;
@@ -702,23 +745,67 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
                 ret = -1;
                 goto out;
         }
-        tmp_filep = fdopen (fd, "r+");
 
-        conf_filep = fopen (volinfo->quota_conf_shandle->path, "r");
-        if (conf_filep == NULL) {
+        conf_fd = open (volinfo->quota_conf_shandle->path, O_RDONLY);
+        if (conf_fd == -1) {
                 ret = -1;
                 goto out;
         }
 
-        while (fscanf (conf_filep, "%s", buf) != EOF) {
-                count++;
 
-                uuid_parse (buf, gfid_iter);
+        ret = glusterd_store_quota_conf_skip_header (this, conf_fd);
+        if (ret) {
+                goto out;
+        }
 
-                if (uuid_compare (gfid, gfid_iter))
-                        fprintf (tmp_filep, "%s\n", buf);
-                else
+        ret = glusterd_store_quota_conf_stamp_header (this, fd);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to add header to tmp "
+                        "file.");
+                goto out;
+        }
+        //gfid is stored as 16 bytes of 'raw' data
+        entry_sz = 16;
+        for (;;) {
+                ret = read (conf_fd, (void*)&buf, entry_sz) ;
+                if (ret <= 0) {
+                        //Finished reading all entries in the conf file
+                        break;
+                }
+                if (ret != 16) {
+                        //This should never happen. We must have a multiple of
+                        //entry_sz bytes in our configuration file.
+                        gf_log (this->name, GF_LOG_CRITICAL, "Quota "
+                                "configuration store may be corrupt.");
+                        ret = -1;
+                        goto out;
+                }
+                count++;
+                if (uuid_compare (gfid, buf)) {
+                        /*If the gfids don't match, write @buf into tmp file. */
+                        ret = write (fd, (void*) buf, entry_sz);
+                        if (ret == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                                        "write %s into quota configuration.",
+                                        uuid_utoa (buf));
+                                goto out;
+                        }
+                } else {
+                        /*If a match is found, write @buf into tmp file for
+                         * limit-usage only.
+                         */
+                        if (opcode == GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) {
+                                ret = write (fd, (void *) buf, entry_sz);
+                                if (ret == -1) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Failed to write %s into quota "
+                                                "configuration.",
+                                                uuid_utoa (buf));
+                                        goto out;
+                                }
+                        }
                         found = _gf_true;
+                }
         }
 
         switch (opcode) {
@@ -727,15 +814,24 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
                          * count = 0 implies that the conf file is empty.
                          * In this case, we directly go ahead and write gfid_str
                          * into the tmp file.
-                         * If count is non-zero and found is false, then we
-                         * append gfid_str to the end of the file.
-                         * If count is non-zero and found is true, then again
+                         * If count is non-zero and found is false, limit is
+                         * being set on a gfid for the first time. So
                          * append gfid_str to the end of the file.
-                         *
-                         * In short, in all of the above cases, append gfid_str
-                         * to the end of the file.
                          */
-                        fprintf (tmp_filep, "%s\n", gfid_str);
+                        if ((count == 0) ||
+                            ((count > 0) && (found == _gf_false))) {
+                                memcpy (buf, gfid, 16);
+                                ret = write (fd, (void *) buf, entry_sz);
+                                if (ret == -1) {
+                                        gf_log (this->name, GF_LOG_ERROR,
+                                                "Failed to write %s into quota "
+                                                "configuration.",
+                                                uuid_utoa (buf));
+                                        goto out;
+                                }
+                                modified = _gf_true;
+                        }
+
                         break;
 
                 case GF_QUOTA_OPTION_TYPE_REMOVE:
@@ -757,12 +853,15 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
                                 goto out;
                         } else {
                                 if (!found) {
-                                gf_asprintf (op_errstr, "Error. gfid %s for "
-                                             "path %s not found in store",
-                                             gfid_str, path);
-                                ret = -1;
-                                goto out;
+                                        gf_asprintf (op_errstr, "Error. gfid %s"
+                                                     " for path %s not found in"
+                                                     " store", gfid_str, path);
+                                        ret = -1;
+                                        goto out;
+                                } else {
+                                        modified = _gf_true;
                                 }
+
                         }
                         break;
 
@@ -771,17 +870,40 @@ glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
                         break;
         }
 
+        if (modified)
+                glusterd_update_quota_conf_version (volinfo);
+
         ret = 0;
 out:
-        if (conf_filep)
-                fclose (conf_filep);
-        if (tmp_filep)
-                fclose (tmp_filep);
+        if (conf_fd != -1) {
+                close (conf_fd);
+        }
+
+        if (fd != -1) {
+                close (fd);
+        }
 
-        if (ret && (fd > 0))
+        if (ret && (fd > 0)) {
                 gf_store_unlink_tmppath (volinfo->quota_conf_shandle);
-        else if (!ret)
+        } else if (!ret) {
                 ret = gf_store_rename_tmppath (volinfo->quota_conf_shandle);
+                if (modified) {
+                        ret = glusterd_compute_cksum (volinfo, _gf_true);
+                        if (ret) {
+                                gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                                        "compute cksum for quota conf file");
+                                goto out;
+                        }
+
+                        ret = glusterd_store_save_quota_version_and_cksum
+                                                                      (volinfo);
+                        if (ret) {
+                                gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                                        "store quota version and cksum");
+                                goto out;
+                        }
+                }
+        }
 
         return ret;
 }
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.c b/xlators/mgmt/glusterd/src/glusterd-store.c
index b9f4792d..db0829bb 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.c
+++ b/xlators/mgmt/glusterd/src/glusterd-store.c
@@ -1099,7 +1099,7 @@ glusterd_store_volinfo (glusterd_volinfo_t *volinfo, glusterd_volinfo_ver_ac_t a
                 goto out;
 
         //checksum should be computed at the end
-        ret = glusterd_volume_compute_cksum (volinfo);
+        ret = glusterd_compute_cksum (volinfo, _gf_false);
         if (ret)
                 goto out;
 
@@ -1977,7 +1977,23 @@ glusterd_store_retrieve_volume (char    *volname)
         if (ret)
                 goto out;
 
-        ret = glusterd_volume_compute_cksum (volinfo);
+        ret = glusterd_compute_cksum (volinfo, _gf_false);
+        if (ret)
+                goto out;
+
+        ret = glusterd_store_retrieve_quota_version (volinfo);
+        if (ret)
+                goto out;
+
+        ret = glusterd_store_create_quota_conf_sh_on_absence (volinfo);
+        if (ret)
+                goto out;
+
+        ret = glusterd_compute_cksum (volinfo, _gf_true);
+        if (ret)
+                goto out;
+
+        ret = glusterd_store_save_quota_version_and_cksum (volinfo);
         if (ret)
                 goto out;
 
@@ -2572,3 +2588,106 @@ out:
         gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
         return ret;
 }
+
+int
+glusterd_store_retrieve_quota_version (glusterd_volinfo_t *volinfo)
+{
+        int                 ret                  = -1;
+        uint32_t            version              = 0;
+        char                cksum_path[PATH_MAX] = {0,};
+        char                path[PATH_MAX]       = {0,};
+        char               *version_str          = NULL;
+        char               *tmp                  = NULL;
+        xlator_t           *this                 = NULL;
+        glusterd_conf_t    *conf                 = NULL;
+        gf_store_handle_t  *handle               = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GLUSTERD_GET_VOLUME_DIR (path, volinfo, conf);
+        snprintf (cksum_path, sizeof (cksum_path), "%s/%s", path,
+                  GLUSTERD_VOL_QUOTA_CKSUM_FILE);
+
+        ret = gf_store_handle_new (cksum_path, &handle);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get store handle "
+                        "for %s", cksum_path);
+                goto out;
+        }
+
+        ret = gf_store_retrieve_value (handle, "version", &version_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG, "Version absent");
+                ret = 0;
+                goto out;
+        }
+
+        version = strtoul (version_str, &tmp, 10);
+         if (version < 0) {
+                 gf_log (this->name, GF_LOG_DEBUG, "Invalid version number");
+                 goto out;
+        }
+        volinfo->quota_conf_version = version;
+        ret = 0;
+
+out:
+        if (version_str)
+                GF_FREE (version_str);
+        gf_store_handle_destroy (handle);
+        return ret;
+}
+
+int
+glusterd_store_save_quota_version_and_cksum (glusterd_volinfo_t *volinfo)
+{
+        int                 ret                  = -1;
+        char                cksum_path[PATH_MAX] = {0,};
+        char                path[PATH_MAX]       = {0,};
+        xlator_t           *this                 = NULL;
+        glusterd_conf_t    *conf                 = NULL;
+        char                buf[256]             = {0,};
+        int                 fd                   = -1;
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GLUSTERD_GET_VOLUME_DIR (path, volinfo, conf);
+        snprintf (cksum_path, sizeof (cksum_path), "%s/%s", path,
+                  GLUSTERD_VOL_QUOTA_CKSUM_FILE);
+
+        fd = open (cksum_path, O_RDWR | O_APPEND | O_CREAT| O_TRUNC, 0600);
+
+        if (-1 == fd) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to open %s,"
+                        "Reason: %s", cksum_path, strerror (errno));
+                ret = -1;
+                goto out;
+        }
+
+        snprintf (buf, sizeof (buf)-1, "%u", volinfo->quota_conf_cksum);
+        ret = gf_store_save_value (fd, "cksum", buf);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to store cksum");
+                goto out;
+        }
+
+        memset (buf, 0, sizeof (buf));
+        snprintf (buf, sizeof (buf)-1, "%u", volinfo->quota_conf_version);
+        ret = gf_store_save_value (fd, "version", buf);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to store version");
+                goto out;
+        }
+
+        ret = 0;
+
+out:
+        if (fd != -1)
+                close (fd);
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.h b/xlators/mgmt/glusterd/src/glusterd-store.h
index 2dc05157..20cbf659 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.h
+++ b/xlators/mgmt/glusterd/src/glusterd-store.h
@@ -127,4 +127,11 @@ glusterd_store_options (xlator_t *this, dict_t *opts);
 
 int32_t
 glusterd_store_create_quota_conf_sh_on_absence (glusterd_volinfo_t *volinfo);
+
+int
+glusterd_store_retrieve_quota_version (glusterd_volinfo_t *volinfo);
+
+int
+glusterd_store_save_quota_version_and_cksum (glusterd_volinfo_t *volinfo);
+
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index 000a6b10..66f4c971 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -1538,89 +1538,85 @@ glusterd_sort_and_redirect (const char *src_filepath, int dest_fd)
 }
 
 int
-glusterd_volume_compute_cksum (glusterd_volinfo_t  *volinfo)
-{
-        int32_t                 ret = -1;
-        glusterd_conf_t         *priv = NULL;
-        char                    path[PATH_MAX] = {0,};
-        char                    cksum_path[PATH_MAX] = {0,};
-        char                    filepath[PATH_MAX] = {0,};
-        int                     fd = -1;
-        uint32_t                cksum = 0;
-        char                    buf[4096] = {0,};
+glusterd_volume_compute_cksum (glusterd_volinfo_t  *volinfo, char *cksum_path,
+                               char *filepath, gf_boolean_t is_quota_conf,
+                               uint32_t *cs)
+{
+        int32_t                 ret                     = -1;
+        uint32_t                cksum                   = 0;
+        int                     fd                      = -1;
+        int                     sort_fd                 = 0;
         char                    sort_filepath[PATH_MAX] = {0};
-        gf_boolean_t            unlink_sortfile = _gf_false;
-        int                     sort_fd = 0;
-        xlator_t               *this = NULL;
+        char                   *cksum_path_final        = NULL;
+        char                    buf[4096]               = {0,};
+        gf_boolean_t            unlink_sortfile         = _gf_false;
+        glusterd_conf_t        *priv                    = NULL;
+        xlator_t               *this                    = NULL;
 
         GF_ASSERT (volinfo);
         this = THIS;
         priv = THIS->private;
         GF_ASSERT (priv);
 
-        GLUSTERD_GET_VOLUME_DIR (path, volinfo, priv);
-
-        snprintf (cksum_path, sizeof (cksum_path), "%s/%s",
-                  path, GLUSTERD_CKSUM_FILE);
-
         fd = open (cksum_path, O_RDWR | O_APPEND | O_CREAT| O_TRUNC, 0600);
 
         if (-1 == fd) {
-                gf_log (this->name, GF_LOG_ERROR, "Unable to open %s, errno: %d",
-                        cksum_path, errno);
+                gf_log (this->name, GF_LOG_ERROR, "Unable to open %s,"
+                        " errno: %d", cksum_path, errno);
                 ret = -1;
                 goto out;
         }
 
-        snprintf (filepath, sizeof (filepath), "%s/%s", path,
-                  GLUSTERD_VOLUME_INFO_FILE);
-        snprintf (sort_filepath, sizeof (sort_filepath), "/tmp/%s.XXXXXX",
-                  volinfo->volname);
+        if (!is_quota_conf) {
+                snprintf (sort_filepath, sizeof (sort_filepath),
+                          "/tmp/%s.XXXXXX", volinfo->volname);
 
-        sort_fd = mkstemp (sort_filepath);
-        if (sort_fd < 0) {
-                gf_log (this->name, GF_LOG_ERROR, "Could not generate temp "
-                        "file, reason: %s for volume: %s", strerror (errno),
-                        volinfo->volname);
-                goto out;
-        } else {
-                unlink_sortfile = _gf_true;
-        }
+                sort_fd = mkstemp (sort_filepath);
+                if (sort_fd < 0) {
+                        gf_log (this->name, GF_LOG_ERROR, "Could not generate "
+                                "temp file, reason: %s for volume: %s",
+                                strerror (errno), volinfo->volname);
+                        goto out;
+                } else {
+                        unlink_sortfile = _gf_true;
+                }
 
-        /* sort the info file, result in sort_filepath */
+                /* sort the info file, result in sort_filepath */
 
-        ret = glusterd_sort_and_redirect (filepath, sort_fd);
-        if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "sorting info file failed");
-                goto out;
-        }
+                ret = glusterd_sort_and_redirect (filepath, sort_fd);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "sorting info file "
+                                "failed");
+                        goto out;
+                }
 
-        ret = close (sort_fd);
-        if (ret)
-                goto out;
+                ret = close (sort_fd);
+                if (ret)
+                        goto out;
+        }
 
-        ret = get_checksum_for_path (sort_filepath, &cksum);
+        cksum_path_final = is_quota_conf ? filepath : sort_filepath;
 
+        ret = get_checksum_for_path (cksum_path_final, &cksum);
         if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "Unable to get checksum"
-                        " for path: %s", sort_filepath);
+                gf_log (this->name, GF_LOG_ERROR, "unable to get "
+                        "checksum for path: %s", cksum_path_final);
                 goto out;
         }
-
-        snprintf (buf, sizeof (buf), "%s=%u\n", "info", cksum);
-        ret = write (fd, buf, strlen (buf));
-
-        if (ret <= 0) {
-                ret = -1;
-                goto out;
+        if (!is_quota_conf) {
+                snprintf (buf, sizeof (buf), "%s=%u\n", "info", cksum);
+                ret = write (fd, buf, strlen (buf));
+                if (ret <= 0) {
+                        ret = -1;
+                        goto out;
+                }
         }
 
         ret = get_checksum_for_file (fd, &cksum);
-
         if (ret)
                 goto out;
 
-        volinfo->cksum = cksum;
+        *cs = cksum;
 
 out:
         if (fd > 0)
@@ -1632,6 +1628,54 @@ out:
         return ret;
 }
 
+int glusterd_compute_cksum (glusterd_volinfo_t *volinfo,
+                               gf_boolean_t is_quota_conf)
+{
+        int               ret                  = -1;
+        uint32_t          cs                   = 0;
+        char              cksum_path[PATH_MAX] = {0,};
+        char              path[PATH_MAX]       = {0,};
+        char              filepath[PATH_MAX]   = {0,};
+        glusterd_conf_t  *conf                 = NULL;
+        xlator_t         *this                 = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GLUSTERD_GET_VOLUME_DIR (path, volinfo, conf);
+
+        if (is_quota_conf) {
+                snprintf (cksum_path, sizeof (cksum_path), "%s/%s", path,
+                          GLUSTERD_VOL_QUOTA_CKSUM_FILE);
+                snprintf (filepath, sizeof (filepath), "%s/%s", path,
+                          GLUSTERD_VOLUME_QUOTA_CONFIG);
+        } else {
+                snprintf (cksum_path, sizeof (cksum_path), "%s/%s", path,
+                          GLUSTERD_CKSUM_FILE);
+                snprintf (filepath, sizeof (filepath), "%s/%s", path,
+                          GLUSTERD_VOLUME_INFO_FILE);
+        }
+
+        ret = glusterd_volume_compute_cksum (volinfo, cksum_path, filepath,
+                                             is_quota_conf, &cs);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to compute checksum "
+                        "for volume %s", volinfo->volname);
+                goto out;
+        }
+
+        if (is_quota_conf)
+                volinfo->quota_conf_cksum = cs;
+        else
+                volinfo->cksum = cs;
+
+        ret = 0;
+out:
+        return ret;
+}
+
 int
 _add_dict_to_prdict (dict_t *this, char *key, data_t *value, void *data)
 {
@@ -1977,6 +2021,18 @@ glusterd_vol_add_quota_conf_to_dict (glusterd_volinfo_t *volinfo, dict_t* load,
         if (ret)
                 goto out;
 
+        snprintf (key, sizeof(key)-1, "volume%d.quota-cksum", vol_idx);
+        key[sizeof(key)-1] = '\0';
+        ret = dict_set_uint32 (load, key, volinfo->quota_conf_cksum);
+        if (ret)
+                goto out;
+
+        snprintf (key, sizeof(key)-1, "volume%d.quota-version", vol_idx);
+        key[sizeof(key)-1] = '\0';
+        ret = dict_set_uint32 (load, key, volinfo->quota_conf_version);
+        if (ret)
+                goto out;
+
         ret = 0;
 out:
         if (fp)
@@ -2049,11 +2105,17 @@ glusterd_compare_friend_volume (dict_t *vols, int32_t count, int32_t *status,
         glusterd_volinfo_t      *volinfo = NULL;
         char                    *volname = NULL;
         uint32_t                cksum = 0;
+        uint32_t                quota_cksum = 0;
+        uint32_t                quota_version = 0;
         int32_t                 version = 0;
+        xlator_t                *this = NULL;
 
         GF_ASSERT (vols);
         GF_ASSERT (status);
 
+        this = THIS;
+        GF_ASSERT (this);
+
         snprintf (key, sizeof (key), "volume%d.name", count);
         ret = dict_get_str (vols, key, &volname);
         if (ret)
@@ -2076,7 +2138,7 @@ glusterd_compare_friend_volume (dict_t *vols, int32_t count, int32_t *status,
         if (version > volinfo->version) {
                 //Mismatch detected
                 ret = 0;
-                gf_log ("", GF_LOG_ERROR, "Version of volume %s differ."
+                gf_log (this->name, GF_LOG_ERROR, "Version of volume %s differ."
                         "local version = %d, remote version = %d on peer %s",
                         volinfo->volname, volinfo->version, version, hostname);
                 *status = GLUSTERD_VOL_COMP_UPDATE_REQ;
@@ -2096,17 +2158,66 @@ glusterd_compare_friend_volume (dict_t *vols, int32_t count, int32_t *status,
 
         if (cksum != volinfo->cksum) {
                 ret = 0;
-                gf_log ("", GF_LOG_ERROR, "Cksums of volume %s differ."
+                gf_log (this->name, GF_LOG_ERROR, "Cksums of volume %s differ."
                         " local cksum = %u, remote cksum = %u on peer %s",
                         volinfo->volname, volinfo->cksum, cksum, hostname);
                 *status = GLUSTERD_VOL_COMP_RJT;
                 goto out;
         }
 
+        memset (key, 0, sizeof (key));
+        snprintf (key, sizeof (key), "volume%d.quota-version", count);
+        ret = dict_get_uint32 (vols, key, &quota_version);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG, "quota-version key absent for"
+                        " volume %s in peer %s's response", volinfo->volname,
+                        hostname);
+                ret = 0;
+        } else {
+                if (quota_version > volinfo->quota_conf_version) {
+                        //Mismatch detected
+                        ret = 0;
+                        gf_log (this->name, GF_LOG_ERROR, "Quota configuration "
+                                "versions of volume %s differ. "
+                                "local version = %d, remote version = %d "
+                                "on peer %s", volinfo->volname,
+                                volinfo->quota_conf_version, quota_version,
+                                hostname);
+                        *status = GLUSTERD_VOL_COMP_UPDATE_REQ;
+                        goto out;
+                } else if (quota_version < volinfo->quota_conf_version) {
+                        *status = GLUSTERD_VOL_COMP_SCS;
+                        goto out;
+                }
+        }
+
+        //Now, versions are same, compare cksums.
+        //
+        memset (key, 0, sizeof (key));
+        snprintf (key, sizeof (key), "volume%d.quota-cksum", count);
+        ret = dict_get_uint32 (vols, key, &quota_cksum);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG, "quota checksum absent for "
+                        "volume %s in peer %s's response", volinfo->volname,
+                        hostname);
+                ret = 0;
+        } else {
+                if (quota_cksum != volinfo->quota_conf_cksum) {
+                        ret = 0;
+                        gf_log (this->name, GF_LOG_ERROR, "Cksums of quota "
+                                "configurations of volume %s differ. "
+                                "local cksum = %u, remote cksum = %u on "
+                                "peer %s", volinfo->volname,
+                                volinfo->quota_conf_cksum, quota_cksum,
+                                hostname);
+                        *status = GLUSTERD_VOL_COMP_RJT;
+                        goto out;
+                }
+        }
         *status = GLUSTERD_VOL_COMP_SCS;
 
 out:
-        gf_log ("", GF_LOG_DEBUG, "Returning with ret: %d, status: %d",
+        gf_log (this->name, GF_LOG_DEBUG, "Returning with ret: %d, status: %d",
                 ret, *status);
         return ret;
 }
@@ -2565,8 +2676,10 @@ glusterd_import_quota_conf (dict_t *vols, int vol_idx,
         char    key[PATH_MAX]    = {0};
         char    *gfid_str        = NULL;
 
-        if (!glusterd_is_volume_quota_enabled (new_volinfo))
+        if (!glusterd_is_volume_quota_enabled (new_volinfo)) {
+                (void) glusterd_clean_up_quota_store (new_volinfo);
                 return 0;
+        }
 
         ret = glusterd_store_create_quota_conf_sh_on_absence (new_volinfo);
         if (ret)
@@ -2583,6 +2696,19 @@ glusterd_import_quota_conf (dict_t *vols, int vol_idx,
                 goto out;
         }
 
+        snprintf (key, sizeof (key)-1, "volume%d.quota-cksum", vol_idx);
+        key[sizeof(key)-1] = '\0';
+        ret = dict_get_uint32 (vols, key, &new_volinfo->quota_conf_cksum);
+        if (ret)
+                gf_log (THIS->name, GF_LOG_DEBUG, "Failed to get quota cksum");
+
+        snprintf (key, sizeof (key)-1, "volume%d.quota-version", vol_idx);
+        key[sizeof(key)-1] = '\0';
+        ret = dict_get_uint32 (vols, key, &new_volinfo->quota_conf_version);
+        if (ret)
+                gf_log (THIS->name, GF_LOG_DEBUG, "Failed to get quota "
+                                                  "version");
+
         snprintf (key, sizeof (key)-1, "volume%d.gfid-count", vol_idx);
         key[sizeof(key)-1] = '\0';
         ret = dict_get_int32 (vols, key, &gfid_count);
@@ -2608,13 +2734,20 @@ glusterd_import_quota_conf (dict_t *vols, int vol_idx,
 
         ret = gf_store_rename_tmppath (new_volinfo->quota_conf_shandle);
 
-out:
-        if (tmp_fp) {
-                fclose (tmp_fp);
+        ret = 0;
 
-        } else if (fd >= 0) {
+out:
+        if (fd != -1)
                 close (fd);
 
+        if (!ret) {
+                ret = glusterd_compute_cksum (new_volinfo, _gf_true);
+                if (ret)
+                        goto out;
+
+                ret = glusterd_store_save_quota_version_and_cksum (new_volinfo);
+                if (ret)
+                        goto out;
         }
 
         if (ret && (fd > 0)) {
@@ -8126,3 +8259,33 @@ glusterd_validate_and_set_gfid (dict_t *op_ctx, dict_t *req_dict,
 out:
         return ret;
 }
+
+void
+glusterd_clean_up_quota_store (glusterd_volinfo_t *volinfo)
+{
+        char      voldir[PATH_MAX]         = {0,};
+        char      quota_confpath[PATH_MAX] = {0,};
+        char      cksum_path[PATH_MAX]     = {0,};
+        xlator_t  *this                    = NULL;
+        glusterd_conf_t *conf              = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GLUSTERD_GET_VOLUME_DIR (voldir, volinfo, conf);
+
+        snprintf (quota_confpath, sizeof (quota_confpath), "%s/%s", voldir,
+                  GLUSTERD_VOLUME_QUOTA_CONFIG);
+        snprintf (cksum_path, sizeof (cksum_path), "%s/%s", voldir,
+                  GLUSTERD_VOL_QUOTA_CKSUM_FILE);
+
+        unlink (quota_confpath);
+        unlink (cksum_path);
+
+        gf_store_handle_destroy (volinfo->quota_conf_shandle);
+        volinfo->quota_conf_shandle = NULL;
+        volinfo->quota_conf_version = 0;
+
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index 1074c46d..3be5458d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -158,7 +158,8 @@ int32_t
 glusterd_compare_friend_data (dict_t  *vols, int32_t *status, char *hostname);
 
 int
-glusterd_volume_compute_cksum (glusterd_volinfo_t  *volinfo);
+glusterd_compute_cksum (glusterd_volinfo_t  *volinfo,
+                        gf_boolean_t is_quota_conf);
 
 void
 glusterd_get_nodesvc_volfile (char *server, char *workdir,
@@ -593,4 +594,7 @@ glusterd_all_volumes_with_quota_stopped ();
 
 int
 glusterd_reconfigure_quotad ();
+
+void
+glusterd_clean_up_quota_store (glusterd_volinfo_t *volinfo);
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 516409dc..3fb303ca 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -285,7 +285,9 @@ struct glusterd_volinfo_ {
         glusterd_replace_brick_t rep_brick;
 
         int                     version;
+        uint32_t                quota_conf_version;
         uint32_t                cksum;
+        uint32_t                quota_conf_cksum;
         gf_transport_type       transport_type;
         gf_transport_type   nfs_transport_type;
 
@@ -344,6 +346,7 @@ enum glusterd_vol_comp_status_ {
 #define GLUSTERD_VOLUME_RBSTATE_FILE "rbstate"
 #define GLUSTERD_BRICK_INFO_DIR "bricks"
 #define GLUSTERD_CKSUM_FILE "cksum"
+#define GLUSTERD_VOL_QUOTA_CKSUM_FILE "quota.cksum"
 #define GLUSTERD_TRASH "trash"
 #define GLUSTERD_NODE_STATE_FILE "node_state.info"