quota: add version to quota xattrs

This is a backport of http://review.gluster.org/#/c/12386/

When a quota is disable and the clean-up process terminated
without completely cleaning-up the quota xattrs.
Now when quota is enabled again, this can mess-up the accounting

A version number is suffixed for all quota xattrs and this version
number is specific to marker xaltor, i.e when quota xattrs are
requested by quotad/client marker will remove the version suffix in the
key before sending the response

> Change-Id: I1ca2c11460645edba0f6b68db70d476d8d26e1eb
> BUG: 1272411
> Signed-off-by: vmallika <vmallika@redhat.com>
> Reviewed-on: http://review.gluster.org/12386
> Tested-by: NetBSD Build System <jenkins@build.gluster.org>
> Reviewed-by: Krishnan Parthasarathi <kparthas@redhat.com>
> Tested-by: Gluster Build System <jenkins@build.gluster.com>
> Reviewed-by: Manikandan Selvaganesh <mselvaga@redhat.com>
> Reviewed-by: Raghavendra G <rgowdapp@redhat.com>

Change-Id: I67b1b930b28411d76b2d476a4e5250c52aa495a0
BUG: 1277080
Signed-off-by: vmallika <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/12487
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Tested-by: Raghavendra G <rgowdapp@redhat.com>

5 files changed, 285 insertions, 69 deletions

