quota: add version to quota xattrs

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>

1 files changed, 207 insertions, 30 deletions

diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index e579417810f..34e0fe73fec 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -23,7 +23,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,
@@ -36,6 +36,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)
 {
@@ -334,7 +408,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);
@@ -346,6 +420,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");
@@ -371,6 +457,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;
 }
@@ -379,13 +466,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)
@@ -423,7 +526,6 @@ out:
         return 0;
 }
 
-
 int32_t
 marker_setxattr_done (call_frame_t *frame)
 {
@@ -1200,7 +1302,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;
@@ -1237,7 +1339,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;
@@ -1301,7 +1404,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, };
 
@@ -1322,7 +1425,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;
@@ -1349,7 +1452,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;
@@ -1368,7 +1471,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;
@@ -2292,6 +2395,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;
 
@@ -2555,12 +2662,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;
 
@@ -2597,14 +2719,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)) {
@@ -2619,10 +2753,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);
 
@@ -2654,9 +2785,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;
 
@@ -2664,6 +2795,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;
 
@@ -2678,7 +2813,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);
@@ -2710,7 +2846,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);
@@ -2734,6 +2869,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)
@@ -2792,10 +2934,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:
@@ -2811,6 +2959,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;
 
@@ -2818,6 +2967,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),
@@ -2831,7 +2984,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,
@@ -2980,6 +3133,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);
@@ -3004,6 +3158,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);
@@ -3059,6 +3225,7 @@ init (xlator_t *this)
         priv = this->private;
 
         priv->feature_enabled = 0;
+        priv->version = 0;
 
         LOCK_INIT (&priv->lock);
 
@@ -3076,6 +3243,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);