features/quota: Improvements to quota

Old implementation
* Client side implementation of quota
    - Not secure
    - Increased traffic in updating the ctx

New Implementation
* 2 stages of quota implementation is done: Soft and hard quota
    Upon reaching soft quota limit on the directory it logs/alerts in the quota
    daemon log (ie DEFAULT_LOG_DIR/quotad.log) and no more writes allowed after
    hard quota limit. After reaching the soft-limit the daemon alerts the
    user/admin repeatively for every 'alert-time', which is configurable.

* Quota is moved to server-side.

    There will be 2 quota xlators
    i. Quota Server
        It takes care of the enforcing the quota and maintains the context
        specific to the brick. Since this doesn't have the complete picture of
        the cluster, cluster wide usage is updated from the quota daemon. This
        updated context is saved and used for the enforcement.

        It updates its context by searching the QUOTA_UPDATE_KEY from the dict
        in the setxattr call, and is updated from nowhere else.

        The quota is always loaded in the server graph and is by passed if the
        feature is not enabled.

        Options specific to quota-server:
        server-quota    - Specifies whether the features is on/off. It is used
                          to by pass the quota if turned off.
        deem-statfs     - If set to on, it takes quota limits into
                          consideration while estimating fs size. (df command)

    ii. Quota Daemon
        This is the new xlator introduced with this patch. Its the
        *gluster client* process with no mount point, started upon enabling
        quota or restarting the volume. This is a single process for all the
        volumes in the cluster. Its volfile stored in
        GLUSTERD_DEFAULT_WORKI_DIR/quotad/quotad.vol.

        It queries for the sizes on all the bricks, aggregates the size and
        sends back the updated size, periodically. The timeout between
        successive updation is configurable and typically/by default more for
        below-soft-quota usage and less for above-soft-quota usage. It
        maintains the timeout inside the limit structure based on the usage;
        below soft limit and above soft limit.

        There will be thread running per volume which iterates through the list
        and decides whether the size to be queried in the current iteration
        based on its timeout. It takes the next iteration time taking the least
        of the timeouts in the list of entries.

        Maintains a separate inode table for each volume in the quotad. In the
        first iteration it builds the table for quota-dirs (dirs on which limit
        is set) and its components.

        Options specific to quotad:
        hard-timeout       - Timeout for updation of usage to the quota-server
                             when the usage is crosses the soft-limit.
        soft-timeout       - Timeout for the updation of usage to the
                             quota-server when the usage is below soft-limit.
        alert-time         - Frequency of logging after the usage reached
                             soft limit.
   Options common to both:
   default-soft-limit - This is used when individual paths are not
                        configured with soft-limit and default value of
                        this option is 90% of the hard-limit.
   limit-set          - String containing all the limits.

   Thus in the current implementation we'll have 2 quota xlators: one in server
   graph and one in trusted client (quota daemon) of which the sole
   purpose will be to aggregate the quota size xattrs from all the bricks and
   send the same to server quota xlator.

* Changes in glusterd and CLI

   A single volfile is created for all the volumes, similar to nfs volfile.
   All files related to quota client (volfile, pid etc) are stored in
   GLUSTERD_DEFAULT_WORK_DIR/quotad/.

   The new pattern of the quota limit stores in
   limit-set = <single-dir-limit>[,<single-dir-limit>]
   single-dir-limit = <abs-path>:<hard-limit>[:<soft-limit-in-percent>]

   It also introduces new options:
   volume quota <VOLNAME> {enable|disable|list [<path> ...]|remove <path>| default-soft-limit <percent>} |
   volume quota <VOLNAME> {limit-usage <path> <size> |soft-limit <path> <percent>} |
   volume quota <VOLNAME> {alert-time|soft-timeout|hard-timeout} {<time>}

Credit:
Raghavendra Bhat        <rabhat@redhat.com>
Varun Shastry           <vshastry@redhat.com>
Shishir Gowda           <sgowda@redhat.com>
Kruthika Dhananjay      <kdhananj@redhat.com>
Brian Foster            <bfoster@redhat.com>
Krishnan Parthasarathi  <kparthas@redhat.com>

