glusterd: Implement persistent store for quota

Setting quota limits on a given directory will cause glusterd to
persist the gfid of the path in WD/vols/<volname>/quota.conf.

Also, executing 'quota remove' will cause glusterd to remove
the gfid of the given path.

This is needed for implementing the 'list-all' variant of 'quota list'
command.

To-Do:

1. Exchange quota.conf when a new node is added into the cluster;
2. Unlink quota.conf on disabling quota (?)

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

8 files changed, 550 insertions, 24 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index bee75f7d..4ad1ef8d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -3809,7 +3809,8 @@ glusterd_op_stage_validate (glusterd_op_t op, dict_t *dict, char **op_errstr,
                         break;
 
                 case GD_OP_QUOTA:
-                        ret = glusterd_op_stage_quota (dict, op_errstr);
+                        ret = glusterd_op_stage_quota (dict, op_errstr,
+                                                       rsp_dict);
                         break;
 
                 case GD_OP_STATUS_VOLUME:
diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index 1be768af..24bd41ad 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -671,14 +671,130 @@ out:
         return ret;
 }
 
+static int
+glusterd_store_quota_config (glusterd_volinfo_t *volinfo, char *path,
+                             char *gfid_str, int opcode, char **op_errstr)
+{
+        int                ret                   = -1;
+        int                count                 = 0;
+        xlator_t          *this                  = NULL;
+        glusterd_conf_t   *conf                  = NULL;
+        char               buf[256]              = {0,};
+        int                fd                    = -1;
+        FILE              *conf_filep            = NULL;
+        FILE              *tmp_filep             = NULL;
+        uuid_t             gfid                  = {0,};
+        uuid_t             gfid_iter             = {0,};
+        gf_boolean_t       found                 = _gf_false;
+
+
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        uuid_parse (gfid_str, gfid);
+
+        glusterd_store_create_quota_conf_sh_on_absence (volinfo);
+
+        fd = gf_store_mkstemp (volinfo->quota_conf_shandle);
+        if (fd < 0) {
+                ret = -1;
+                goto out;
+        }
+        tmp_filep = fdopen (fd, "r+");
+
+        conf_filep = fopen (volinfo->quota_conf_shandle->path, "r");
+        if (conf_filep == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        while (fscanf (conf_filep, "%s", buf) != EOF) {
+                count++;
+
+                uuid_parse (buf, gfid_iter);
+
+                if (uuid_compare (gfid, gfid_iter))
+                        fprintf (tmp_filep, "%s\n", buf);
+                else
+                        found = _gf_true;
+        }
+
+        switch (opcode) {
+                case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
+                        /*
+                         * 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
+                         * 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);
+                        break;
+
+                case GF_QUOTA_OPTION_TYPE_REMOVE:
+                        /*
+                         * count = 0 is not a valid scenario and must be treated
+                         * as error.
+                         * If count is non-zero and found is false, then it is
+                         * an error.
+                         * If count is non-zero and found is true, take no
+                         * action, by virtue of which the gfid is as good as
+                         * deleted from the store.
+                         */
+                        if (count == 0) {
+                                gf_asprintf (op_errstr, "Cannot remove limit on"
+                                             " %s. The quota configuration file"
+                                             " for volume %s is empty.", path,
+                                             volinfo->volname);
+                                ret = -1;
+                                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;
+                                }
+                        }
+                        break;
+
+                default:
+                        ret = 0;
+                        break;
+        }
+
+        ret = 0;
+out:
+        if (conf_filep)
+                fclose (conf_filep);
+        if (tmp_filep)
+                fclose (tmp_filep);
+
+        if (ret && (fd > 0))
+                gf_store_unlink_tmppath (volinfo->quota_conf_shandle);
+        else if (!ret)
+                ret = gf_store_rename_tmppath (volinfo->quota_conf_shandle);
+
+        return ret;
+}
+
 int32_t
 glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict,
-                            char **op_errstr)
+                            int opcode, char **op_errstr)
 {
         int32_t          ret                = -1;
         char            *path               = NULL;
         char            *hard_limit         = NULL;
         char            *soft_limit         = NULL;
+        char            *gfid_str           = NULL;
         xlator_t        *this               = NULL;
 
         this = THIS;
@@ -726,6 +842,19 @@ glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict,
                 if (ret)
                         goto out;
         }
