features/marker: Filter quota xattrs on file as well

Problem:
Quota contributions of a file/directory are tracked by quota
xlator using xattrs on the file. Quota allows these xattrs to be
healed as part of metadata self-heal. This leads to
wrong quota calculations on this brick after self-heal because
quota xattrs don't represent the actual contributions on the
brick anymore.

Fix:
Don't let self-heal of this xattr happen as part of self-heal
by filtering quota xattrs on file in listxattr.

Change-Id: Iea68a116595ba271e58c6fdcc3dd21c7bb55ebb3
BUG: 1035576
Signed-off-by: Pranith Kumar K <pkarampu@redhat.com>
Reviewed-on: http://review.gluster.org/6374
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

4 files changed, 128 insertions, 3 deletions

diff --git a/tests/bugs/bug-1035576.t b/tests/bugs/bug-1035576.t
new file mode 100644
index 00000000000..08c8a5ea249
--- /dev/null
+++ b/tests/bugs/bug-1035576.t
@@ -0,0 +1,51 @@
+#!/bin/bash
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+
+#This script tests that self-heal of limit-set xattr is happening on a directory
+#but self-heal of quota.size xattr is not happening
+
+cleanup;
+
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{0,1}
+TEST $CLI volume start $V0
+#Lets disable perf-xls so that lookup would reach afr
+TEST $CLI volume set $V0 performance.quick-read off
+TEST $CLI volume set $V0 performance.io-cache off
+TEST $CLI volume set $V0 performance.write-behind off
+TEST $CLI volume set $V0 performance.stat-prefetch off
+TEST $CLI volume set $V0 performance.read-ahead off
+TEST $CLI volume set $V0 background-self-heal-count 0
+TEST $CLI volume set $V0 self-heal-daemon off
+TEST $CLI volume quota $V0 enable
+sleep 5 # wait for brick to connect to quotad
+TEST kill_brick $V0 $H0 $B0/${V0}0
+TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 $M0 --attribute-timeout=0 --entry-timeout=0
+cd $M0
+TEST mkdir $M0/a
+TEST $CLI volume quota $V0 limit-usage /a 1GB
+echo abc > $M0/a/f
+$CLI volume start $V0 force
+EXPECT_WITHIN 20 "1" afr_child_up_status $V0 0
+quota_limit_val1=$(get_hex_xattr trusted.glusterfs.quota.limit-set $B0/${V0}1/a)
+quota_size_val1=$(get_hex_xattr trusted.glusterfs.quota.size $B0/${V0}1/a)
+
+#Trigger entry,metadata self-heal
+TEST stat $M0/a
+quota_limit_val0=$(get_hex_xattr trusted.glusterfs.quota.limit-set $B0/${V0}0/a)
+quota_size_val0=$(get_hex_xattr trusted.glusterfs.quota.size $B0/${V0}0/a)
+
+#Test that limit-set xattr is healed
+TEST [ $quota_limit_val0 == $quota_limit_val1 ]
+
+#Only entry, metadata self-heal is done quota size value should not be same
+TEST [ $quota_size_val0 != $quota_size_val1 ]
+TEST stat $M0/a/f
+
+#Now that data self-heal is done quota size value should be same
+quota_size_val0=$(get_hex_xattr trusted.glusterfs.quota.size $B0/${V0}0/a)
+TEST [ $quota_size_val0 == $quota_size_val1 ]
+cleanup
diff --git a/tests/volume.rc b/tests/volume.rc
index 98712242020..b1aa904fb27 100644
--- a/tests/volume.rc
+++ b/tests/volume.rc
@@ -272,3 +272,10 @@ function get_backend_paths {
 
        getfattr -m . -n trusted.glusterfs.pathinfo $path | tr ' ' '\n' | sed -n 's/<POSIX.*:.*:\(.*\)>.*/\1/p'
 }
+
+#Gets the xattr value in hex, also removed 0x in front of the value
+function get_hex_xattr {
+        local key=$1
+        local path=$2
+        getfattr -d -m. -e hex $2 2>/dev/null | grep $1 | cut -f2 -d'=' | cut -f2 -d'x'
+}
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c
index afc568fbd9e..a758e938f69 100644
--- a/xlators/features/marker/src/marker-quota.c
+++ b/xlators/features/marker/src/marker-quota.c
@@ -1687,7 +1687,7 @@ unlock:
         }
         UNLOCK (&contribution->lock);
 