Change-Id: I16ec5be0c2faaf42b14034b9ccaf17796adef082
BUG: 969461
Signed-off-by: Varun Shastry <vshastry@redhat.com>

1 files changed, 535 insertions, 185 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-quota.c b/xlators/mgmt/glusterd/src/glusterd-quota.c
index 31826719..ec4a6eb8 100644
--- a/xlators/mgmt/glusterd/src/glusterd-quota.c
+++ b/xlators/mgmt/glusterd/src/glusterd-quota.c
@@ -24,6 +24,22 @@
 
 #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_SOFT_LIMIT]         = "soft-limit",
+        [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 +47,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 +100,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 1. Executing command "
+                         "%s is disallowed in this state",
+                         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 +143,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 +151,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 +176,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 +211,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 +228,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,22 +394,30 @@ _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,
+                                 glusterd_volinfo_t *volinfo, char *volname,
+                                 dict_t *dict, char **op_errstr,
                                  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            i               = 0;
+        int32_t            ret             = 0;
+        int32_t            count           = 0;
+        int                entry_count     = 0;
+        char              *path            = NULL;
+        char               cmd_str [1024]  = {0, };
+        char              *ret_str         = NULL;
+        xlator_t          *this            = NULL;
+        glusterd_conf_t   *conf            = NULL;
+        char              *default_limit   = NULL;
+        char              *val             = NULL;
 
         if (rsp_dict == NULL)
                 return 0;
 
+        this = THIS;
+        GF_ASSERT (this);
+        conf = this->private;
+        GF_ASSERT (conf);
+
         ret = dict_get_int32 (dict, "count", &count);
         if (ret < 0)
                 goto out;
@@ -404,22 +425,60 @@ glusterd_quota_get_limit_usages (glusterd_conf_t *priv,
         if (count == 0) {
                 ret_str = _glusterd_quota_get_limit_usages (volinfo, NULL,
                                                             op_errstr);
+                ret = dict_set_dynstr (rsp_dict, "limit_list", ret_str);
+                if (ret)
+                        goto out;
         } else {
                 i = 0;
                 while (count--) {
-                        snprintf (cmd_str, 1024, "path%d", i++);
+                        snprintf (cmd_str, sizeof (cmd_str), "path%d", i++);
 
                         ret = dict_get_str (dict, cmd_str, &path);
                         if (ret < 0)
                                 goto out;
+                        ret = gf_canonicalize_path (path);
+                        if (ret) {
+                                goto out;
+                        }
 
-                        ret_str = _glusterd_quota_get_limit_usages (volinfo, path, op_errstr);
+                        ret_str = _glusterd_quota_get_limit_usages (volinfo,
+                                                                    path,
+                                                                    op_errstr);
+                        /* Despite quota limits being absent on @path, we go
+                         * ahead and place it in @rsp_dict with
+                         * value = "Not set". This is because after commit op,
+                         * as part of aggregation of @rsp_dict with @op_ctx,
+                         * when we copy the rsp_dict into op_ctx, op_ctx would
+                         * still be containing the old key (same as @cmd_str)
+                         * with the old value. In order to overwrite the old
+                         * value, we replace it with "Not set", something that
+                         * the cli can easily interpret as a case of quota
+                         * limits not having been set on the given path.
+                         */
+                        if (!ret_str) {
+                                ret = dict_set_str (rsp_dict, cmd_str,
+                                                    "Not set");
+                        } else {
+                                ret = dict_set_dynstr (rsp_dict, cmd_str,
+                                                       ret_str);
+                                entry_count = entry_count + 1;
+                        }
                 }
         }
+        ret = dict_set_int32 (rsp_dict, "entry-count", entry_count);
+        if (ret)
+                goto out;
 
-        if (ret_str) {
-                ret = dict_set_dynstr (rsp_dict, "limit_list", ret_str);
-        }
+        ret = dict_set_uint32 (rsp_dict, "op-version", conf->op_version);
+
+        ret = glusterd_volinfo_get (volinfo, "features.default-soft-limit",
+                                    &default_limit);
+        if (default_limit)
+                val = gf_strdup (default_limit);
+        else
+                val = gf_strdup ("90%");
+
+        ret = dict_set_dynstr (rsp_dict, "default-soft-limit", val);
 out:
         return ret;
 }
@@ -430,10 +489,14 @@ 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 "
@@ -449,15 +512,15 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 
         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;
         }
 