+
+        ret = dict_get_str (dict, "gfid", &gfid_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get gfid of path "
+                        "%s", path);
+                goto out;
+        }
+
+        ret = glusterd_store_quota_config (volinfo, path, gfid_str, opcode,
+                                           op_errstr);
+        if (ret)
+                goto out;
+
         ret = 0;
 out:
 
@@ -771,10 +900,11 @@ out:
 
 int32_t
 glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
-                              char **op_errstr)
+                              int opcode, char **op_errstr)
 {
         int32_t         ret                   = -1;
         char            *path                 = NULL;
+        char            *gfid_str             = NULL;
         xlator_t        *this                 = NULL;
 
         this = THIS;
@@ -807,6 +937,20 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
                 if (ret)
                         goto out;
         }
+
+        ret = dict_get_str (dict, "gfid", &gfid_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get gfid of path "
+                        "%s", path);
+                goto out;
+        }
+
+        ret = glusterd_store_quota_config (volinfo, path, gfid_str, opcode,
+                                           op_errstr);
+        if (ret)
+                goto out;
+
+
         ret = 0;
 
 out:
@@ -939,12 +1083,12 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                         break;
 
                 case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
-                        ret = glusterd_quota_limit_usage (volinfo, dict,
+                        ret = glusterd_quota_limit_usage (volinfo, dict, type,
                                                           op_errstr);
                         goto out;
 
                 case GF_QUOTA_OPTION_TYPE_REMOVE:
-                        ret = glusterd_quota_remove_limits (volinfo, dict,
+                        ret = glusterd_quota_remove_limits (volinfo, dict, type,
                                                             op_errstr);
                         goto out;
 
@@ -1028,16 +1172,112 @@ out:
         return ret;
 }
 
+/*
+ * glusterd_get_gfid_from_brick() fetches the 'trusted.gfid' attribute of @path
+ * from each brick in the backend and places the same in the rsp_dict with the
+ * keys being gfid0, gfid1, gfid2 and so on. The absence of @path in the backend
+ * is not treated as error.
+ */
+static int
+glusterd_get_gfid_from_brick (dict_t *dict, glusterd_volinfo_t *volinfo,
+                              dict_t *rsp_dict, char **op_errstr)
+{
+        int                    ret                    = -1;
+        int                    count                  = 0;
+        char                  *path                   = NULL;
+        char                   backend_path[PATH_MAX] = {0,};
+        xlator_t              *this                   = NULL;
+        glusterd_conf_t       *priv                   = NULL;
+        glusterd_brickinfo_t  *brickinfo              = NULL;
+        char                   key[256]               = {0,};
+        char                  *gfid_str               = NULL;
+        uuid_t                 gfid;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        ret = dict_get_str (dict, "path", &path);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get path");
+                goto out;
+        }
+
+        list_for_each_entry (brickinfo, &volinfo->bricks, brick_list) {
+                ret = glusterd_resolve_brick (brickinfo);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, FMTSTR_RESOLVE_BRICK,
+                                brickinfo->hostname, brickinfo->path);
+                        goto out;
+                }
+
+                if (uuid_compare (brickinfo->uuid, MY_UUID))
+                        continue;
+
+                if (volinfo->backend == GD_VOL_BK_BD)
+                        continue;
+
+                snprintf (backend_path, sizeof (backend_path), "%s%s",
+                          brickinfo->path, path);
+
+                ret = gf_lstat_dir (backend_path, NULL);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_INFO, "Failed to find "
+                                "directory %s. Reason : %s", backend_path,
+                                strerror (errno));
+                        ret = 0;
+                        continue;
+                }
+                ret = sys_lgetxattr (backend_path, GFID_XATTR_KEY, gfid, 16);
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_INFO, "Failed to get "
+                                "extended attribute %s for directory %s. "
+                                "Reason : %s", GFID_XATTR_KEY, backend_path,
+                                     strerror (errno));
+                        ret = 0;
+                        continue;
+                }
+                snprintf (key, sizeof (key), "gfid%d", count);
+
+                gfid_str = gf_strdup (uuid_utoa (gfid));
+                if (!gfid_str) {
+                        ret = -1;
+                        goto out;
+                }
+
+                ret = dict_set_dynstr (rsp_dict, key, gfid_str);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to place "
+                                "gfid of %s in dict", backend_path);
+                        GF_FREE (gfid_str);
+                        goto out;
+                }
+                count++;
+        }
+
+        ret = dict_set_int32 (rsp_dict, "count", count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set count");
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
 int
-glusterd_op_stage_quota (dict_t *dict, char **op_errstr)
+glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
 {
-        int                ret           = 0;
-        char              *volname       = NULL;
-        gf_boolean_t       exists        = _gf_false;
-        int                type          = 0;
-        dict_t            *ctx           = NULL;
-        xlator_t          *this          = NULL;
-        glusterd_conf_t   *priv          = NULL;
+        int                 ret           = 0;
+        char               *volname       = NULL;
+        gf_boolean_t        exists        = _gf_false;
+        int                 type          = 0;
+        dict_t             *ctx           = NULL;
+        xlator_t           *this          = NULL;
+        glusterd_conf_t    *priv          = NULL;
+        glusterd_volinfo_t *volinfo       = NULL;
 
         this = THIS;
         GF_ASSERT (this);
@@ -1059,6 +1299,12 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr)
                 ret = -1;
                 goto out;
         }
