cli/glusterd: Changes to quota command Quota feature

 re-work.

Following are the cli commands that are new/re-worked:
======================================================

volume quota <VOLNAME> {enable|disable|list [<path> ...]|remove <path>| default-soft-limit <percent>} |
volume quota <VOLNAME> {limit-usage <path> <size> [<percent>]} |
volume quota <VOLNAME> {alert-time|soft-timeout|hard-timeout} {<time>}
volume status [all | <VOLNAME> [nfs|shd|<BRICK>|quotad]] [detail|clients|mem|inode|fd|callpool]
volume statedump <VOLNAME> [nfs|quotad] [all|mem|iobuf|callpool|priv|fd|inode|history]

glusterd changes:
=================
* Quota limits are now set as extended attributes by glusterd from
  the aux mount created by the cli.
* The gfids of the directories on which quota limits are set
  for a given volume are stored in
  /var/lib/glusterd/vols/<volname>/quota.conf file in binary format,
  and whose cksum and version is stored in
  /var/lib/glusterd/vols/<volname>/quota.cksum.

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

BUG: 969461
Change-Id: If32bba36c67f9c2a30417af9c6389045b2b7c13b
Signed-off-by: Krutika Dhananjay <kdhananj@redhat.com>
Signed-off-by: Raghavendra G <rgowdapp@redhat.com>
Reviewed-on: http://review.gluster.org/6003
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

1 files changed, 904 insertions, 278 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index cef2baf599c..56a24c74306 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -21,9 +21,27 @@
 #include "glusterd-utils.h"
 #include "glusterd-volgen.h"
 #include "run.h"
+#include "syscall.h"
+#include "byte-order.h"
+#include "compat-errno.h"
 
 #include <sys/wait.h>
 