-        gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 "%"PRId64,
+        gf_log (this->name, GF_LOG_DEBUG, "%s %"PRId64 " %"PRId64,
                 local->loc.path, size_int, contri_int);
 
         local->delta = size_int - contri_int;
diff --git a/xlators/features/marker/src/marker.c b/xlators/features/marker/src/marker.c
index dbe9d530fad..280d43ae38b 100644
--- a/xlators/features/marker/src/marker.c
+++ b/xlators/features/marker/src/marker.c
@@ -281,15 +281,65 @@ 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)
 {
+        int            ret    = 0;
+        char           *src   = NULL;
+        char           *dst   = NULL;
+        int            len    = 0;
+        marker_local_t *local = NULL;
+
+        local = frame->local;
+
+
         if (cookie) {
                 gf_log (this->name, GF_LOG_DEBUG,
                         "Filtering the quota extended attributes");
 
-                dict_foreach_fnmatch (dict, "trusted.glusterfs.quota*",
-                                      marker_filter_quota_xattr, NULL);
+                /* If the getxattr is from a non special client, then do not
+                   copy the quota related xattrs (except the quota limit key
+                   i.e trusted.glusterfs.quota.limit-set which has been set by
+                   glusterd on the directory on which quota limit is set.) for
+                   directories. Let the healing of xattrs happen upon lookup.
+                   NOTE: setting of trusted.glusterfs.quota.limit-set as of now
+                   happens from glusterd. It should be moved to quotad. Also
+                   trusted.glusterfs.quota.limit-set is set on directory which
+                   is permanent till quota is removed on that directory or limit
+                   is changed. So let that xattr be healed by other xlators
+                   properly whenever directory healing is done.
+                */
+                ret = dict_get_ptr_and_len (dict, QUOTA_LIMIT_KEY,
+                                            (void **)&src, &len);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_DEBUG, "dict_get on %s "
+                                "failed", QUOTA_LIMIT_KEY);
+                } else {
+                        dst = GF_CALLOC (len, sizeof (char), gf_common_mt_char);
+                        if (dst)
+                                memcpy (dst, src, len);
+                }
+
+                /*
+                 * Except limit-set xattr, rest of the xattrs are maintained
+                 * by quota xlator. Don't expose them to other xlators.
+                 * This filter makes sure quota xattrs are not healed as part of
+                 * metadata self-heal
+                 */
+                GF_REMOVE_INTERNAL_XATTR ("trusted.glusterfs.quota*", dict);
+                if (!ret && IA_ISDIR (local->loc.inode->ia_type) && dst) {
+                        ret = dict_set_dynptr (dict, QUOTA_LIMIT_KEY,
+                                               dst, len);
+                        if (ret)
+                                gf_log (this->name, GF_LOG_WARNING, "setting "
+                                        "key %s failed", QUOTA_LIMIT_KEY);
+                        else
+                                dst = NULL;
+                }
         }
 
+        GF_FREE (dst);
+
+        frame->local = NULL;
         STACK_UNWIND_STRICT (getxattr, frame, op_ret, op_errno, dict, xdata);
+        marker_local_unref (local);
         return 0;
 }
 
@@ -300,9 +350,21 @@ marker_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
         gf_boolean_t   ret    = _gf_false;
         marker_conf_t *priv   = NULL;
         unsigned long  cookie = 0;
+        marker_local_t *local = NULL;
 
         priv = this->private;
 
+        frame->local = mem_get0 (this->local_pool);
+        local = frame->local;
+        if (local == NULL)
+                goto out;
+
+        MARKER_INIT_LOCAL (frame, local);
+
+        ret = loc_copy (&local->loc, loc);
+        if (ret < 0)
+                goto out;
+
         gf_log (this->name, GF_LOG_DEBUG, "USER:PID = %d", frame->root->pid);
 
         if (priv && priv->feature_enabled & GF_XTIME)
@@ -324,6 +386,11 @@ marker_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
         }
 
         return 0;
+out:
+        frame->local = NULL;
+        STACK_UNWIND_STRICT (getxattr, frame, -1, ENOMEM, NULL, NULL);
+        marker_local_unref (local);
+        return 0;
 }