+        ret = glusterd_volinfo_find (volname, &volinfo);
+        if (ret) {
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
+                goto out;
+        }
+
 
         ret = dict_get_int32 (dict, "type", &type);
         if (ret) {
@@ -1089,6 +1335,15 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr)
                  }
          }
 
+         if (type == GF_QUOTA_OPTION_TYPE_LIMIT_USAGE ||
+             type == GF_QUOTA_OPTION_TYPE_REMOVE) {
+                 ret = glusterd_get_gfid_from_brick (dict, volinfo, rsp_dict,
+                                                     op_errstr);
+                 if (ret)
+                         goto out;
+        }
+        ret = 0;
+
  out:
         if (ret && op_errstr && *op_errstr)
                 gf_log (this->name, GF_LOG_ERROR, "%s", *op_errstr);
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.c b/xlators/mgmt/glusterd/src/glusterd-store.c
index 71c8428e..b9f4792d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.c
+++ b/xlators/mgmt/glusterd/src/glusterd-store.c
@@ -683,6 +683,21 @@ glusterd_store_node_state_path_set (glusterd_volinfo_t *volinfo,
                   GLUSTERD_NODE_STATE_FILE);
 }
 
+static void
+glusterd_store_quota_conf_path_set (glusterd_volinfo_t *volinfo,
+                                    char *quota_conf_path, size_t len)
+{
+        char    voldirpath[PATH_MAX] = {0,};
+        GF_ASSERT (volinfo);
+        GF_ASSERT (quota_conf_path);
+        GF_ASSERT (len <= PATH_MAX);
+
+        glusterd_store_voldirpath_set (volinfo, voldirpath,
+                                       sizeof (voldirpath));
+        snprintf (quota_conf_path, len, "%s/%s", voldirpath,
+                  GLUSTERD_VOLUME_QUOTA_CONFIG);
+}
+
 int32_t
 glusterd_store_create_rbstate_shandle_on_absence (glusterd_volinfo_t *volinfo)
 {
@@ -728,6 +743,22 @@ glusterd_store_create_nodestate_sh_on_absence (glusterd_volinfo_t *volinfo)
 }
 
 int32_t