+
+const char *gd_quota_op_list[GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT+1] = {
+        [GF_QUOTA_OPTION_TYPE_NONE]               = "none",
+        [GF_QUOTA_OPTION_TYPE_ENABLE]             = "enable",
+        [GF_QUOTA_OPTION_TYPE_DISABLE]            = "disable",
+        [GF_QUOTA_OPTION_TYPE_LIMIT_USAGE]        = "limit-usage",
+        [GF_QUOTA_OPTION_TYPE_REMOVE]             = "remove",
+        [GF_QUOTA_OPTION_TYPE_LIST]               = "list",
+        [GF_QUOTA_OPTION_TYPE_VERSION]            = "version",
+        [GF_QUOTA_OPTION_TYPE_ALERT_TIME]         = "alert-time",
+        [GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT]       = "soft-timeout",
+        [GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT]       = "hard-timeout",
+        [GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT] = "default-soft-limit",
+};
+
 int
 __glusterd_handle_quota (rpcsvc_request_t *req)
 {
@@ -31,15 +49,17 @@ __glusterd_handle_quota (rpcsvc_request_t *req)
         gf_cli_req                      cli_req = {{0,}};
         dict_t                         *dict = NULL;
         glusterd_op_t                   cli_op = GD_OP_QUOTA;
-        char                            operation[256] = {0, };
         char                           *volname = NULL;
         int32_t                         type = 0;
         char                            msg[2048] = {0,};
         xlator_t                       *this = NULL;
+        glusterd_conf_t                *conf = NULL;
 
         GF_ASSERT (req);
         this = THIS;
         GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
 
         ret = xdr_to_generic (req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
         if (ret < 0) {
@@ -82,23 +102,16 @@ __glusterd_handle_quota (rpcsvc_request_t *req)
                goto out;
         }
 
-        switch (type) {
-        case GF_QUOTA_OPTION_TYPE_ENABLE:
-                strncpy (operation, "enable", sizeof (operation));
-                break;
-
-        case GF_QUOTA_OPTION_TYPE_DISABLE:
-                strncpy (operation, "disable", sizeof (operation));
-                break;
-
-        case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
-                strncpy (operation, "limit-usage", sizeof (operation));
-                break;
+        if ((conf->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                snprintf (msg, sizeof (msg), "Cannot execute command. The "
+                         "cluster is operating at version %d. Quota command %s "
+                         "is unavailable in this version", conf->op_version,
+                         gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
+       }
 
-        case GF_QUOTA_OPTION_TYPE_REMOVE:
-                strncpy (operation, "remove", sizeof (operation));
-                break;
-        }
         ret = glusterd_op_begin_synctask (req, GD_OP_QUOTA, dict);
 
 out:
@@ -132,7 +145,6 @@ glusterd_check_if_quota_trans_enabled (glusterd_volinfo_t *volinfo)
         }
 
         if (flag == _gf_false) {
-                gf_log ("", GF_LOG_ERROR, "first enable the quota translator");
                 ret = -1;
                 goto out;
         }
@@ -141,13 +153,21 @@ out:
         return ret;
 }
 
-/* At the end of the function, the variable found will be set
+/* At the end of the function, the variable @found will be set
  * to true if the path to be removed was present in the limit-list,
  * else will be false.
+ *
+ * In addition, the function does the following things:
+ *
+ * a. places the path to be removed, if found, in @removed_path,
+ * b. places the new limit list formed after removing @path's entry, in
+ *    @new_list. If @path is not found, the input limit string @quota_limits is
+ *    dup'd as is and placed in @new_list.
  */
 int32_t
-_glusterd_quota_remove_limits (char **quota_limits, char *path,
-                               gf_boolean_t *found)
+_glusterd_quota_remove_limits (char *quota_limits, char *path,
+                               gf_boolean_t *found, char **new_list,
+                               char **removed_path)
 {
         int      ret      = 0;
         int      i        = 0;
@@ -158,14 +178,15 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
         int      flag     = 0;
         char    *limits   = NULL;
         char    *qlimits  = NULL;
+        char    *rp       = NULL;
 
         if (found != NULL)
                 *found = _gf_false;
 
-        if (*quota_limits == NULL)
+        if (quota_limits == NULL)
                 return -1;
 
-        qlimits = *quota_limits;
+        qlimits = quota_limits;
 
         pathlen = strlen (path);
 
@@ -192,6 +213,15 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
                 } else {
                         skiplen = size + 1;
                         size = len - i - size;
+                        if (removed_path) {
+                                rp = GF_CALLOC (skiplen, sizeof (char), gf_gld_mt_char);
+                                if (!rp) {
+                                        ret = -1;
+                                        goto out;
+                                }
+                                strncpy (rp, &qlimits[i], skiplen - 1);
+                                *removed_path = rp;
+                        }
                         memcpy ((void *) &limits [i], (void *) &qlimits [i + skiplen], size);
                         break;
                 }
@@ -200,42 +230,27 @@ _glusterd_quota_remove_limits (char **quota_limits, char *path,
                 size = 0;
         }
 
-        if (!flag) {
-                ret = 1;
-        } else {
-                len = strlen (limits);
-
-                if (len == 0) {
-                        GF_FREE (qlimits);
-
-                        *quota_limits = NULL;
-
-                        goto out;
-                }
-
-                if (limits[len - 1] == ',') {
-                        limits[len - 1] = '\0';
-                        len --;
-                }
-
-                GF_FREE (qlimits);
-
-                qlimits = GF_CALLOC (len + 1, sizeof (char), gf_gld_mt_char);
-
-                if (!qlimits) {
-                        ret = -1;
-                        goto out;
-                }
-
-                memcpy ((void *) qlimits, (void *) limits, len + 1);
+        len = strlen (limits);
+        if (len == 0)
+                goto out;
 
-                *quota_limits = qlimits;
+        if (limits[len - 1] == ',') {
+                limits[len - 1] = '\0';
+                len --;
+        }
 
-                ret = 0;
+        *new_list = GF_CALLOC (len + 1, sizeof (char), gf_gld_mt_char);
+        if (!*new_list) {
+                ret = -1;
+                goto out;
         }
 
+        memcpy ((void *) *new_list, (void *) limits, len + 1);
+        ret = 0;
 out:
         GF_FREE (limits);
+        if (ret != -1)
+                ret = flag ? 0 : 1;
 
         return ret;
 }
@@ -381,46 +396,38 @@ _glusterd_quota_get_limit_usages (glusterd_volinfo_t *volinfo,
 }
 
 int32_t
-glusterd_quota_get_limit_usages (glusterd_conf_t *priv,
-                                 glusterd_volinfo_t *volinfo,
-                                 char *volname,
-                                 dict_t *dict,
-                                 char **op_errstr,
-                                 dict_t *rsp_dict)
+glusterd_quota_get_default_soft_limit (glusterd_volinfo_t *volinfo,
+                                       dict_t *rsp_dict)
 {
-        int32_t i               = 0;
-        int32_t ret             = 0;
-        int32_t count           = 0;
-        char    *path           = NULL;
-        char    cmd_str [1024]  = {0, };
-        char   *ret_str         = NULL;
+        int32_t            ret             = 0;
+        xlator_t          *this            = NULL;
+        glusterd_conf_t   *conf            = NULL;
+        char              *default_limit   = NULL;
+        char              *val             = NULL;
 
         if (rsp_dict == NULL)
-                return 0;
-
-        ret = dict_get_int32 (dict, "count", &count);
-        if (ret < 0)
-                goto out;
+                return -1;
 
-        if (count == 0) {
-                ret_str = _glusterd_quota_get_limit_usages (volinfo, NULL,
-                                                            op_errstr);
-        } else {
-                i = 0;
-                while (count--) {
-                        snprintf (cmd_str, 1024, "path%d", i++);
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
 
-                        ret = dict_get_str (dict, cmd_str, &path);
-                        if (ret < 0)
-                                goto out;
+        ret = glusterd_volinfo_get (volinfo, "features.default-soft-limit",
+                                    &default_limit);
+        if (default_limit)
+                val = gf_strdup (default_limit);
+        else
+                val = gf_strdup ("80%");
 
-                        ret_str = _glusterd_quota_get_limit_usages (volinfo, path, op_errstr);
-                }
+        ret = dict_set_dynstr (rsp_dict, "default-soft-limit", val);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set default "
+                        "soft-limit into dict");
+                goto out;
         }
+        ret = 0;
 
-        if (ret_str) {
-                ret = dict_set_dynstr (rsp_dict, "limit_list", ret_str);
-        }
 out:
         return ret;
 }
@@ -431,54 +438,74 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 {
         int32_t         ret     = -1;
         char            *quota_status = NULL;
+        xlator_t        *this         = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", crawl, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, crawl, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         if (glusterd_is_volume_started (volinfo) == 0) {
                 *op_errstr = gf_strdup ("Volume is stopped, start volume "
                                         "to enable quota.");
+                ret = -1;
                 goto out;
         }
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == 0) {
                 *op_errstr = gf_strdup ("Quota is already enabled");
+                ret = -1;
                 goto out;
         }
 
         quota_status = gf_strdup ("on");
         if (!quota_status) {
-                gf_log ("", GF_LOG_ERROR, "memory allocation failed");
-                *op_errstr = gf_strdup ("Enabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "memory allocation failed");
+                ret = -1;
                 goto out;
         }
 
-        ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA, quota_status);
+        ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA,
+                               quota_status);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "dict set failed");
-                *op_errstr = gf_strdup ("Enabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "dict set failed");
                 goto out;
         }
 
