cluster/dht: Create linkfile with file uid/gid

Currently, linkfile creation happens as root.

use uid/gid returned from _cbk (link/rename) to set the correct ownership of
the link files.

Also added test/dht.rc to implement common dht functions

BUG: 884597
Change-Id: I6bc0e04f62d4716fc033681e5678e852a1be7a2f
Signed-off-by: shishir gowda <sgowda@redhat.com>
Reviewed-on: http://review.gluster.org/4607
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>

6 files changed, 282 insertions, 4 deletions

diff --git a/tests/bugs/bug-884597.t b/tests/bugs/bug-884597.t
new file mode 100755
index 00000000000..0d3495a6597
--- /dev/null
+++ b/tests/bugs/bug-884597.t
@@ -0,0 +1,109 @@
+#!/bin/bash
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../dht.rc
+
+cleanup;
+BRICK_COUNT=3
+function uid_gid_compare()
+{
+        val=1
+
+        if [ "$1" == "$3" ]
+        then
+                if [ "$2" == "$4" ]
+                        then
+                                val=0
+                fi
+        fi
+        echo "$val"
+}
+
+
+TEST glusterd
+TEST pidof glusterd
+
+TEST $CLI volume create $V0 $H0:$B0/${V0}0 $H0:$B0/${V0}1 $H0:$B0/${V0}2
+TEST $CLI volume start $V0
+
+## Mount FUSE
+TEST glusterfs --attribute-timeout=0 --entry-timeout=0 -s $H0 --volfile-id $V0 $M0;
+
+i=1
+NEW_UID=36
+NEW_GID=36
+
+TEST touch $M0/$i
+
+chown $NEW_UID:$NEW_GID $M0/$i
+## rename till file gets a linkfile
+
+while [ $i -ne 0 ]
+do
+        TEST mv $M0/$i $M0/$(( $i+1 ))
+        let i++
+        file_has_linkfile $i
+        has_link=$?
+        if [ $has_link -eq 2 ]
+        then
+                break;
+        fi
+done
+
+get_hashed_brick $i
+cached=$?
+
+# check if uid/gid on linkfile is created with correct uid/gid
+BACKEND_UID=`stat --printf=%u $B0/${V0}$cached/$i`;
+BACKEND_GID=`stat --printf=%g $B0/${V0}$cached/$i`;
+
+EXPECT "0" uid_gid_compare $NEW_UID $NEW_GID $BACKEND_UID $BACKEND_GID
+
+# remove linkfile from backend, and trigger a lookup heal. uid/gid should match
+rm -rf $B0/${V0}$cached/$i
+
+# without a unmount, we are not able to trigger a lookup based heal
+
+TEST umount $M0
+
+## Mount FUSE
+TEST glusterfs --attribute-timeout=0 --entry-timeout=0 -s $H0 --volfile-id $V0 $M0;
+
+lookup=`ls -l $M0/$i 2>/dev/null`
+
+# check if uid/gid on linkfile is created with correct uid/gid
+BACKEND_UID=`stat --printf=%u $B0/${V0}$cached/$i`;
+BACKEND_GID=`stat --printf=%g $B0/${V0}$cached/$i`;
+
+EXPECT "0" uid_gid_compare $NEW_UID $NEW_GID $BACKEND_UID $BACKEND_GID
+# create hardlinks. Make sure a linkfile gets created
+
+i=1
+NEW_UID=36
+NEW_GID=36
+
+TEST touch $M0/file
+chown $NEW_UID:$NEW_GID $M0/file;
+
+## ln till file gets a linkfile
+
+while [ $i -ne 0 ]
+do
+        TEST ln $M0/file $M0/link$i
+
+        file_has_linkfile link$i
+        has_link=$?
+        if [ $has_link -eq 2 ]
+        then
+                break;
+        fi
+        let i++
+done
+
+get_hashed_brick link$i
+cached=$?
+
+# check if uid/gid on linkfile is created with correct uid/gid
+BACKEND_UID=`stat --printf=%u $B0/${V0}$cached/link$i`;
+BACKEND_GID=`stat --printf=%g $B0/${V0}$cached/link$i`;
+
+EXPECT "0" uid_gid_compare $NEW_UID $NEW_GID $BACKEND_UID $BACKEND_GID
diff --git a/tests/dht.rc b/tests/dht.rc
new file mode 100644
index 00000000000..1c811a63c6e
--- /dev/null
+++ b/tests/dht.rc
@@ -0,0 +1,72 @@
+#!/bin/bash
+
+function get_layout()
+{
+        layout=`getfattr -n trusted.glusterfs.dht -e hex $1 2>&1|grep dht |cut -d = -f2`
+        if [ $? -eq 1]
+        then
+                return -1
+        else
+                return $layout
+        fi
+
+}
+
+## populates $BRICK1 and $BRICK2 with hashed/cached subvolume. These will be
+## used by get_cached_brick and get_hashed_brick
+
+function file_has_linkfile()
+{
+        k=0
+        l=0
+        while [ $k -lt $BRICK_COUNT ]
+        do
+                stat=`stat $B0/${V0}$k/$1 2>/dev/null`
+                if [ $? -eq 0 ]
+                then
+                        let l++
+                        let "BRICK${l}=$k"
+
+                fi
+                let k++
+        done
+        return $l
+}
+
+function get_cached_brick()
+{
+        i=1
+        brick=$BRICK1
+        while [ $i -lt 3 ]
+        do
+                test=`getfattr -n trusted.glusterfs.dht.linkto -e text $B0/${V0}$brick/$1 2>&1`
+                if [ $? -eq 1 ]
+                then
+                        cached=$brick
+                        i=$(( $i+3 ))
+                fi
+                brick=$BRICK1
+                let i++
+        done
+
+        return $cached
+}
+
+function get_hashed_brick()
+{
+        j=1
+        brick=$BRICK1
+        while [ $j -lt 3 ]
+        do
+                test=`getfattr -n trusted.glusterfs.dht.linkto -e text $B0/${V0}$brick/$1 2>&1`
+                if [ $? -eq 0 ]
+                then
+                        hashed=$brick
+                        j=$(( $j+3 ))
+                fi
+                brick=$BRICK2
+                let j++
+        done
+
+        return $hashed
+}
diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 3285f46f27a..9457978204a 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -760,6 +760,9 @@ dht_lookup_linkfile_create_cbk (call_frame_t *frame, void *cookie,
         }
 
 unwind:
+        if (local->linked == _gf_true)
+                dht_linkfile_attr_heal (frame, this);
+
         DHT_STRIP_PHASE1_FLAGS (&local->stbuf);
         DHT_STACK_UNWIND (lookup, frame, local->op_ret, local->op_errno,
                           local->inode, &local->stbuf, local->xattr,
@@ -3419,6 +3422,8 @@ dht_newfile_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 op_errno = EINVAL;
                 goto out;
         }
+        if (local->linked == _gf_true)
+                dht_linkfile_attr_heal (frame, this);
 out:
         /*
          * FIXME: ia_size and st_blocks of preparent and postparent do not have
@@ -3427,7 +3432,6 @@ out:
          * corresponding values from each of the subvolume.
          * See dht_iatt_merge for reference.
          */
-
         DHT_STRIP_PHASE1_FLAGS (stbuf);
         DHT_STACK_UNWIND (mknod, frame, op_ret, op_errno, inode, stbuf,
                           preparent, postparent, xdata);
@@ -3687,7 +3691,10 @@ dht_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 dht_inode_ctx_time_update (local->loc.parent, this,
                                            postparent, 1);
         }
-
+        if (local->linked == _gf_true) {
+                local->stbuf = *stbuf;
+                dht_linkfile_attr_heal (frame, this);
+        }
 out:
         DHT_STRIP_PHASE1_FLAGS (stbuf);
         DHT_STACK_UNWIND (link, frame, op_ret, op_errno, inode, stbuf, preparent,
@@ -3831,7 +3838,10 @@ dht_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 op_errno = EINVAL;
                 goto out;
         }