@@ -467,17 +530,34 @@ glusterd_quota_enable (glusterd_volinfo_t *volinfo, char **op_errstr,
 
         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[]     = {VKEY_FEATURES_LIMIT_USAGE,
+                                     "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) {
@@ -487,47 +567,84 @@ 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;
         }
 
+        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]);
+                }
+                if ((i == 0) && (conf->op_version == GD_OP_VERSION_MIN))
+                        break;
+         }
+
         *op_errstr = gf_strdup ("Disabling quota has been successful");
+        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;
+}
 
-        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
-                                    &quota_limits);
-        if (ret) {
-                gf_log ("", GF_LOG_WARNING, "failed to get the quota limits");
+static void
+gd_quota_get_formatted_limit_list (char **value, char *existing_list,
+                                   char *path, char *hard_limit,
+                                   char *soft_limit)
+{
+        if (!existing_list) {
+                if (!soft_limit)
+                        gf_asprintf (value, "%s:%s", path, hard_limit);
+                else
+                        gf_asprintf (value, "%s:%s:%s", path, hard_limit,
+                                     soft_limit);
         } else {
-                GF_FREE (quota_limits);
+                if (!soft_limit)
+                        gf_asprintf (value, "%s,%s:%s", existing_list, path,
+                                     hard_limit);
+                else
+                        gf_asprintf (value, "%s,%s:%s:%s", existing_list, path,
+                                     hard_limit, soft_limit);
         }
-
-        dict_del (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE);
-
-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,
+                            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            *limit              = NULL;
+        char            *value              = NULL;
+        char            *sl                 = NULL;
+        char             msg[5120]          = {0,};
+        char            *quota_limits       = NULL;
+        char            *new_list           = NULL;
+        xlator_t        *this               = NULL;
+        glusterd_conf_t *priv               = NULL;
+        char            *removed_path       = 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);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        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) {
@@ -539,109 +656,222 @@ glusterd_quota_limit_usage (glusterd_volinfo_t *volinfo, dict_t *dict, char **op
         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");
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
                 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);
         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 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");
-                        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;
-                }
+                ; //do nothing and go past the else block
         } else {
-                ret = gf_asprintf (&value, "%s,%s:%s",
-                                   quota_limits, path, limit);
+                ret = _glusterd_quota_remove_limits (quota_limits, path, NULL,
+                                                     &new_list, &removed_path);
                 if (ret == -1) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
-                        *op_errstr = gf_strdup ("failed to set limit");
+                        gf_log (this->name, GF_LOG_ERROR, "Unable to remove "
+                                                          "limit");
                         goto out;
                 }
 
-                GF_FREE (quota_limits);
+                if (removed_path && (priv->op_version > GD_OP_VERSION_MIN)) {
+                        ret = gf_get_soft_limit (removed_path, &sl);
+                        if (ret == -1)
+                                goto out;
+                }
         }
-
-        quota_limits = value;
+        gd_quota_get_formatted_limit_list (&value, new_list, path, limit, sl);
 
         ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                            quota_limits);
+                            value);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
-                *op_errstr = gf_strdup ("failed to set limit");
+                gf_log (this->name, GF_LOG_ERROR, "Unable to set quota limits");
                 goto out;
         }
-        snprintf (msg, 1024, "limit set on %s", path);
-        *op_errstr = gf_strdup (msg);
 
+        snprintf (msg, sizeof (msg), "hard limit set on %s", path);
+        *op_errstr = gf_strdup (msg);
         ret = 0;
 out:
+        GF_FREE (sl);
+        GF_FREE (removed_path);
+        GF_FREE (new_list);
+
+        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;
+}
+
+int
+glusterd_quota_soft_limit (glusterd_volinfo_t *volinfo, dict_t *dict,
+                           char **op_errstr)
+{
+        int     ret                = 0;
+        char   *path               = NULL;
+        char   *limit              = NULL;
+        char   *quota_limits       = NULL;
+        char   *new_list           = NULL;
+        char   *removed_path       = NULL;
+        char    msg[1024]          = {0,};
+        char   *hl                 = NULL;
+        char   *value              = NULL;
+        xlator_t *this             = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        if (!volinfo || !dict || !op_errstr) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = glusterd_check_if_quota_trans_enabled (volinfo);
+        if (ret == -1) {
+                *op_errstr = gf_strdup ("Quota is disabled, please enable "
+                                        "quota");
+                goto out;
+        }
+
+        ret = glusterd_volinfo_get (volinfo, VKEY_FEATURES_LIMIT_USAGE,
+                                    &quota_limits);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "path", &path);
+        if (ret) {
+                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);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch limit");
+                goto out;
+        }
+
+        if (quota_limits == NULL) {
+                gf_asprintf (op_errstr, "Soft-limit cannot be set on path %s, "
+                             "without setting hard-limit on it first.", path);
+                ret = -1;
+                goto out;
+        } else {
+                ret = _glusterd_quota_remove_limits (quota_limits, path, NULL,
+                                                     &new_list, &removed_path);
+                 if (ret == -1) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to remove "
+                                                          "limit");
+                         goto out;
+                 }
+
+                if (!removed_path) {
+                        gf_asprintf (op_errstr, "Soft-limit cannot be set on "
+                                     "path %s without setting hard-limit on it "
+                                     "first.", path);
+                        ret = -1;
+                        goto out;
+                } else {
+                        ret = gf_get_hard_limit (removed_path, &hl);
+                        if (ret)
+                                goto out;
+                }
+         }
+
+         gd_quota_get_formatted_limit_list (&value, new_list, path, hl, limit);
+
+         ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
+                             value);
+         if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to set quota limits");
+                 goto out;
+         }
+
+         snprintf (msg, sizeof (msg), "soft limit set on %s", path);
+         *op_errstr = gf_strdup (msg);
+         ret = 0;
+
+out:
+        GF_FREE (hl);
+        GF_FREE (removed_path);
+        GF_FREE (new_list);
+
+        if (ret && op_errstr && !*op_errstr)
+                gf_asprintf (op_errstr, "Failed to set soft limit for path %s "
+                                        "on volume %s", path, volinfo->volname);
         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,
+                              char **op_errstr)
 {
         int32_t         ret                   = -1;
         char            str [PATH_MAX + 1024] = {0,};
         char            *quota_limits         = NULL;
+        char            *new_list             = NULL;
+        char            *value                = NULL;
         char            *path                 = NULL;
         gf_boolean_t     flag                 = _gf_false;
+        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);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "failed to get the quota limits");
+                gf_log (this->name, GF_LOG_ERROR, "failed to get quota limits");
                 goto out;
         }
 
         ret = dict_get_str (dict, "path", &path);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to fetch quota limits" );
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch path");
                 goto out;
         }
 
-        ret = _glusterd_quota_remove_limits (&quota_limits, path, &flag);
+        ret = gf_canonicalize_path (path);
+        if (ret)
+                goto out;
+
+        ret = _glusterd_quota_remove_limits (quota_limits, path, &flag,
+                                             &new_list, NULL);
         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);
+                        snprintf (str, sizeof (str), "%s has no limit set",
+                                  path);
                 *op_errstr = gf_strdup (str);
                 goto out;
         } else {
@@ -654,11 +884,13 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **
                 *op_errstr = gf_strdup (str);
         }
 
-        if (quota_limits) {
+        if (new_list) {
+                value = gf_strdup (new_list);
                 ret = dict_set_str (volinfo->dict, VKEY_FEATURES_LIMIT_USAGE,
-                                    quota_limits);
+                                    value);
                 if (ret) {
-                        gf_log ("", GF_LOG_ERROR, "Unable to set quota limits" );
+                        gf_log (this->name, GF_LOG_ERROR, "Unable to set quota "
+                                                          "limits");
                         goto out;
                 }
         } else {
@@ -668,9 +900,46 @@ glusterd_quota_remove_limits (glusterd_volinfo_t *volinfo, dict_t *dict, char **
         ret = 0;
 
 out:
+        GF_FREE (new_list);
         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;
+        }
+        gf_asprintf (op_errstr, "%s on volume %s set", key, volinfo->volname);
+
+        return 0;
+}
 
 int
 glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