+glusterd_store_create_quota_conf_sh_on_absence (glusterd_volinfo_t *volinfo)
+{
+        char            quota_conf_path[PATH_MAX] = {0};
+        int32_t         ret                       = 0;
+
+        GF_ASSERT (volinfo);
+
+        glusterd_store_quota_conf_path_set (volinfo, quota_conf_path,
+                                            sizeof (quota_conf_path));
+        ret =
+          gf_store_handle_create_on_absence (&volinfo->quota_conf_shandle,
+                                             quota_conf_path);
+
+        return ret;
+}
+int32_t
 glusterd_store_brickinfos (glusterd_volinfo_t *volinfo, int vol_fd)
 {
         int32_t                 ret = 0;
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.h b/xlators/mgmt/glusterd/src/glusterd-store.h
index e4f9e4a0..2dc05157 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.h
+++ b/xlators/mgmt/glusterd/src/glusterd-store.h
@@ -124,4 +124,7 @@ glusterd_store_retrieve_options (xlator_t *this);
 
 int32_t
 glusterd_store_options (xlator_t *this, dict_t *opts);
+
+int32_t
+glusterd_store_create_quota_conf_sh_on_absence (glusterd_volinfo_t *volinfo);
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-syncop.c b/xlators/mgmt/glusterd/src/glusterd-syncop.c
index 7270dbcb..4fc05877 100644
--- a/xlators/mgmt/glusterd/src/glusterd-syncop.c
+++ b/xlators/mgmt/glusterd/src/glusterd-syncop.c
@@ -261,14 +261,18 @@ glusterd_syncop_aggr_rsp_dict (glusterd_op_t op, dict_t *aggr, dict_t *rsp)
 
         break;
 
-        case GD_OP_QUOTA:
         case GD_OP_CLEARLOCKS_VOLUME:
                 ret = glusterd_use_rsp_dict (aggr, rsp);
                 if (ret)
                         goto out;
-
         break;
 
+        case GD_OP_QUOTA:
+                ret = glusterd_volume_quota_copy_to_op_ctx_dict (aggr, rsp);
+                if (ret)
+                        goto out;
+                break;
+
         case GD_OP_SYS_EXEC:
                 ret = glusterd_sys_exec_output_rsp_dict (aggr, rsp);
                 if (ret)
@@ -878,7 +882,7 @@ gd_stage_op_phase (struct list_head *peers, glusterd_op_t op, dict_t *op_ctx,
                 goto stage_done;
         }
 
-        if ((op == GD_OP_REPLACE_BRICK)) {
+        if ((op == GD_OP_REPLACE_BRICK || op == GD_OP_QUOTA)) {
                 ret = glusterd_syncop_aggr_rsp_dict (op, op_ctx, rsp_dict);
                 if (ret) {
                         gf_log (this->name, GF_LOG_ERROR, "%s",
@@ -913,6 +917,10 @@ stage_done:
                                                op, req_dict, op_ctx);
                 peer_cnt++;
         }
+
+        gf_log (this->name, GF_LOG_DEBUG, "Sent stage op req for 'Volume %s' "
+                "to %d peers", gd_op_list[op], peer_cnt);
+
         gd_synctask_barrier_wait((&args), peer_cnt);
 
         if (args.errstr)
@@ -922,9 +930,14 @@ stage_done:
 
         ret = args.op_ret;
 
-        gf_log (this->name, GF_LOG_DEBUG, "Sent stage op req for 'Volume %s' "
-                "to %d peers", gd_op_list[op], peer_cnt);
 out:
+        if ((ret == 0) && (op == GD_OP_QUOTA)) {
+                ret = glusterd_validate_and_set_gfid (op_ctx, req_dict,
+                                                      op_errstr);
+                if (ret)
+                        goto out;
+        }
+
         if (rsp_dict)
                 dict_unref (rsp_dict);
         return ret;
@@ -943,6 +956,7 @@ gd_commit_op_phase (struct list_head *peers, glusterd_op_t op, dict_t *op_ctx,
         uuid_t              tmp_uuid        = {0};
         char                *errstr         = NULL;
         struct syncargs     args            = {0};
+        int                 type            = GF_QUOTA_OPTION_TYPE_NONE;
 
         this = THIS;
         rsp_dict = dict_new ();
@@ -956,15 +970,28 @@ gd_commit_op_phase (struct list_head *peers, glusterd_op_t op, dict_t *op_ctx,
                 hostname = "localhost";
                 goto commit_done;
         }
-        if (op != GD_OP_SYNC_VOLUME) {
-                ret =  glusterd_syncop_aggr_rsp_dict (op, op_ctx, rsp_dict);
+
+        if (op == GD_OP_QUOTA) {
+                ret = dict_get_int32 (op_ctx, "type", &type);
                 if (ret) {
-                        gf_log (this->name, GF_LOG_ERROR, "%s",
-                                "Failed to aggregate response "
-                                "from node/brick");
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to get "
+                                "opcode");
                         goto out;
                 }
         }
+
+        if (((op == GD_OP_QUOTA) && (type == GF_QUOTA_OPTION_TYPE_LIST)) ||
+            ((op != GD_OP_SYNC_VOLUME) && (op != GD_OP_QUOTA))) {
+
+                ret =  glusterd_syncop_aggr_rsp_dict (op, op_ctx,
+                                                      rsp_dict);
+                if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "%s", "Failed to aggregate "
+                        "response from node/brick");
+                goto out;
+                }
+        }
+
         dict_unref (rsp_dict);
         rsp_dict = NULL;
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index d42bac9e..bb6bcf83 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -7245,6 +7245,86 @@ _profile_volume_add_brick_rsp (dict_t *this, char *key, data_t *value,
 }
 
 int
+glusterd_volume_quota_copy_to_op_ctx_dict (dict_t *dict, dict_t *rsp_dict)
+{
+        int        ret            = -1;
+        int        i              = 0;
+        int        count          = 0;
+        int        rsp_dict_count = 0;
+        char      *uuid_str       = NULL;
+        char      *uuid_str_dup   = NULL;
+        char       key[256]       = {0,};
+        xlator_t  *this           = NULL;
+        int        type           = GF_QUOTA_OPTION_TYPE_NONE;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        ret = dict_get_int32 (dict, "type", &type);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get quota opcode");
+                goto out;
+        }
+
+        if ((type != GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) &&
+            (type != GF_QUOTA_OPTION_TYPE_REMOVE)) {
+                dict_copy (rsp_dict, dict);
+                ret = 0;
+                goto out;
+        }
+
+        ret = dict_get_int32 (rsp_dict, "count", &rsp_dict_count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get the count of "
+                        "gfids from the rsp dict");
+                goto out;
+        }
+        ret = dict_set_int32 (dict, "count", count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set count of gfids"
+                        " from req dict");
+                goto out;
+        }
+
+        for (i = 0; i < rsp_dict_count; i++) {
+                snprintf (key, sizeof(key)-1, "gfid%d", i);
+
+                ret = dict_get_str (rsp_dict, key, &uuid_str);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to get gfid "
+                                "from rsp dict");
+                        goto out;
+                }
+
+                snprintf (key, sizeof (key)-1, "gfid%d", i + count);
+
+                uuid_str_dup = gf_strdup (uuid_str);
+                if (!uuid_str_dup) {
+                        ret = -1;
+                        goto out;
+                }
+
+                ret = dict_set_dynstr (dict, key, uuid_str_dup);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to set gfid "
+                                "from rsp dict into req dict");
+                        GF_FREE (uuid_str_dup);
+                        goto out;
+                }
+        }
+
+        ret = dict_set_int32 (dict, "count", rsp_dict_count + count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,  "Failed to set aggregated "
+                        "count in req dict");
+                goto out;
+        }
+
+out:
+        return ret;
+}
+
+int
 glusterd_profile_volume_brick_rsp (void *pending_entry,
                                    dict_t *rsp_dict, dict_t *op_ctx,
                                    char **op_errstr, gd_node_type type)