-
+        if (local->linked == _gf_true) {
+                local->stbuf = *stbuf;
+                dht_linkfile_attr_heal (frame, this);
+        }
 out:
         DHT_STRIP_PHASE1_FLAGS (stbuf);
         DHT_STACK_UNWIND (create, frame, op_ret, op_errno, fd, inode, stbuf, preparent,
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index e0164ee3f44..8f266ef8ef9 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -177,6 +177,9 @@ struct dht_local {
 
         glusterfs_fop_t      fop;
 
+        gf_boolean_t     linked;
+        xlator_t        *link_subvol;
+
         struct dht_rebalance_ rebalance;
 
 };
@@ -729,4 +732,6 @@ xlator_t *
 dht_subvol_with_free_space_inodes (xlator_t *this, xlator_t *subvol);
 xlator_t *
 dht_subvol_maxspace_nonzeroinode (xlator_t *this, xlator_t *subvol);
+int
+dht_linkfile_attr_heal (call_frame_t *frame, xlator_t *this);
 #endif/* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-linkfile.c b/xlators/cluster/dht/src/dht-linkfile.c
index 803f344aff1..67d6ce583a9 100644
--- a/xlators/cluster/dht/src/dht-linkfile.c
+++ b/xlators/cluster/dht/src/dht-linkfile.c
@@ -20,7 +20,7 @@
 #include "dht-common.h"
 
 
-
+#define is_equal(a, b) (a == b)
 int
 dht_linkfile_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                          int op_ret, int op_errno, inode_t *inode,
@@ -31,6 +31,9 @@ dht_linkfile_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         local = frame->local;
 
+        if (!op_ret)
+                local->linked = _gf_true;
+
         local->linkfile.linkfile_cbk (frame, cookie, this, op_ret, op_errno,
                                       inode, stbuf, preparent, postparent,
                                       xdata);
@@ -51,6 +54,8 @@ dht_linkfile_create (call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
         local->linkfile.linkfile_cbk = linkfile_cbk;
         local->linkfile.srcvol = tovol;
 
+        local->linked = _gf_false;
+
         dict = local->params;
         if (!dict) {
                 dict = dict_new ();
@@ -81,6 +86,7 @@ dht_linkfile_create (call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
                 goto out;
         }
 
+        local->link_subvol = fromvol;
         STACK_WIND (frame, dht_linkfile_create_cbk,
                     fromvol, fromvol->fops->mknod, loc,
                     S_IFREG | DHT_LINKFILE_MODE, 0, 0, dict);
@@ -188,3 +194,68 @@ dht_linkfile_subvol (xlator_t *this, inode_t *inode, struct iatt *stbuf,
 out:
         return subvol;
 }
+
+int
+dht_linkfile_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                          int op_ret, int op_errno, struct iatt *statpre,
+                          struct iatt *statpost, dict_t *xdata)
+{
+        dht_local_t *local = NULL;
+        loc_t *loc = NULL;
+
+        local = frame->local;
+        loc = &local->loc;
+
+        if (op_ret)
+                gf_log (this->name, GF_LOG_ERROR, "setattr of uid/gid on %s"
+                        " :<gfid:%s> failed (%s)",
+                        (loc->path? loc->path: "NULL"),
+                        uuid_utoa(local->gfid), strerror(op_errno));
+
+        DHT_STACK_DESTROY (frame);
+
+       return 0;
+}
+
+int
+dht_linkfile_attr_heal (call_frame_t *frame, xlator_t *this)
+{
+        int     ret = -1;
+        call_frame_t *copy = NULL;
+        dht_local_t  *local = NULL;
+        dht_local_t  *copy_local = NULL;
+        xlator_t     *subvol = NULL;
+        struct iatt   stbuf = {0,};
+
+        local = frame->local;
+
+        GF_VALIDATE_OR_GOTO ("dht", local, out);
+        GF_VALIDATE_OR_GOTO ("dht", local->link_subvol, out);
+
+        if ((local->stbuf.ia_type == IA_INVAL) ||
+            (is_equal (frame->root->uid, local->stbuf.ia_uid) &&
+             is_equal (frame->root->gid, local->stbuf.ia_gid)))
+                return 0;
+
+        copy = copy_frame (frame);
+
+        if (!copy)
+                goto out;
+
+       copy_local = dht_local_init (copy, &local->loc, NULL, 0);
+
+       if (!copy_local)
+               goto out;
+
+        stbuf = local->stbuf;
+        subvol = local->link_subvol;
+
+        copy->local = copy_local;
+
+        STACK_WIND (copy, dht_linkfile_setattr_cbk, subvol,
+                    subvol->fops->setattr, &copy_local->loc,
+                    &stbuf, (GF_SET_ATTR_UID | GF_SET_ATTR_GID), NULL);
+        ret = 0;
+out:
+        return ret;
+}
diff --git a/xlators/cluster/dht/src/dht-rename.c b/xlators/cluster/dht/src/dht-rename.c
index f1a4364df20..35fedeaa7a4 100644
--- a/xlators/cluster/dht/src/dht-rename.c
+++ b/xlators/cluster/dht/src/dht-rename.c
@@ -441,6 +441,10 @@ dht_rename_links_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         local->loc.path, prev->this->name, strerror (op_errno));
         }
 
+        if (local->linked == _gf_true) {
+                local->linked = _gf_false;
+                dht_linkfile_attr_heal (frame, this);
+        }
         DHT_STACK_DESTROY (frame);
 
         return 0;
@@ -511,6 +515,11 @@ err:
         dht_iatt_merge (this, &local->preparent, prenewparent, prev->this);
         dht_iatt_merge (this, &local->postparent, postnewparent, prev->this);
 
+        if (local->linked == _gf_true) {
+                local->linked = _gf_false;
+                dht_linkfile_attr_heal (frame, this);
+        }
+
         /* NOTE: rename_subvol is the same subvolume from which dht_rename_cbk
          *       is called. since rename has already happened on rename_subvol,
          *       unlink should not be sent for oldpath (either linkfile or cached-file)
@@ -645,6 +654,8 @@ dht_rename_links_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 local->op_ret   = -1;
                 if (op_errno != ENOENT)
                         local->op_errno = op_errno;
+        } else {
+                dht_iatt_merge (this, &local->stbuf, stbuf, prev->this);
         }
 
         this_call_cnt = dht_frame_return (frame);