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.

Change-Id: I5345cff193d5095442ca446fbe5ea05f2c2d86a3
Signed-off-by: shishir gowda <sgowda@redhat.com>
BUG: 884597
Reviewed-on: http://review.gluster.org/5024
Reviewed-by: Vijay Bellur <vbellur@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>

4 files changed, 96 insertions, 3 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index ade05f38d3c..aedc453e83a 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -705,6 +705,8 @@ dht_lookup_linkfile_create_cbk (call_frame_t *frame, void *cookie,
 
 unwind:
         WIPE (&local->postparent);
+        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,
@@ -3188,6 +3190,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
@@ -3196,7 +3200,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);
@@ -3584,7 +3587,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 a92e9bccb77..0af21d10d0a 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -158,6 +158,9 @@ struct dht_local {
 
         glusterfs_fop_t      fop;
 
+        gf_boolean_t     linked;
+        xlator_t        *link_subvol;
+
         struct dht_rebalance_ rebalance;
 
 };
@@ -669,4 +672,6 @@ 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);