@@ -7786,3 +7866,123 @@ glusterd_is_volume_quota_enabled (glusterd_volinfo_t *volinfo)
 {
         return (glusterd_volinfo_get_boolean (volinfo, VKEY_FEATURES_QUOTA));
 }
+
+int
+glusterd_validate_and_set_gfid (dict_t *op_ctx, dict_t *req_dict,
+                                char **op_errstr)
+{
+        int        ret           = -1;
+        int        count         = 0;
+        int        i             = 0;
+        int        op_code       = GF_QUOTA_OPTION_TYPE_NONE;
+        uuid_t     uuid1         = {0};
+        uuid_t     uuid2         = {0,};
+        char      *path          = NULL;
+        char       key[256]      = {0,};
+        char      *uuid1_str     = NULL;
+        char      *uuid1_str_dup = NULL;
+        char      *uuid2_str     = NULL;
+        xlator_t  *this          = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        ret = dict_get_int32 (op_ctx, "type", &op_code);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get quota opcode");
+                goto out;
+        }
+
+        if ((op_code != GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) &&
+            (op_code != GF_QUOTA_OPTION_TYPE_REMOVE)) {
+                ret = 0;
+                goto out;
+        }
+
+        ret = dict_get_str (op_ctx, "path", &path);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get path");
+                goto out;
+        }
+
+        ret = dict_get_int32 (op_ctx, "count", &count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get count");
+                goto out;
+        }
+
+        /* If count is 0, fail the command with ENOENT.
+         *
+         * If count is 1, treat gfid0 as the gfid on which the operation
+         * is to be performed and resume the command.
+         *
+         * if count > 1, get the 0th gfid from the op_ctx and,
+         * compare it with the remaining 'count -1' gfids.
+         * If they are found to be the same, set gfid0 in the op_ctx and
+         * resume the operation, else error out.
+         */
+
+        if (count == 0) {
+                gf_asprintf (op_errstr, "Failed to get trusted.gfid attribute "
+                             "on path %s. Reason : %s", path,
+                             strerror (ENOENT));
+                ret = -1;
+                goto out;
+        }
+
+        snprintf (key, sizeof (key) - 1, "gfid%d", 0);
+
+        ret = dict_get_str (op_ctx, key, &uuid1_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to get key '%s'",
+                        key);
+                goto out;
+        }
+
+        uuid_parse (uuid1_str, uuid1);
+
+        for (i = 1; i < count; i++) {
+                snprintf (key, sizeof (key)-1, "gfid%d", i);
+
+                ret = dict_get_str (op_ctx, key, &uuid2_str);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to get key "
+                                "'%s'", key);
+                        goto out;
+                }
+
+                uuid_parse (uuid2_str, uuid2);
+
+                if (uuid_compare (uuid1, uuid2)) {
+                        gf_asprintf (op_errstr, "gfid mismatch between %s and "
+                                     "%s for path %s", uuid1_str, uuid2_str,
+                                     path);
+                        ret = -1;
+                        goto out;
+                }
+        }
+
+        if (i == count) {
+                uuid1_str_dup = gf_strdup (uuid1_str);
+                if (!uuid1_str_dup) {
+                        ret = -1;
+                        goto out;
+                }
+
+                ret = dict_set_dynstr (req_dict, "gfid", uuid1_str_dup);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to set gfid");
+                        GF_FREE (uuid1_str_dup);
+                        goto out;
+                }
+        } else {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to iterate through %d"
+                        " entries in the req dict", count);
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index 4c6c2e4b..9fe08d06 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -498,6 +498,9 @@ int
 glusterd_volume_rebalance_use_rsp_dict (dict_t *aggr, dict_t *rsp_dict);
 int
 glusterd_volume_heal_use_rsp_dict (dict_t *aggr, dict_t *rsp_dict);