@@ -681,26 +950,41 @@ 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 ((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. Option %s "
+                                        "is disallowed in this state.",
+                                        priv->op_version,
+                                        gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
+        }
+
         if (type == GF_QUOTA_OPTION_TYPE_ENABLE) {
                 ret = glusterd_quota_enable (volinfo, op_errstr, &start_crawl);
                 if (ret < 0)
@@ -708,7 +992,6 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
 
                 goto create_vol;
         }
-
         if (type == GF_QUOTA_OPTION_TYPE_DISABLE) {
                 ret = glusterd_quota_disable (volinfo, op_errstr);
                 if (ret < 0)
@@ -742,15 +1025,59 @@ glusterd_op_quota (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                 }
 
                 ret = glusterd_quota_get_limit_usages (priv, volinfo, volname,
-                                                       dict, op_errstr, rsp_dict);
+                                                       dict, op_errstr,
+                                                       rsp_dict);
 
                 goto out;
         }
+
+        if (type == GF_QUOTA_OPTION_TYPE_SOFT_LIMIT) {
+                ret = glusterd_quota_soft_limit (volinfo, dict, op_errstr);
+	                if (ret < 0)
+                                goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_SOFT_TIMEOUT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.soft-timeout",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_HARD_TIMEOUT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.hard-timeout",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
+        if (type == GF_QUOTA_OPTION_TYPE_ALERT_TIME) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.alert-time",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+        if (type == GF_QUOTA_OPTION_TYPE_DEFAULT_SOFT_LIMIT) {
+                ret = glusterd_set_quota_option (volinfo, dict,
+                                                 "features.default-soft-limit",
+                                                 op_errstr);
+                if (ret)
+                        goto out;
+                goto create_vol;
+        }
+
 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;
         }
@@ -759,20 +1086,31 @@ create_vol:
         if (ret)
                 goto out;
 
-        if (GLUSTERD_STATUS_STARTED == volinfo->status)
-                ret = glusterd_check_generate_start_nfs ();
-
+        if (GLUSTERD_STATUS_STARTED == volinfo->status) {
+                if (priv->op_version == GD_OP_VERSION_MIN)
+                        ret = glusterd_check_generate_start_nfs ();
+        }
         ret = 0;
 
 out:
         if (rsp_dict && start_crawl == _gf_true)
                 glusterd_quota_initiate_fs_crawl (priv, volname);
 
+        if (priv->op_version > GD_OP_VERSION_MIN &&
+            is_origin_glusterd ()) {
+                if (type != GF_QUOTA_OPTION_TYPE_LIST) {
+                        if (glusterd_all_volumes_with_quota_stopped ())
+                                ret = glusterd_quotad_stop ();
+                        else
+                                ret = glusterd_check_generate_start_quotad ();
+                }
+        }
+
         if (rsp_dict && *op_errstr) {
                 ret = dict_set_dynstr (rsp_dict, "errstr", *op_errstr);
                 if (ret) {
                         GF_FREE (*op_errstr);
-                        gf_log ("", GF_LOG_DEBUG,
+                        gf_log (this->name, GF_LOG_DEBUG,
                                 "failed to set error message in ctx");
                 }
                 *op_errstr = NULL;
@@ -789,51 +1127,63 @@ glusterd_op_stage_quota (dict_t *dict, char **op_errstr)
         gf_boolean_t       exists        = _gf_false;
         int                type          = 0;
         dict_t            *ctx           = NULL;
+        xlator_t          *this          = NULL;
+        glusterd_conf_t   *priv          = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
 
         GF_ASSERT (dict);
         GF_ASSERT (op_errstr);
 
         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 = 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;
         }
 
-
-        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. Option %s "
+                                        "is disallowed in this state.",
+                                         priv->op_version,
+                                         gd_quota_op_list[type]);
+                ret = -1;
+                goto out;
         }
 
-out:
-        gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
-
-        return 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;
+                 }
+         }
+
+ out:
+        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;
 }