-        *op_errstr = gf_strdup ("Enabling quota has been successful");
-
         *crawl = _gf_true;
 
         ret = 0;
 out:
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Enabling quota on volume %s has been "
+                             "unsuccessful", volinfo->volname);
         return ret;
 }
 
 int32_t
 glusterd_quota_disable (glusterd_volinfo_t *volinfo, char **op_errstr)
 {
-        int32_t  ret            = -1;
-        char    *quota_status   = NULL, *quota_limits = NULL;
+        int32_t    ret            = -1;
+        int        i              =  0;
+        char      *quota_status   = NULL;
+        char      *value          = NULL;
+        xlator_t  *this           = NULL;
+        glusterd_conf_t *conf     = NULL;
+        char *quota_options[]     = {"features.soft-timeout",
+                                     "features.hard-timeout",
+                                     "features.alert-time",
+                                     "features.default-soft-limit", NULL};
 
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
@@ -488,47 +515,345 @@ glusterd_quota_disable (glusterd_volinfo_t *volinfo, char **op_errstr)
 
         quota_status = gf_strdup ("off");
         if (!quota_status) {
-                gf_log ("", GF_LOG_ERROR, "memory allocation failed");
-                *op_errstr = gf_strdup ("Disabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "memory allocation failed");
+                ret = -1;
                 goto out;
         }
 
         ret = dict_set_dynstr (volinfo->dict, VKEY_FEATURES_QUOTA, quota_status);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "dict set failed");
-                *op_errstr = gf_strdup ("Disabling quota has been unsuccessful");
+                gf_log (this->name, GF_LOG_ERROR, "dict set failed");
                 goto out;
         }
 
-        *op_errstr = gf_strdup ("Disabling quota has been successful");
+        for (i = 0; quota_options [i]; i++) {
+                ret = glusterd_volinfo_get (volinfo, quota_options[i], &value);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_INFO, "failed to get option"
+                                                         " %s",
+                                                         quota_options[i]);
+                } else {
+                dict_del (volinfo->dict, quota_options[i]);
+                }
+        }
 
