features/shard: Fix permission issues

        Backport of: http://review.gluster.org/11992

This patch does the following:
* reverts commit b467af0e99b39ef708420d3f7f6696b0ca618512
* changes ownership on shards under /.shard to be root:root
* makes readv, writev, [f]truncate, rename, and unlink fops
  to perform operations on files under /.shard with
  frame->root->{uid,gid} as 0.

This would ensure that a [f]setattr on a sharded file
does not need to be called on all the shards associated with it.

Change-Id: I50d8533bd2b769a4dfe8cd1b49bdcfc117a7e660
BUG: 1253151
Signed-off-by: Krutika Dhananjay <kdhananj@redhat.com>
Reviewed-on: http://review.gluster.org/12052
Reviewed-by: Pranith Kumar Karampuri <pkarampu@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Tested-by: NetBSD Build System <jenkins@build.gluster.org>

2 files changed, 57 insertions, 8 deletions

diff --git a/xlators/features/shard/src/shard.c b/xlators/features/shard/src/shard.c
index 2c893cc1101..6fb57168371 100644
--- a/xlators/features/shard/src/shard.c
+++ b/xlators/features/shard/src/shard.c
@@ -913,6 +913,8 @@ shard_truncate_last_shard_cbk (call_frame_t *frame, void *cookie,
 
         local = frame->local;
 
+        SHARD_UNSET_ROOT_FS_ID (frame, local);
+
         if (op_ret < 0) {
                 local->op_ret = op_ret;
                 local->op_errno = op_errno;
@@ -948,6 +950,19 @@ shard_truncate_last_shard (call_frame_t *frame, xlator_t *this, inode_t *inode)
 
         local = frame->local;
 
+        /* A NULL inode could be due to the fact that the last shard which
+         * needs to be truncated does not exist due to it lying in a hole
+         * region. So the only thing left to do in that case would be an
+         * update to file size xattr.
+         */
+        if (!inode) {
+                shard_update_file_size (frame, this, NULL, &local->loc,
+                                       shard_post_update_size_truncate_handler);
+                return 0;
+        }
+
+        SHARD_SET_ROOT_FS_ID (frame, local);
+
         loc.inode = inode_ref (inode);
         gf_uuid_copy (loc.gfid, inode->gfid);
 
@@ -1012,6 +1027,8 @@ shard_truncate_htol (call_frame_t *frame, xlator_t *this, inode_t *inode)
 
         local->call_count = call_count;
         i = 1;
+
+        SHARD_SET_ROOT_FS_ID (frame, local);
         while (cur_block <= last_block) {
                 if (!local->inode_list[i]) {
                         cur_block++;
@@ -1145,7 +1162,9 @@ shard_common_lookup_shards_cbk (call_frame_t *frame, void *cookie,
         if (op_ret < 0) {
                 /* Ignore absence of shards in the backend in truncate fop. */
                 if (((local->fop == GF_FOP_TRUNCATE) ||
-                    (local->fop == GF_FOP_FTRUNCATE)) && (op_errno == ENOENT))
+                    (local->fop == GF_FOP_FTRUNCATE) ||
+                    (local->fop == GF_FOP_RENAME) ||
+                    (local->fop == GF_FOP_UNLINK)) && (op_errno == ENOENT))
                         goto done;
                 local->op_ret = op_ret;
                 local->op_errno = op_errno;
@@ -1711,6 +1730,8 @@ shard_unlink_shards_do_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 done:
         call_count = shard_call_count_return (frame);
         if (call_count == 0) {
+                SHARD_UNSET_ROOT_FS_ID (frame, local);
+
                 if (local->fop == GF_FOP_UNLINK)
                         shard_unlink_base_file (frame, this);
                 else if (local->fop == GF_FOP_RENAME)
@@ -1768,6 +1789,8 @@ shard_unlink_shards_do (call_frame_t *frame, xlator_t *this, inode_t *inode)
                 return 0;
         }
 
+        SHARD_SET_ROOT_FS_ID (frame, local);
+
         while (cur_block <= last_block) {
                 /* The base file is unlinked in the end to mark the
                  * successful completion of the fop.
@@ -2377,6 +2400,7 @@ out:
                 fd_unref (anon_fd);
         call_count = shard_call_count_return (frame);
         if (call_count == 0) {
+                SHARD_UNSET_ROOT_FS_ID (frame, local);
                 if (local->op_ret < 0) {
                         SHARD_STACK_UNWIND (readv, frame, local->op_ret,
                                             local->op_errno, NULL, 0, NULL,
@@ -2423,6 +2447,8 @@ shard_readv_do (call_frame_t *frame, xlator_t *this)
         remaining_size = local->total_size;
         local->call_count = call_count = local->num_blocks;
 
+        SHARD_SET_ROOT_FS_ID (frame, local);
+
         while (cur_block <= last_block) {
                 if (wind_failed) {
                         shard_readv_do_cbk (frame, (void *) (long) 0,  this, -1,
@@ -2552,8 +2578,7 @@ shard_common_mknod_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 done:
         call_count = shard_call_count_return (frame);
         if (call_count == 0) {
-                frame->root->uid = local->uid;
-                frame->root->gid = local->gid;
+                SHARD_UNSET_ROOT_FS_ID (frame, local);
                 local->post_mknod_handler (frame, this);
         }
 
@@ -2587,10 +2612,7 @@ shard_common_resume_mknod (call_frame_t *frame, xlator_t *this,
         call_count = local->call_count = local->create_count;
         local->post_mknod_handler = post_mknod_handler;
 
-        local->uid = frame->root->uid;
-        local->gid = frame->root->gid;
-        frame->root->uid = local->prebuf.ia_uid;
-        frame->root->gid = local->prebuf.ia_gid;
+        SHARD_SET_ROOT_FS_ID (frame, local);
 
         ret = shard_inode_ctx_get_all (fd->inode, this, &ctx_tmp);
         if (ret) {
@@ -2676,6 +2698,7 @@ err:
          * This block is for handling failure in shard_inode_ctx_get_all().
          * Failures in the while-loop are handled within the loop.
          */
+        SHARD_UNSET_ROOT_FS_ID (frame, local);
         post_mknod_handler (frame, this);
         return 0;
 }
@@ -2925,6 +2948,7 @@ shard_writev_do_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
         call_count = shard_call_count_return (frame);
         if (call_count == 0) {
+                SHARD_UNSET_ROOT_FS_ID (frame, local);
                 if (local->op_ret < 0) {
                         SHARD_STACK_UNWIND (writev, frame, local->written_size,
                                             local->op_errno, NULL, NULL, NULL);
@@ -2974,10 +2998,13 @@ shard_writev_do (call_frame_t *frame, xlator_t *this)
         local->call_count = call_count = local->num_blocks;
         last_block = local->last_block;
 
+        SHARD_SET_ROOT_FS_ID (frame, local);
+
         if (dict_set_uint32 (local->xattr_req,
                              GLUSTERFS_WRITE_UPDATE_ATOMIC, 4)) {
                 local->op_ret = -1;
                 local->op_errno = ENOMEM;
+                local->call_count = 1;
                 shard_writev_do_cbk (frame, (void *)(long)0, this, -1, ENOMEM,
                                      NULL, NULL, NULL);
                 return 0;
@@ -3152,6 +3179,8 @@ shard_writev_mkdir_dot_shard_cbk (call_frame_t *frame, void *cookie,
 
         local = frame->local;
 
+        SHARD_UNSET_ROOT_FS_ID (frame, local);
+
         if (op_ret == -1) {
                 if (op_errno != EEXIST) {
                         goto unwind;
@@ -3201,9 +3230,11 @@ shard_writev_mkdir_dot_shard (call_frame_t *frame, xlator_t *this)
                 goto err;
         }
 
+        SHARD_SET_ROOT_FS_ID (frame, local);
+
         STACK_WIND (frame, shard_writev_mkdir_dot_shard_cbk, FIRST_CHILD(this),
                     FIRST_CHILD(this)->fops->mkdir, &local->dot_shard_loc,
-                    0777, 0, xattr_req);
+                    0755, 0, xattr_req);
         dict_unref (xattr_req);
         return 0;
 
diff --git a/xlators/features/shard/src/shard.h b/xlators/features/shard/src/shard.h
index f480c2912b6..e859b96c4db 100644
--- a/xlators/features/shard/src/shard.h
+++ b/xlators/features/shard/src/shard.h
@@ -116,6 +116,23 @@
         }                                                                     \
 } while (0)
 
+#define SHARD_SET_ROOT_FS_ID(frame, local) do {                               \
+                if (!local->is_set_fsid) {                                    \
+                        local->uid = frame->root->uid;                        \
+                        local->gid = frame->root->gid;                        \
+                        frame->root->uid = 0;                                 \
+                        frame->root->gid = 0;                                 \
+                        local->is_set_fsid = _gf_true;                        \
+                }                                                             \
+} while (0)
+
+#define SHARD_UNSET_ROOT_FS_ID(frame, local) do {                             \
+                if (local->is_set_fsid) {                                     \
+                        frame->root->uid = local->uid;                        \
+                        frame->root->gid = local->gid;                        \
+                        local->is_set_fsid = _gf_false;                       \
+                }                                                             \
+} while (0)
 
 typedef struct shard_priv {
         uint64_t block_size;
@@ -184,6 +201,7 @@ typedef struct shard_local {
         struct iobref *iobref;
         struct iobuf *iobuf;
         gf_dirent_t entries_head;
+        gf_boolean_t is_set_fsid;
         gf_boolean_t list_inited;
         gf_boolean_t is_write_extending;
         shard_post_fop_handler_t handler;