+
+int
+glusterd_volume_quota_copy_to_op_ctx_dict (dict_t *aggr, dict_t *rsp);
 int
 _profile_volume_add_brick_rsp (dict_t *this, char *key, data_t *value,
                              void *data);
@@ -542,6 +545,10 @@ int
 glusterd_generate_and_set_task_id (dict_t *dict, char *key);
 
 int
+glusterd_validate_and_set_gfid (dict_t *op_ctx, dict_t *req_dict,
+                                char **op_errstr);
+
+int
 glusterd_copy_uuid_to_dict (uuid_t uuid, dict_t *dict, char *key);
 
 gf_boolean_t
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 02d2c85e..516409dc 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -276,6 +276,7 @@ struct glusterd_volinfo_ {
         gf_store_handle_t *shandle;
         gf_store_handle_t *rb_shandle;
         gf_store_handle_t *node_state_shandle;
+        gf_store_handle_t *quota_conf_shandle;
 
         /* Defrag/rebalance related */
         glusterd_rebalance_t    rebal;
@@ -336,6 +337,7 @@ enum glusterd_vol_comp_status_ {
 #define GLUSTERD_DEFAULT_WORKDIR "/var/lib/glusterd"
 #define GLUSTERD_DEFAULT_PORT    GF_DEFAULT_BASE_PORT
 #define GLUSTERD_INFO_FILE      "glusterd.info"
+#define GLUSTERD_VOLUME_QUOTA_CONFIG "quota.conf"
 #define GLUSTERD_VOLUME_DIR_PREFIX "vols"
 #define GLUSTERD_PEER_DIR_PREFIX "peers"
 #define GLUSTERD_VOLUME_INFO_FILE "info"
@@ -717,7 +719,7 @@ int glusterd_op_sys_exec (dict_t *dict, char **op_errstr, dict_t *rsp_dict);
 int glusterd_op_stage_gsync_create (dict_t *dict, char **op_errstr);
 int glusterd_op_gsync_create (dict_t *dict, char **op_errstr, dict_t *rsp_dict);
 int glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict);
-int glusterd_op_stage_quota (dict_t *dict, char **op_errstr);
+int glusterd_op_stage_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict);
 int glusterd_op_stage_replace_brick (dict_t *dict, char **op_errstr,
                                      dict_t *rsp_dict);
 int glusterd_op_replace_brick (dict_t *dict, dict_t *rsp_dict);