-        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
-                                    &quota_limits);
+        //Remove aux mount of the volume on every node in the cluster
+        ret = glusterd_remove_auxiliary_mount (volinfo->volname);
+        if (ret)
+                goto out;
+
+        (void) glusterd_clean_up_quota_store (volinfo);
+
+        ret = 0;
+out:
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Disabling quota on volume %s has been "
+                             "unsuccessful", volinfo->volname);
+        return ret;
+}
+
+static int
+glusterd_set_quota_limit (char *volname, char *path, char *hard_limit,
+                          char *soft_limit, char **op_errstr)
+{
+        int               ret                = -1;
+        xlator_t         *this               = NULL;
+        char              abspath[PATH_MAX]  = {0,};
+        glusterd_conf_t  *priv               = NULL;
+        double           soft_lim            = 0;
+
+        typedef struct quota_limits {
+                int64_t hl;
+                int64_t sl;
+        } __attribute__ ((__packed__)) quota_limits_t;
+
+	quota_limits_t existing_limit = {0,};
+	quota_limits_t new_limit = {0,};
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        snprintf (abspath, sizeof (abspath)-1, "/tmp/%s%s", volname, path);
+
+        ret = gf_lstat_dir (abspath, NULL);
         if (ret) {
-                gf_log ("", GF_LOG_WARNING, "failed to get the quota limits");
+                gf_asprintf (op_errstr, "Failed to find the directory %s. "
+                             "Reason : %s", abspath, strerror (errno));
+                goto out;
+        }
+
+        if (!soft_limit) {
+                ret = sys_lgetxattr (abspath,
+                                     "trusted.glusterfs.quota.limit-set",
+                                     (void *)&existing_limit,
+                                     sizeof (existing_limit));
+                if (ret < 0) {
+                        switch (errno) {
+                        case ENOATTR:
+                                existing_limit.sl = -1;
+                            break;
+                        default:
+                                gf_asprintf (op_errstr, "Failed to get the xattr "
+                                             "'trusted.glusterfs.quota.limit-set' from "
+                                             "%s. Reason : %s", abspath,
+                                             strerror (errno));
+                                goto out;
+                        }
+                } else {
+                        existing_limit.hl = ntoh64 (existing_limit.hl);
+                        existing_limit.sl = ntoh64 (existing_limit.sl);
+                }
+                new_limit.sl = existing_limit.sl;
+
         } else {
-                GF_FREE (quota_limits);
+                ret = gf_string2percent (soft_limit, &soft_lim);
+                if (ret)
+                        goto out;
+                new_limit.sl = soft_lim;
         }
 
-        dict_del (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE);
+        new_limit.sl = hton64 (new_limit.sl);
+
+        ret = gf_string2bytesize (hard_limit, (uint64_t*)&new_limit.hl);
+        if (ret)
+                goto out;
+
+        new_limit.hl = hton64 (new_limit.hl);
+
+        ret = sys_lsetxattr (abspath, "trusted.glusterfs.quota.limit-set",
+                             (char *)(void *)&new_limit, sizeof (new_limit), 0);
+        if (ret) {
+                gf_asprintf (op_errstr, "setxattr of "
+                             "'trusted.glusterfs.quota.limit-set' failed on %s."
+                             " Reason : %s", abspath, strerror (errno));
+                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)
+{
+        int                ret                   = -1;
+        int                count                 = 0;
+        xlator_t          *this                  = NULL;
+        glusterd_conf_t   *conf                  = NULL;
+        unsigned char      buf[16]               = {0,};
+        int                fd                    = -1;
+        int                conf_fd               = -1;
+        size_t             entry_sz              = 16;
+        uuid_t             gfid                  = {0,};
+        gf_boolean_t       found                 = _gf_false;
+        gf_boolean_t       modified              = _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;
+        }
+
+        conf_fd = open (volinfo->quota_conf_shandle->path, O_RDONLY);
+        if (conf_fd == -1) {
+                ret = -1;
+                goto out;
+        }
+
+
+        ret = glusterd_store_quota_conf_skip_header (this, conf_fd);
+        if (ret) {
+                goto out;
+        }
+
+        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) {
+                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, limit is
+                         * being set on a gfid for the first time. So
+                         * append gfid_str to the end of the file.
+                         */
+                        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:
+                        /*
+                         * 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;
+                                } else {
+                                        modified = _gf_true;
+                                }
+
+                        }
+                        break;
+
+                default:
+                        ret = 0;
+                        break;
+        }
+
+        if (modified)
+                glusterd_update_quota_conf_version (volinfo);
+
+        ret = 0;
+out:
+        if (conf_fd != -1) {
+                close (conf_fd);
+        }
+
+        if (fd != -1) {
+                close (fd);
+        }
+
+        if (ret && (fd > 0)) {
+                gf_store_unlink_tmppath (volinfo->quota_conf_shandle);
+        } 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;
+}
+
 int32_t
-glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict, char **op_errstr)
+glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict,
+                            int opcode, char **op_errstr)
 {
-        int32_t          ret    = -1;
-        char            *path   = NULL;
-        char            *limit  = NULL;
-        char            *value  = NULL;
-        char             msg [1024] = {0,};
-        char            *quota_limits = NULL;
+        int32_t          ret                = -1;
+        char            *path               = NULL;
+        char            *hard_limit         = NULL;
+        char            *soft_limit         = NULL;
+        char            *gfid_str           = NULL;
+        xlator_t        *this               = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", dict, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+
+        GF_VALIDATE_OR_GOTO (this->name, dict, out);
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
@@ -537,141 +862,217 @@ glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict, char **op
                 goto out;
         }
 
-        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
-                                    &quota_limits);
-        if (ret) {
-                gf_log ("", GF_LOG_ERROR, "failed to get the quota limits");
-                *op_errstr = gf_strdup ("failed to set limit");
-                goto out;
-        }
-
         ret = dict_get_str (dict, "path", &path);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
                 goto out;
         }
+        ret = gf_canonicalize_path (path);
+        if (ret)
+                goto out;
 
-        ret = dict_get_str (dict, "limit", &limit);
+        ret = dict_get_str (dict, "hard-limit", &hard_limit);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch hard limit");
                 goto out;
         }
 
-        if (quota_limits) {
-                ret = _glusterd_quota_remove_limits (&quota_limits, path, NULL);
-                if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
+        if (dict_get (dict, "soft-limit")) {
+                ret = dict_get_str (dict, "soft-limit", &soft_limit);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Unable to fetch "
+                                "soft limit");
                         goto out;
                 }
         }
 
-        if (quota_limits == NULL) {
-                ret = gf_asprintf (&value, "%s:%s", path, limit);
-                if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
-                        goto out;
-                }
-        } else {
-                ret = gf_asprintf (&value, "%s,%s:%s",
-                                   quota_limits, path, limit);
-                if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
+        if (is_origin_glusterd ()) {
+                ret = glusterd_set_quota_limit (volinfo->volname, path,
+                                                hard_limit, soft_limit,
+                                                op_errstr);
+                if (ret)
                         goto out;
-                }
+        }
 
-                GF_FREE (quota_limits);
+        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;
         }
 
-        quota_limits = value;
+        ret = glusterd_store_quota_config (volinfo, path, gfid_str, opcode,
+                                           op_errstr);
+        if (ret)
+                goto out;
+
+        ret = 0;
+out:
 
-        ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                            quota_limits);
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Failed to set hard limit on path %s "
+                             "for volume %s", path, volinfo->volname);
+        return ret;
+}
+
+static int
+glusterd_remove_quota_limit (char *volname, char *path, char **op_errstr)
+{
+        int               ret                = -1;
+        xlator_t         *this               = NULL;
+        char              abspath[PATH_MAX]  = {0,};
+        glusterd_conf_t  *priv               = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        snprintf (abspath, sizeof (abspath)-1, "/tmp/%s%s", volname, path);
+
+        ret = gf_lstat_dir (abspath, NULL);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_asprintf (op_errstr, "Failed to find the directory %s. "
+                             "Reason : %s", abspath, strerror (errno));
                 goto out;
         }
-        snprintf (msg, 1024, "limit set on %s", path);
-        *op_errstr = gf_strdup (msg);
 
+        ret = sys_lremovexattr (abspath, "trusted.glusterfs.quota.limit-set");
+        if (ret) {
+                gf_asprintf (op_errstr, "removexattr failed on %s. Reason : %s",
+                             abspath, strerror (errno));
+                goto out;
+        }
         ret = 0;
+
 out:
         return ret;
 }
 
 int32_t
-glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **op_errstr)
+glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict,
+                              int opcode, char **op_errstr)
 {
         int32_t         ret                   = -1;
-        char            str [PATH_MAX + 1024] = {0,};
-        char            *quota_limits         = NULL;
         char            *path                 = NULL;
-        gf_boolean_t     flag                 = _gf_false;
+        char            *gfid_str             = NULL;
+        xlator_t        *this                 = NULL;
 
-        GF_VALIDATE_OR_GOTO ("glusterd", dict, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", volinfo, out);
-        GF_VALIDATE_OR_GOTO ("glusterd", op_errstr, out);
+        this = THIS;
+        GF_ASSERT (this);
+
+        GF_VALIDATE_OR_GOTO (this->name, dict, out);
+        GF_VALIDATE_OR_GOTO (this->name, volinfo, out);
+        GF_VALIDATE_OR_GOTO (this->name, op_errstr, out);
 
         ret = glusterd_check_if_quota_trans_enabled (volinfo);
         if (ret == -1) {
-                *op_errstr = gf_strdup ("Quota is disabled, please enable quota");
+                *op_errstr = gf_strdup ("Quota is disabled, please enable "
+                                        "quota");
                 goto out;
         }
 
-        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
-                                    &quota_limits);
+        ret = dict_get_str (dict, "path", &path);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "failed to get the quota limits");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
                 goto out;
         }
 
-        ret = dict_get_str (dict, "path", &path);
-        if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
+        ret = gf_canonicalize_path (path);
+        if (ret)
                 goto out;
+
+        if (is_origin_glusterd ()) {
+                ret = glusterd_remove_quota_limit (volinfo->volname, path,
+                                                   op_errstr);
+                if (ret)
+                        goto out;
         }
 
-        ret = _glusterd_quota_remove_limits (&quota_limits, path, &flag);
-        if (ret == -1) {
-                if (flag == _gf_true)
-                        snprintf (str, sizeof (str), "Removing limit on %s has "
-                                  "been unsuccessful", path);
-                else
-                        snprintf (str, sizeof (str), "%s has no limit set", path);
-                *op_errstr = gf_strdup (str);
+        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;
-        } else {
-                if (flag == _gf_true)
-                        snprintf (str, sizeof (str), "Removed quota limit on "
-                                  "%s", path);
-                else
-                        snprintf (str, sizeof (str), "no limit set on %s",
-                                  path);
-                *op_errstr = gf_strdup (str);
-        }
-
-        if (quota_limits) {
-                ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                                    quota_limits);
-                if (ret) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
-                        goto out;
-                }
-        } else {
-               dict_del (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE);
         }
 
+        ret = glusterd_store_quota_config (volinfo, path, gfid_str, opcode,
+                                           op_errstr);
+        if (ret)
+                goto out;
+
+
         ret = 0;
 
 out:
         return ret;
 }
 
+int
+glusterd_set_quota_option (glusterd_volinfo_t *volinfo, dict_t *dict,
+                           char *key, char **op_errstr)
+{
+        int        ret    = 0;
+        char      *value  = NULL;
+        xlator_t  *this   = NULL;
+        char      *option = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        ret = glusterd_check_if_quota_trans_enabled (volinfo);
+        if (ret == -1) {
+                gf_asprintf (op_errstr, "Cannot set %s. Quota on volume %s is "
+                                        "disabled", key, volinfo->volname);
+                return -1;
+        }
+
+        ret = dict_get_str (dict, "value", &value);
+        if(ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Option value absent.");
+                return -1;
+        }
+
+        option = gf_strdup (value);
+        ret = dict_set_dynstr (volinfo->dict, key, option);
+        if(ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set option %s",
+                        key);
+                return -1;
+        }
+
+        return 0;
+}
+
+static int
+glusterd_quotad_op (int opcode)
+{
+        int ret = -1;
+
+        switch (opcode) {
+                case GF_QUOTA_OPTION_TYPE_ENABLE:
+                case GF_QUOTA_OPTION_TYPE_DISABLE:
+
+                        if (glusterd_all_volumes_with_quota_stopped ())
+                                ret = glusterd_quotad_stop ();
+                        else
+                                ret = glusterd_check_generate_start_quotad ();
+                        break;
+
+                case GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT:
+                case GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT:
+                case GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT:
+                case GF_QUOTA_OPTION_TYPE_ALERT_TIME:
+
+                        ret = glusterd_reconfigure_quotad ();
+                        break;
+
+                default:
+                        ret = 0;
+                        break;
+        }
+        return ret;
+}
 
 int
 glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
@@ -682,76 +1083,120 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
         int                     type         = -1;
         gf_boolean_t            start_crawl  = _gf_false;
         glusterd_conf_t        *priv         = NULL;
+        xlator_t               *this         = NULL;
 
         GF_ASSERT (dict);
         GF_ASSERT (op_errstr);
 
-        priv = THIS->private;
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         ret = dict_get_str (dict, "volname", &volname);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get volume name " );
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volume name");
                 goto out;
         }
 
         ret = glusterd_volinfo_find (volname, &volinfo);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
                 goto out;
         }
 
         ret = dict_get_int32 (dict, "type", &type);
 
-        if (type == GF_QUOTA_OPTION_TYPE_ENABLE) {
-                ret = glusterd_quota_enable (volinfo, op_errstr, &start_crawl);
-                if (ret < 0)
-                        goto out;
-
-                goto create_vol;
+        if ((priv->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                gf_asprintf (op_errstr, "Volume quota failed. The cluster is "
+                                        "operating at version %d. Quota command"
+                                        " %s is unavailable in this version.",
+                                        priv->op_version,
+                                        gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
         }
 
-        if (type == GF_QUOTA_OPTION_TYPE_DISABLE) {
-                ret = glusterd_quota_disable (volinfo, op_errstr);
-                if (ret < 0)
-                        goto out;
+        switch (type) {
+                case GF_QUOTA_OPTION_TYPE_ENABLE:
+                        ret = glusterd_quota_enable (volinfo, op_errstr,
+                                                     &start_crawl);
+                        if (ret < 0)
+                                goto out;
+                        break;
 
-                goto create_vol;
-        }
+                case GF_QUOTA_OPTION_TYPE_DISABLE:
+                        ret = glusterd_quota_disable (volinfo, op_errstr);
+                        if (ret < 0)
+                                goto out;
 
-        if (type == GF_QUOTA_OPTION_TYPE_LIMIT_USAGE) {
-                ret = glusterd_quota_limit_usage (volinfo, dict, op_errstr);
-                if (ret < 0)
+                        break;
+
+                case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
+                        ret = glusterd_quota_limit_usage (volinfo, dict, type,
+                                                          op_errstr);
                         goto out;
 
-                goto create_vol;
-        }
+                case GF_QUOTA_OPTION_TYPE_REMOVE:
+                        ret = glusterd_quota_remove_limits (volinfo, dict, type,
+                                                            op_errstr);
+                        goto out;
 
-        if (type == GF_QUOTA_OPTION_TYPE_REMOVE) {
-                ret = glusterd_quota_remove_limits (volinfo, dict, op_errstr);
-                if (ret < 0)
+                case GF_QUOTA_OPTION_TYPE_LIST:
+                        ret = glusterd_check_if_quota_trans_enabled (volinfo);
+                        if (ret == -1) {
+                                *op_errstr = gf_strdup ("Cannot list limits, "
+                                                        "quota is disabled");
+                                goto out;
+                        }
+                        ret = glusterd_quota_get_default_soft_limit (volinfo,
+                                                               rsp_dict);
                         goto out;
 
-                goto create_vol;
-        }
+                case GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT:
+                        ret = glusterd_set_quota_option (volinfo, dict,
+                                                        "features.soft-timeout",
+                                                         op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
 
-        if (type == GF_QUOTA_OPTION_TYPE_LIST) {
-                ret = glusterd_check_if_quota_trans_enabled (volinfo);
-                if (ret == -1) {
-                        *op_errstr = gf_strdup ("cannot list the limits, "
-                                                "quota is disabled");
-                        goto out;
-                }
+                case GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT:
+                        ret = glusterd_set_quota_option (volinfo, dict,
+                                                        "features.hard-timeout",
+                                                         op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
 
-                ret = glusterd_quota_get_limit_usages (priv, volinfo, volname,
-                                                       dict, op_errstr, rsp_dict);
+                case GF_QUOTA_OPTION_TYPE_ALERT_TIME:
+                        ret = glusterd_set_quota_option (volinfo, dict,
+                                                         "features.alert-time",
+                                                         op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
 
-                goto out;
+                case GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT:
+                        ret = glusterd_set_quota_option (volinfo, dict,
+                                                  "features.default-soft-limit",
+                                                  op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
+
+                default:
+                        gf_asprintf (op_errstr, "Quota command failed. Invalid "
+                                     "opcode");
+                        ret = -1;
+                        goto out;
         }
-create_vol:
+
         ret = glusterd_create_volfiles_and_notify_services (volinfo);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to re-create volfile for"
-                                          " 'quota'");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to re-create "
+                                                  "volfiles");
                 ret = -1;
                 goto out;
         }
@@ -760,81 +1205,262 @@ create_vol:
         if (ret)
                 goto out;
 
-        if (GLUSTERD_STATUS_STARTED == volinfo->status)
-                ret = glusterd_check_generate_start_nfs ();
-
-        ret = 0;
+        if (GLUSTERD_STATUS_STARTED == volinfo->status) {
+                if (priv->op_version == GD_OP_VERSION_MIN)
+                        ret = glusterd_check_generate_start_nfs ();
+        }
 
-out:
         if (rsp_dict && start_crawl == _gf_true)
                 glusterd_quota_initiate_fs_crawl (priv, volname);
 
-        if (rsp_dict && *op_errstr) {
-                ret = dict_set_dynstr (rsp_dict, "errstr", *op_errstr);
+        if (priv->op_version > GD_OP_VERSION_MIN) {
+                ret = glusterd_quotad_op (type);
+                if (ret)
+                        goto out;
+        }
+        ret = 0;
+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 (brickinfo->vg[0])
+                        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_FREE (*op_errstr);
-                        gf_log ("", GF_LOG_DEBUG,
-                                "failed to set error message in ctx");
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to place "
+                                "gfid of %s in dict", backend_path);
+                        GF_FREE (gfid_str);
+                        goto out;
                 }
-                *op_errstr = NULL;
+                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;
+        int                 ret           = 0;
+        int                 type          = 0;
+        int                 i             = 0;
+        char               *volname       = NULL;
+        char               *value         = NULL;
+        gf_boolean_t        exists        = _gf_false;
+        dict_t             *ctx           = NULL;
+        dict_t             *tmp_dict      = NULL;
+        xlator_t           *this          = NULL;
+        glusterd_conf_t    *priv          = NULL;
+        glusterd_volinfo_t *volinfo       = NULL;
+
+        struct {
+                int opcode;
+                char *key;
+        } optable[] = {
+                {GF_QUOTA_OPTION_TYPE_ALERT_TIME,
+                                                "features.alert-time"},
+                {GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT, "features.soft-timeout"},
+                {GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT, "features.hard-timeout"},
+                {GF_QUOTA_OPTION_TYPE_NONE, NULL}
+        };
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         GF_ASSERT (dict);
         GF_ASSERT (op_errstr);
 
+        tmp_dict = dict_new ();
+        if (!tmp_dict)
+                goto out;
+
         ret = dict_get_str (dict, "volname", &volname);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get volume name");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volume name");
                 goto out;
         }
 
         exists = glusterd_check_volume_exists (volname);
         if (!exists) {
-                gf_log ("", GF_LOG_ERROR, "Volume with name: %s "
-                                "does not exist",
-                                volname);
-                *op_errstr = gf_strdup ("Invalid volume name");
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
+                ret = -1;
+                goto out;
+        }
+        ret = glusterd_volinfo_find (volname, &volinfo);
+        if (ret) {
+                gf_asprintf (op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
+                goto out;
+        }
+
+        if (!glusterd_is_volume_started (volinfo)) {
+                *op_errstr = gf_strdup ("Volume is stopped, start volume "
+                                        "before executing quota command.");
                 ret = -1;
                 goto out;
         }
 
         ret = dict_get_int32 (dict, "type", &type);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to get 'type' for quota op");
-                *op_errstr = gf_strdup ("Volume quota failed, internal error "
-                                        ", unable to get type of operation");
+                *op_errstr = gf_strdup ("Volume quota failed, internal error, "
+                                        "unable to get type of operation");
                 goto out;
         }
 
+        if ((!glusterd_is_volume_quota_enabled (volinfo)) &&
+            (type != GF_QUOTA_OPTION_TYPE_ENABLE)) {
+                *op_errstr = gf_strdup ("Quota is disabled, please enable "
+                                        "quota");
+                ret = -1;
+                goto out;
+        }
 
-        ctx = glusterd_op_get_ctx();
-        if (ctx && (type == GF_QUOTA_OPTION_TYPE_ENABLE
-                    || type == GF_QUOTA_OPTION_TYPE_LIST)) {
-                /* Fuse mount req. only for enable & list-usage options*/
-                if (!glusterd_is_fuse_available ()) {
-                        gf_log ("glusterd", GF_LOG_ERROR, "Unable to open /dev/"
-                                "fuse (%s), quota command failed",
-                                strerror (errno));
-                        *op_errstr = gf_strdup ("Fuse unavailable");
-                        ret = -1;
-                        goto out;
-                }
+        if ((priv->op_version == GD_OP_VERSION_MIN) &&
+            (type > GF_QUOTA_OPTION_TYPE_VERSION)) {
+                gf_asprintf (op_errstr, "Volume quota failed. The cluster is "
+                                        "operating at version %d. Quota command"
+                                        " %s is unavailable in this version.",
+                                         priv->op_version,
+                                         gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
         }
 
-out:
-        gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
+         ctx = glusterd_op_get_ctx();
+         if (ctx && (type == GF_QUOTA_OPTION_TYPE_ENABLE
+                     || type == GF_QUOTA_OPTION_TYPE_LIST)) {
+                 /* Fuse mount req. only for enable & list-usage options*/
+                 if (!glusterd_is_fuse_available ()) {
+                         *op_errstr = gf_strdup ("Fuse unavailable");
+                         ret = -1;
+                         goto out;
+                 }
+         }
 
-        return ret;
+        switch (type) {
+                case GF_QUOTA_OPTION_TYPE_LIMIT_USAGE:
+                case GF_QUOTA_OPTION_TYPE_REMOVE:
+                        ret = glusterd_get_gfid_from_brick (dict, volinfo,
+                                                            rsp_dict,
+                                                            op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
+
+                case GF_QUOTA_OPTION_TYPE_ALERT_TIME:
+                case GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT:
+                case GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT:
+                        ret = dict_get_str (dict, "value", &value);
+                        if (ret)
+                                goto out;
+
+                        for (i = 0; optable[i].key; i++) {
+                                if (type == optable[i].opcode)
+                                        break;
+                        }
+                        ret = dict_set_str (tmp_dict, optable[i].key, value);
+                        if (ret)
+                                goto out;
+
+                        ret = glusterd_validate_reconfopts (volinfo, tmp_dict,
+                                                                    op_errstr);
+                        if (ret)
+                                goto out;
+                        break;
+
+                default:
+                        ret = 0;
+        }
+
+        ret = 0;
+
+ out:
+        if (tmp_dict)
+                dict_unref (tmp_dict);
+        if (ret && op_errstr && *op_errstr)
+                gf_log (this->name, GF_LOG_ERROR, "%s", *op_errstr);
+        gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);
+
+         return ret;
 }