diff --git a/xlators/features/marker/src/marker-quota-helper.c b/xlators/features/marker/src/marker-quota-helper.c
index df76f12cf47..4aba599e12e 100644
--- a/xlators/features/marker/src/marker-quota-helper.c
+++ b/xlators/features/marker/src/marker-quota-helper.c
@@ -260,19 +260,19 @@ mq_dict_set_contribution (xlator_t *this, dict_t *dict, loc_t *loc,
                           uuid_t gfid, char *contri_key)
 {
         int32_t ret                  = -1;
-        char    key[CONTRI_KEY_MAX]  = {0, };
+        char    key[QUOTA_KEY_MAX]   = {0, };
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", dict, out);
         GF_VALIDATE_OR_GOTO ("marker", loc, out);
 
         if (gfid && !gf_uuid_is_null(gfid)) {
-                GET_CONTRI_KEY (key, gfid, ret);
+                GET_CONTRI_KEY (this, key, gfid, ret);
         } else if (loc->parent) {
-                GET_CONTRI_KEY (key, loc->parent->gfid, ret);
+                GET_CONTRI_KEY (this, key, loc->parent->gfid, ret);
         } else {
                 /* nameless lookup, fetch contributions to all parents */
-                GET_CONTRI_KEY (key, NULL, ret);
+                GET_CONTRI_KEY (this, key, NULL, ret);
         }
 
         if (ret < 0)
@@ -283,7 +283,7 @@ mq_dict_set_contribution (xlator_t *this, dict_t *dict, loc_t *loc,
                 goto out;
 
         if (contri_key)
-                strncpy (contri_key, key, CONTRI_KEY_MAX);
+                strncpy (contri_key, key, QUOTA_KEY_MAX);
 
 out:
         if (ret < 0)
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index 645faa445c2..bb65c8787b2 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -211,10 +211,12 @@ _quota_dict_get_meta (xlator_t *this, dict_t *dict, char *key,
 }
 
 int32_t
-quota_dict_set_size_meta (dict_t *dict, const quota_meta_t *meta)
+quota_dict_set_size_meta (xlator_t *this, dict_t *dict,
+                          const quota_meta_t *meta)
 {
-        int32_t         ret      = -ENOMEM;
-        quota_meta_t   *value    = NULL;
+        int32_t         ret                        = -ENOMEM;
+        quota_meta_t   *value                      = NULL;
+        char            size_key[QUOTA_KEY_MAX]    = {0, };
 
         value = GF_CALLOC (2, sizeof (quota_meta_t), gf_common_quota_meta_t);
         if (value == NULL) {
@@ -228,7 +230,10 @@ quota_dict_set_size_meta (dict_t *dict, const quota_meta_t *meta)
         value[1].file_count = 0;
         value[1].dir_count = hton64 (1);
 
-        ret = dict_set_bin (dict, QUOTA_SIZE_KEY, value,
+        GET_SIZE_KEY (this, size_key, ret);
+        if (ret < 0)
+                goto out;
+        ret = dict_set_bin (dict, size_key, value,
                             (sizeof (quota_meta_t) * 2));
         if (ret < 0) {
                 gf_log_callingfn ("quota", GF_LOG_ERROR, "dict set failed");
@@ -285,7 +290,8 @@ mq_are_xattrs_set (xlator_t *this, loc_t *loc, gf_boolean_t *contri_set,
                    gf_boolean_t *size_set)
 {
         int32_t        ret                         = -1;
-        char           contri_key[CONTRI_KEY_MAX]  = {0, };
+        char           contri_key[QUOTA_KEY_MAX]   = {0, };
+        char           size_key[QUOTA_KEY_MAX]     = {0, };
         quota_meta_t   meta                        = {0, };
         struct iatt    stbuf                       = {0,};
         dict_t        *dict                        = NULL;
@@ -297,7 +303,7 @@ mq_are_xattrs_set (xlator_t *this, loc_t *loc, gf_boolean_t *contri_set,
                 goto out;
         }
 
-        ret = mq_req_xattr (this, loc, dict, contri_key);
+        ret = mq_req_xattr (this, loc, dict, contri_key, size_key);
         if (ret < 0)
                 goto out;
 
@@ -316,8 +322,7 @@ mq_are_xattrs_set (xlator_t *this, loc_t *loc, gf_boolean_t *contri_set,
         *contri_set = _gf_true;
         *size_set = _gf_true;
         if (loc->inode->ia_type == IA_IFDIR) {
-                ret = quota_dict_get_inode_meta (rsp_dict, QUOTA_SIZE_KEY,
-                                                 &meta);
+                ret = quota_dict_get_inode_meta (rsp_dict, size_key, &meta);
                 if (ret < 0 || meta.dir_count == 0)
                         *size_set = _gf_false;
         }
@@ -361,7 +366,7 @@ mq_create_size_xattrs (xlator_t *this, quota_inode_ctx_t *ctx, loc_t *loc)
                 goto out;
         }
 
-        ret = quota_dict_set_size_meta (dict, &size);
+        ret = quota_dict_set_size_meta (this, dict, &size);
         if (ret < 0)
                 goto out;
 
@@ -582,7 +587,8 @@ _mq_get_metadata (xlator_t *this, loc_t *loc, quota_meta_t *contri,
 {
         int32_t            ret                         = -1;
         quota_meta_t       meta                        = {0, };
-        char               contri_key[CONTRI_KEY_MAX]  = {0, };
+        char               contri_key[QUOTA_KEY_MAX]   = {0, };
+        char               size_key[QUOTA_KEY_MAX]     = {0, };
         dict_t            *dict                        = NULL;
         dict_t            *rsp_dict                    = NULL;
         struct iatt        stbuf                       = {0,};
@@ -600,7 +606,10 @@ _mq_get_metadata (xlator_t *this, loc_t *loc, quota_meta_t *contri,
         }
 
         if (size && loc->inode->ia_type == IA_IFDIR) {
-                ret = dict_set_int64 (dict, QUOTA_SIZE_KEY, 0);
+                GET_SIZE_KEY (this, size_key, ret);
+                if (ret < 0)
+                        goto out;
+                ret = dict_set_int64 (dict, size_key, 0);
                 if (ret < 0) {
                         gf_log (this->name, GF_LOG_ERROR, "dict_set failed.");
                         goto out;
@@ -625,7 +634,7 @@ _mq_get_metadata (xlator_t *this, loc_t *loc, quota_meta_t *contri,
 
         if (size) {
                 if (loc->inode->ia_type == IA_IFDIR) {
-                        ret = quota_dict_get_meta (rsp_dict, QUOTA_SIZE_KEY,
+                        ret = quota_dict_get_meta (rsp_dict, size_key,
                                                    &meta);
                         if (ret < 0) {
                                 gf_log (this->name, GF_LOG_ERROR,
@@ -754,9 +763,9 @@ mq_remove_contri (xlator_t *this, loc_t *loc, quota_inode_ctx_t *ctx,
                   inode_contribution_t *contri, quota_meta_t *delta)
 {
         int32_t              ret                         = -1;
-        char                 contri_key[CONTRI_KEY_MAX]  = {0, };
+        char                 contri_key[QUOTA_KEY_MAX]   = {0, };
 
-        GET_CONTRI_KEY (contri_key, contri->gfid, ret);
+        GET_CONTRI_KEY (this, contri_key, contri->gfid, ret);
         if (ret < 0) {
                 gf_log (this->name, GF_LOG_ERROR, "get contri_key "
                         "failed for %s", uuid_utoa(contri->gfid));
@@ -803,7 +812,7 @@ mq_update_contri (xlator_t *this, loc_t *loc, inode_contribution_t *contri,
                   quota_meta_t *delta)
 {
         int32_t              ret                         = -1;
-        char                 contri_key[CONTRI_KEY_MAX]  = {0, };
+        char                 contri_key[QUOTA_KEY_MAX]   = {0, };
         dict_t              *dict                        = NULL;
 
         GF_VALIDATE_OR_GOTO ("marker", loc, out);
@@ -823,7 +832,7 @@ mq_update_contri (xlator_t *this, loc_t *loc, inode_contribution_t *contri,
                 goto out;
         }
 
-        GET_CONTRI_KEY (contri_key, contri->gfid, ret);
+        GET_CONTRI_KEY (this, contri_key, contri->gfid, ret);
         if (ret < 0) {
                 gf_log (this->name, GF_LOG_ERROR, "get contri_key "
                         "failed for %s", uuid_utoa(contri->gfid));
@@ -889,7 +898,7 @@ mq_update_size (xlator_t *this, loc_t *loc, quota_meta_t *delta)
                 goto out;
         }
 
-        ret = quota_dict_set_size_meta (dict, delta);
+        ret = quota_dict_set_size_meta (this, dict, delta);
         if (ret < 0)
                 goto out;
 
@@ -1800,12 +1809,13 @@ mq_inspect_directory_xattr (xlator_t *this, quota_inode_ctx_t *ctx,
                             inode_contribution_t *contribution, loc_t *loc,
                             dict_t *dict, struct iatt buf)
 {
-        int32_t               ret                          = 0;
+        int32_t               ret                          = -1;
         int8_t                dirty                        = -1;
         quota_meta_t          size                         = {0, };
         quota_meta_t          contri                       = {0, };
         quota_meta_t          delta                        = {0, };
-        char                  contri_key[CONTRI_KEY_MAX]   = {0, };
+        char                  contri_key[QUOTA_KEY_MAX]    = {0, };
+        char                  size_key[QUOTA_KEY_MAX]      = {0, };
         gf_boolean_t          status                       = _gf_false;
 
         ret = dict_get_int8 (dict, QUOTA_DIRTY_KEY, &dirty);
@@ -1817,13 +1827,16 @@ mq_inspect_directory_xattr (xlator_t *this, quota_inode_ctx_t *ctx,
                 dirty = 0;
         }
 
-        ret = _quota_dict_get_meta (this, dict, QUOTA_SIZE_KEY, &size,
+        GET_SIZE_KEY (this, size_key, ret);
+        if (ret < 0)
+                goto out;
+        ret = _quota_dict_get_meta (this, dict, size_key, &size,
                                     IA_IFDIR, _gf_false);
         if (ret < 0)
                 goto create_xattr;
 
         if (!loc_is_root(loc)) {
-                GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
+                GET_CONTRI_KEY (this, contri_key, contribution->gfid, ret);
                 if (ret < 0)
                         goto out;
 
@@ -1888,7 +1901,7 @@ mq_inspect_file_xattr (xlator_t *this, quota_inode_ctx_t *ctx,
         quota_meta_t          size                         = {0, };
         quota_meta_t          contri                       = {0, };
         quota_meta_t          delta                        = {0, };
-        char                  contri_key[CONTRI_KEY_MAX]   = {0, };
+        char                  contri_key[QUOTA_KEY_MAX]    = {0, };
         gf_boolean_t          status                       = _gf_false;
 
         LOCK (&ctx->lock);
@@ -1903,7 +1916,7 @@ mq_inspect_file_xattr (xlator_t *this, quota_inode_ctx_t *ctx,
         }
         UNLOCK (&ctx->lock);
 
-        GET_CONTRI_KEY (contri_key, contribution->gfid, ret);
+        GET_CONTRI_KEY (this, contri_key, contribution->gfid, ret);
         if (ret < 0)
                 goto out;
 
@@ -1980,9 +1993,10 @@ out:
 
 int32_t
 mq_req_xattr (xlator_t *this, loc_t *loc, dict_t *dict,
-              char *contri_key)
+              char *contri_key, char *size_key)
 {
-        int32_t               ret       = -1;
+        int32_t               ret                = -1;
+        char                  key[QUOTA_KEY_MAX] = {0, };
 
         GF_VALIDATE_OR_GOTO ("marker", this, out);
         GF_VALIDATE_OR_GOTO ("marker", loc, out);
@@ -1995,7 +2009,13 @@ mq_req_xattr (xlator_t *this, loc_t *loc, dict_t *dict,
                         goto out;
         }
 
-        ret = dict_set_uint64 (dict, QUOTA_SIZE_KEY, 0);
+        GET_SIZE_KEY (this, key, ret);
+        if (ret < 0)
+                goto out;
+        if (size_key)
+                strncpy (size_key, key, QUOTA_KEY_MAX);
+
+        ret = dict_set_uint64 (dict, key, 0);
         if (ret < 0)
                 goto out;
 
diff --git a/xlators/features/marker/src/marker-quota.h b/xlators/features/marker/src/marker-quota.h
index 39e67e97f5c..d655f125435 100644
--- a/xlators/features/marker/src/marker-quota.h
+++ b/xlators/features/marker/src/marker-quota.h
@@ -24,7 +24,7 @@
 #define QUOTA_DIRTY_KEY "trusted.glusterfs.quota.dirty"
 
 #define CONTRIBUTION  "contri"
-#define CONTRI_KEY_MAX 512
+#define QUOTA_KEY_MAX 512
 #define READDIR_BUF 4096
 
 
@@ -61,22 +61,40 @@
                 ret = 0;                                \
         } while (0);
 
-#define GET_CONTRI_KEY(var, _gfid, _ret)                                  \
+#define GET_CONTRI_KEY(_this, var, _gfid, _ret)                           \
         do {                                                              \
+                marker_conf_t  *_priv = _this->private;                   \
                 if (_gfid != NULL) {                                      \
                         char _gfid_unparsed[40];                          \
                         gf_uuid_unparse (_gfid, _gfid_unparsed);          \
-                        _ret = snprintf (var, CONTRI_KEY_MAX,             \
+                        _ret = snprintf (var, QUOTA_KEY_MAX,              \
                                          QUOTA_XATTR_PREFIX               \
-                                         ".%s.%s." CONTRIBUTION, "quota", \
-                                         _gfid_unparsed);                 \
+                                         ".%s.%s." CONTRIBUTION ".%d",    \
+                                         "quota", _gfid_unparsed,         \
+                                         _priv->version);                 \
                 } else {                                                  \
-                        _ret = snprintf (var, CONTRI_KEY_MAX,             \
+                        _ret = snprintf (var, QUOTA_KEY_MAX,              \
                                          QUOTA_XATTR_PREFIX               \
-                                         ".%s.." CONTRIBUTION, "quota");  \
+                                         ".%s.." CONTRIBUTION ".%d",      \
+                                         "quota", _priv->version);        \
                 }                                                         \
         } while (0)
 
+#define GET_QUOTA_KEY(_this, var, key, _ret)                              \
+        do {                                                              \
+                marker_conf_t  *_priv = _this->private;                   \
+                if (_priv->version > 0)                                   \
+                        _ret = snprintf (var, QUOTA_KEY_MAX, "%s.%d",     \
+                                         key, _priv->version);            \
+                else                                                      \
+                        _ret = snprintf (var, QUOTA_KEY_MAX, "%s", key);  \
+        } while (0)
+
+#define GET_SIZE_KEY(_this, var, _ret)                                    \
+        {                                                                 \
+                GET_QUOTA_KEY (_this, var, QUOTA_SIZE_KEY, _ret);         \
+        }
+
 #define QUOTA_SAFE_INCREMENT(lock, var)         \
         do {                                    \
                 LOCK (lock);                    \
@@ -117,7 +135,7 @@ struct inode_contribution {
 typedef struct inode_contribution inode_contribution_t;
 
 int32_t
-mq_req_xattr (xlator_t *, loc_t *, dict_t *, char *);
+mq_req_xattr (xlator_t *, loc_t *, dict_t *, char *, char *);
 
 int32_t
 mq_xattr_state (xlator_t *, loc_t *, dict_t *, struct iatt);
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index c80237a3684..41c3e1dc74d 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -27,7 +27,7 @@
 
 #define _GF_UID_GID_CHANGED 1
 
-static char   *quota_external_xattrs[] = {
+static char *mq_ext_xattrs[] = {
         QUOTA_SIZE_KEY,
         QUOTA_LIMIT_KEY,
         QUOTA_LIMIT_OBJECTS_KEY,
@@ -40,6 +40,80 @@ fini (xlator_t *this);
 int32_t
 marker_start_setxattr (call_frame_t *, xlator_t *);
 
+/* When client/quotad request for quota xattrs,
+ * replace the key-name by adding the version number
+ * in end of the key-name.
+ * In the cbk, result value of xattrs for original
+ * key-name.
+ * Below function marker_key_replace_with_ver and
+ * marker_key_set_ver is used for setting/removing
+ * version for the key-name
+ */
+int
+marker_key_replace_with_ver (xlator_t *this, dict_t *dict)
+{
+        int                ret                     = -1;
+        int                i                       = 0;
+        marker_conf_t     *priv                    = NULL;
+        char               key[QUOTA_KEY_MAX]      = {0, };
+
+        priv = this->private;
+
+        if (dict == NULL || priv->version <= 0) {
+                ret = 0;
+                goto out;
+        }
+
+        for (i = 0; mq_ext_xattrs[i]; i++) {
+                if (dict_get (dict, mq_ext_xattrs[i])) {
+                        GET_QUOTA_KEY (this, key, mq_ext_xattrs[i], ret);
+                        if (ret < 0)
+                                goto out;
+
+                        ret = dict_set (dict, key,
+                                        dict_get (dict, mq_ext_xattrs[i]));
+                        if (ret < 0)
+                                goto out;
+
+                        dict_del (dict, mq_ext_xattrs[i]);
+                }
+        }
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+int
+marker_key_set_ver (xlator_t *this, dict_t *dict)
+{
+        int              ret                     = -1;
+        int              i                       = -1;
+        marker_conf_t   *priv                    = NULL;
+        char             key[QUOTA_KEY_MAX]      = {0, };
+
+        priv = this->private;
+
+        if (dict == NULL || priv->version <= 0) {
+                ret = 0;
+                goto out;
+        }
+
+        for (i = 0; mq_ext_xattrs[i]; i++) {
+                GET_QUOTA_KEY (this, key, mq_ext_xattrs[i], ret);
+                if (ret < 0)
+                        goto out;
+
+                if (dict_get (dict, key))
+                        dict_set (dict, mq_ext_xattrs[i], dict_get (dict, key));
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
 marker_local_t *
 marker_local_ref (marker_local_t *local)
 {
@@ -336,7 +410,7 @@ marker_filter_internal_xattrs (xlator_t *this, dict_t *xattrs)
 
         priv = this->private;
         if (priv->feature_enabled & GF_QUOTA)
-                ext = quota_external_xattrs;
+                ext = mq_ext_xattrs;
 
         dict_foreach_match (xattrs, _is_quota_internal_xattr, ext,
                             dict_remove_foreach_fn, NULL);
@@ -348,6 +422,18 @@ marker_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                      int32_t op_ret, int32_t op_errno, dict_t *dict,
                      dict_t *xdata)
 {
+        int32_t     ret  = -1;
+
+        if (op_ret < 0)
+                goto unwind;
+
+        ret = marker_key_set_ver (this, dict);
+        if (ret < 0) {
+                op_ret = -1;
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+
         if (cookie) {
                 gf_log (this->name, GF_LOG_DEBUG,
                         "Filtering the quota extended attributes");
@@ -373,6 +459,7 @@ marker_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 marker_filter_internal_xattrs (frame->this, dict);
         }
 
+unwind:
         MARKER_STACK_UNWIND (getxattr, frame, op_ret, op_errno, dict, xdata);
         return 0;
 }
@@ -381,13 +468,29 @@ int32_t
 marker_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                  const char *name, dict_t *xdata)
 {
-        gf_boolean_t    is_true  = _gf_false;
-        marker_conf_t   *priv    = NULL;
-        unsigned long   cookie   = 0;
-        marker_local_t  *local   = NULL;
+        gf_boolean_t     is_true  = _gf_false;
+        marker_conf_t   *priv                       = NULL;
+        unsigned long    cookie                     = 0;
+        marker_local_t  *local                      = NULL;
+        char             key[QUOTA_KEY_MAX]         = {0, };
+        int32_t          ret                        = -1;
+        int32_t          i                          = 0;
 
         priv = this->private;
 
+        if (name) {
+                for (i = 0; mq_ext_xattrs[i]; i++) {
+                        if (strcmp (name, mq_ext_xattrs[i]))
+                                continue;
+
+                        GET_QUOTA_KEY (this, key, mq_ext_xattrs[i], ret);
+                        if (ret < 0)
+                                goto out;
+                        name = key;
+                        break;
+                }
+        }
+
         frame->local = mem_get0 (this->local_pool);
         local = frame->local;
         if (local == NULL)
@@ -425,7 +528,6 @@ out:
         return 0;
 }
 
-
 int32_t
 marker_setxattr_done (call_frame_t *frame)
 {
@@ -1202,7 +1304,7 @@ marker_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         marker_local_t *oplocal                         = NULL;
         call_stub_t    *stub                            = NULL;
         int32_t         ret                             = 0;
-        char            contri_key[CONTRI_KEY_MAX]      = {0, };
+        char            contri_key[QUOTA_KEY_MAX]       = {0, };
         loc_t           newloc                          = {0, };
 
         local = (marker_local_t *) frame->local;
@@ -1239,7 +1341,8 @@ marker_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
                 local->stub = stub;
 
-                GET_CONTRI_KEY (contri_key, oplocal->loc.parent->gfid, ret);
+                GET_CONTRI_KEY (this, contri_key, oplocal->loc.parent->gfid,
+                                ret);
                 if (ret < 0) {
                         local->err = ENOMEM;
                         goto quota_err;
@@ -1303,7 +1406,7 @@ marker_do_rename (call_frame_t *frame, void *cookie, xlator_t *this,
 {
         marker_local_t       *local                      = NULL;
         marker_local_t       *oplocal                    = NULL;
-        char                  contri_key[CONTRI_KEY_MAX] = {0, };
+        char                  contri_key[QUOTA_KEY_MAX]  = {0, };
         int32_t               ret                        = 0;
         quota_meta_t          contribution               = {0, };
 
@@ -1324,7 +1427,7 @@ marker_do_rename (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto err;
         }
 
-        GET_CONTRI_KEY (contri_key, oplocal->loc.parent->gfid, ret);
+        GET_CONTRI_KEY (this, contri_key, oplocal->loc.parent->gfid, ret);
         if (ret < 0) {
                 local->err = errno ? errno : ENOMEM;
                 goto err;
@@ -1351,7 +1454,7 @@ marker_get_oldpath_contribution (call_frame_t *lk_frame, void *cookie,
         call_frame_t    *frame                      = NULL;
         marker_local_t  *local                      = NULL;
         marker_local_t  *oplocal                    = NULL;
-        char             contri_key[CONTRI_KEY_MAX] = {0, };
+        char             contri_key[QUOTA_KEY_MAX]  = {0, };
         int32_t          ret                        = 0;
 
         local = lk_frame->local;
@@ -1370,7 +1473,7 @@ marker_get_oldpath_contribution (call_frame_t *lk_frame, void *cookie,
                 local->lk_frame = NULL;
         }
 
-        GET_CONTRI_KEY (contri_key, oplocal->loc.parent->gfid, ret);
+        GET_CONTRI_KEY (this, contri_key, oplocal->loc.parent->gfid, ret);
         if (ret < 0) {
                 local->err = errno ? errno : ENOMEM;
                 goto err;
@@ -2294,6 +2397,10 @@ marker_setxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
                 return 0;
         }
 
+        ret = marker_key_replace_with_ver (this, dict);
+        if (ret < 0)
+                goto err;
+
         if (priv->feature_enabled == 0)
                 goto wind;
 
@@ -2557,12 +2664,27 @@ int32_t
 marker_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                     const char *name, dict_t *xdata)
 {
-        int32_t          ret   = 0;
-        marker_local_t  *local = NULL;
-        marker_conf_t   *priv  = NULL;
+        int32_t          ret                  = -1;
+        int32_t          i                    = 0;
+        marker_local_t  *local                = NULL;
+        marker_conf_t   *priv                 = NULL;
+        char             key[QUOTA_KEY_MAX]   = {0, };
 
         priv = this->private;
 
+        if (name) {
+                for (i = 0; mq_ext_xattrs[i]; i++) {
+                        if (strcmp (name, mq_ext_xattrs[i]))
+                                continue;
+
+                        GET_QUOTA_KEY (this, key, mq_ext_xattrs[i], ret);
+                        if (ret < 0)
+                                goto err;
+                        name = key;
+                        break;
+                }
+        }
+
         if (priv->feature_enabled == 0)
                 goto wind;
 
@@ -2599,14 +2721,26 @@ marker_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                    int32_t op_ret, int32_t op_errno, inode_t *inode,
                    struct iatt *buf, dict_t *dict, struct iatt *postparent)
 {
-        marker_conf_t  *priv    = NULL;
-        marker_local_t *local   = NULL;
-        dict_t         *xattrs  = NULL;
+        marker_conf_t  *priv                       = NULL;
+        marker_local_t *local                      = NULL;
+        dict_t         *xattrs                     = NULL;
+        int32_t         ret                        = -1;
+
         priv = this->private;
+        local = (marker_local_t *) frame->local;
+        frame->local = NULL;
 
         if (op_ret == -1) {
                 gf_log (this->name, GF_LOG_TRACE, "lookup failed with %s",
                         strerror (op_errno));
+                goto unwind;
+        }
+
+        ret = marker_key_set_ver (this, dict);
+        if (ret < 0) {
+                op_ret = -1;
+                op_errno = ENOMEM;
+                goto unwind;
         }
 
         if (dict && __has_quota_xattrs (dict)) {
@@ -2621,10 +2755,7 @@ marker_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 xattrs = dict_ref (dict);
         }
 
-        local = (marker_local_t *) frame->local;
-
-        frame->local = NULL;
-
+unwind:
         STACK_UNWIND_STRICT (lookup, frame, op_ret, op_errno, inode, buf,
                              xattrs, postparent);
 
@@ -2656,9 +2787,9 @@ int32_t
 marker_lookup (call_frame_t *frame, xlator_t *this,
                loc_t *loc, dict_t *xattr_req)
 {
-        int32_t         ret     = 0;
-        marker_local_t *local   = NULL;
-        marker_conf_t  *priv    = NULL;
+        int32_t         ret                     = 0;
+        marker_local_t *local                   = NULL;
+        marker_conf_t  *priv                    = NULL;
 
         priv = this->private;
 
@@ -2666,6 +2797,10 @@ marker_lookup (call_frame_t *frame, xlator_t *this,
         if (!xattr_req)
                 goto err;
 
+        ret = marker_key_replace_with_ver (this, xattr_req);
+        if (ret < 0)
+                goto err;
+
         if (priv->feature_enabled == 0)
                 goto wind;
 
@@ -2680,7 +2815,8 @@ marker_lookup (call_frame_t *frame, xlator_t *this,
                 goto err;
 
         if ((priv->feature_enabled & GF_QUOTA))
-                mq_req_xattr (this, loc, xattr_req, NULL);
+                mq_req_xattr (this, loc, xattr_req, NULL, NULL);
+
 wind:
         STACK_WIND (frame, marker_lookup_cbk, FIRST_CHILD(this),
                     FIRST_CHILD(this)->fops->lookup, loc, xattr_req);
@@ -2712,7 +2848,6 @@ marker_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto out;
         }
 
-
         list_for_each_entry (entry, &entries->list, list) {
                 if (entry->inode == entry->inode->table->root) {
                         inode_unref (parent);
@@ -2736,6 +2871,13 @@ marker_build_ancestry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 inode_unref (parent);
                 parent = inode_ref (entry->inode);
                 loc_wipe (&loc);
+
+                ret = marker_key_set_ver (this, entry->dict);
+                if (ret < 0) {
+                        op_ret = -1;
+                        op_errno = ENOMEM;
+                        break;
+                }
         }
 
         if (parent)
@@ -2794,10 +2936,16 @@ marker_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 }
 
                 mq_xattr_state (this, &loc, entry->dict, entry->d_stat);
-
                 loc_wipe (&loc);
                 GF_FREE (resolvedpath);
                 resolvedpath = NULL;
+
+                ret = marker_key_set_ver (this, entry->dict);
+                if (ret < 0) {
+                        op_ret = -1;
+                        op_errno = ENOMEM;
+                        goto unwind;
+                }
         }
 
 unwind:
@@ -2813,6 +2961,7 @@ marker_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
         marker_conf_t  *priv  = NULL;
         loc_t           loc   = {0, };
         marker_local_t *local = NULL;
+        int             ret   = -1;
 
         priv = this->private;
 
@@ -2820,6 +2969,10 @@ marker_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
         if (!dict)
                 goto unwind;
 
+        ret = marker_key_replace_with_ver (this, dict);
+        if (ret < 0)
+                goto unwind;
+
         if (dict_get (dict, GET_ANCESTRY_DENTRY_KEY)) {
                 STACK_WIND (frame, marker_build_ancestry_cbk,
                             FIRST_CHILD(this),
@@ -2833,7 +2986,7 @@ marker_readdirp (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
 
                         loc.parent = local->loc.inode = inode_ref (fd->inode);
 
-                        mq_req_xattr (this, &loc, dict, NULL);
+                        mq_req_xattr (this, &loc, dict, NULL, NULL);
                 }
 
                 STACK_WIND (frame, marker_readdirp_cbk,
@@ -2982,6 +3135,7 @@ reconfigure (xlator_t *this, dict_t *options)
         data_t         *data    = NULL;
         gf_boolean_t    flag    = _gf_false;
         marker_conf_t  *priv    = NULL;
+        int32_t         version = 0;
 
         GF_ASSERT (this);
         GF_ASSERT (this->private);
@@ -3006,6 +3160,18 @@ reconfigure (xlator_t *this, dict_t *options)
                         priv->feature_enabled |= GF_INODE_QUOTA;
         }
 
+        data = dict_get (options, "quota-version");
+        if (data)
+                ret = gf_string2int32 (data->data, &version);
+
+        if (priv->feature_enabled) {
+                if (version >= 0)
+                        priv->version = version;
+                else
+                        gf_log (this->name, GF_LOG_ERROR, "Invalid quota "
+                                "version %d", priv->version);
+        }
+
         data = dict_get (options, "xtime");
         if (data) {
                 ret = gf_string2boolean (data->data, &flag);
@@ -3061,6 +3227,7 @@ init (xlator_t *this)
         priv = this->private;
 
         priv->feature_enabled = 0;
+        priv->version = 0;
 
         LOCK_INIT (&priv->lock);
 
@@ -3078,6 +3245,16 @@ init (xlator_t *this)
                         priv->feature_enabled |= GF_INODE_QUOTA;
         }
 
+        data = dict_get (options, "quota-version");
+        if (data)
+                ret = gf_string2int32 (data->data, &priv->version);
+
+        if (priv->feature_enabled && priv->version < 0) {
+                gf_log (this->name, GF_LOG_ERROR, "Invalid quota version %d",
+                        priv->version);
+                goto err;
+        }
+
         data = dict_get (options, "xtime");
         if (data) {
                 ret = gf_string2boolean (data->data, &flag);
diff --git a/xlators/features/marker/src/marker.h b/xlators/features/marker/src/marker.h
index e6e19331d70..19f65474453 100644
--- a/xlators/features/marker/src/marker.h
+++ b/xlators/features/marker/src/marker.h
@@ -147,6 +147,7 @@ struct marker_conf{
         char        *marker_xattr;
         uint64_t     quota_lk_owner;
         gf_lock_t    lock;
+        int32_t      version;
 };
 typedef struct marker_conf marker_conf_t;