make snapview-server more compatible with NFS server

* There was no handle based API for listxattr. With this change, glfs_h_getxattrs
  also handles the listxattr functionality by checking whether the name is NULL
  or not (like posix). But all the gfapi functions for listxattr
  (glfs_h_getxattrs AND glfs_listxattr AND glfs_flistxattr) returns the names of
  the xattrs in a buffer provided by the caller. But snapview-server has to
  return the list of xattrs in a dict itself (similar to posix xlator). But
  the buffer just contains the names of the xattrs. So for each xattr, a zero
  byte value is set (i.e. "") into the dict and sent back. Translators which
  do xattr caching (as of now md-cache which caches selinux and acl related
  xattrs) should not cache those xattrs whose value is a zero byte data ("").
  So made changes in md-cache to ignore zero byte values.

* NFS server was not linking the inodes to inode table in readdirp. This was
  leading to applications getting errors. The below set of operations would
  lead to applications getting error
  1) ls -l in one of the snaopshots (snapview-server would generate gfids for
     each entry on the fly and link the inodes associated with those entries)
  2) NFS server upon getting readdirp reply would not link the inodes of the
     entries. But it used to generate filehandles for each entry and associate
     the gfid of that entry with the filehandle and send it as part of the
     reply to nfs client.
  3) NFS client would send the filehandle of one of those entries when some
     activity is done on it.
  4) NFS server would not be able to find the inode for the gfid present in the
     filehandle (as the inode was not linked) and would go for hard resolution
     by sending a lookup on the gfid by creating a new inode.
  5) snapview-client will not able to identify whether the inode is a real inode
     existing in the main volume or a virtual inode existing in the snapshots
     as there would not be any inode context.
  6) Since the gfid upon which lookup is sent is a virtual gfid which is not
     present in the disk, lookup would fail and the application would get an
     error.

  To handle above situation, now nfs server also does inode linking in readdirp.

Change-Id: Ibb191408347b6b5f21cff72319ccee619ea77bcd
BUG: 1115949
Signed-off-by: Raghavendra Bhat <raghavendra@redhat.com>
Reviewed-on: http://review.gluster.org/8230
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Niels de Vos <ndevos@redhat.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

1 files changed, 204 insertions, 78 deletions

diff --git a/xlators/features/snapview-server/src/snapview-server.c b/xlators/features/snapview-server/src/snapview-server.c
index 47344113406..99965a4ccef 100644
--- a/xlators/features/snapview-server/src/snapview-server.c
+++ b/xlators/features/snapview-server/src/snapview-server.c
@@ -1627,18 +1627,73 @@ out:
         return 0;
 }
 
+/*
+ * This function adds the xattr keys present in the list (@list) to the dict.
+ * But the list contains only the names of the xattrs (and no value, as
+ * the gfapi functions for the listxattr operations would return only the
+ * names of the xattrs in the buffer provided by the caller, though they had
+ * got the values of those xattrs from posix) as described in the man page of
+ * listxattr. But before unwinding snapview-server has to put those names
+ * back into the dict. But to get the values for those xattrs it has to do the
+ * getxattr operation on each xattr which might turn out to be a costly
+ * operation. So for each of the xattrs present in the list, a 0 byte value
+ * ("") is set into the dict before unwinding. This can be treated as an
+ * indicator to other xlators which want to cache the xattrs (as of now,
+ * md-cache which caches acl and selinux related xattrs) to not to cache the
+ * values of the xattrs present in the dict.
+ */
+int32_t
+svs_add_xattrs_to_dict (xlator_t *this, dict_t *dict, char *list, ssize_t size)
+{
+        char           keybuffer[4096]  = {0,};
+        size_t         remaining_size   = 0;
+        int32_t        list_offset      = 0;
+        int32_t        ret              = -1;
+
+        GF_VALIDATE_OR_GOTO ("snapview-daemon", this, out);
+        GF_VALIDATE_OR_GOTO (this->name, dict, out);
+        GF_VALIDATE_OR_GOTO (this->name, list, out);
+
+        remaining_size = size;
+        list_offset = 0;
+        while (remaining_size > 0) {
+                strcpy (keybuffer, list + list_offset);
+#ifdef GF_DARWIN_HOST_OS
+                /* The protocol expect namespace for now */
+                char *newkey = NULL;
+                gf_add_prefix (XATTR_USER_PREFIX, keybuffer, &newkey);
+                strcpy (keybuffer, newkey);
+                GF_FREE (newkey);
+#endif
+                ret = dict_set_str (dict, keybuffer, "");
+                if (ret < 0) {
+                        gf_log (this->name, GF_LOG_ERROR, "dict set operation "
+                                "for the key %s failed.", keybuffer);
+                        goto out;
+                }
+
+                remaining_size -= strlen (keybuffer) + 1;
+                list_offset += strlen (keybuffer) + 1;
+        } /* while (remaining_size > 0) */
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
 int32_t
 svs_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name,
               dict_t *xdata)
 {
-        svs_inode_t   *inode_ctx  = NULL;
-        int32_t        op_ret     = -1;
-        int32_t        op_errno   = EINVAL;
-        glfs_t        *fs         = NULL;
-        glfs_object_t *object     = NULL;
-        char          *value      = 0;
-        ssize_t        size       = 0;
-        dict_t        *dict       = NULL;
+        svs_inode_t   *inode_ctx        = NULL;
+        int32_t        op_ret           = -1;
+        int32_t        op_errno         = EINVAL;
+        glfs_t        *fs               = NULL;
+        glfs_object_t *object           = NULL;
+        char          *value            = 0;
+        ssize_t        size             = 0;
+        dict_t        *dict             = NULL;
 
         GF_VALIDATE_OR_GOTO ("snap-view-daemon", this, out);
         GF_VALIDATE_OR_GOTO ("snap-view-daemon", frame, out);
@@ -1654,8 +1709,11 @@ svs_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name,
                 goto out;
         }
 
-        /* Fake success is sent if the getxattr is on entry point directory
-           or the inode is SNAP_VIEW_ENTRY_POINT_INODE
+        /* EINVAL is sent if the getxattr is on entry point directory
+           or the inode is SNAP_VIEW_ENTRY_POINT_INODE. Entry point is
+           a virtual directory on which setxattr operations are not
+           allowed. If getxattr has to be faked as success, then a value
+           for the name of the xattr has to be sent which we dont have.
         */
         if (inode_ctx->type == SNAP_VIEW_ENTRY_POINT_INODE) {
                 op_ret = -1;
@@ -1664,19 +1722,30 @@ svs_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name,
         } else if (inode_ctx->type == SNAP_VIEW_VIRTUAL_INODE) {
                 fs = inode_ctx->fs;
                 object = inode_ctx->object;
+                dict = dict_new ();
+                if (!dict) {
+                        gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                "allocate dict");
+                        op_ret = -1;
+                        op_errno = ENOMEM;
+                        goto out;
+                }
+
                 size = glfs_h_getxattrs (fs, object, name, NULL, 0);
                 if (size == -1) {
-                        gf_log (this->name, GF_LOG_ERROR, "getxattr on %s "
-                                "failed (key: %s)", loc->name, name);
+                        gf_log (this->name, GF_LOG_ERROR, "getxattr "
+                                "on %s failed (key: %s)", loc->name,
+                                name);
                         op_ret = -1;
                         op_errno = errno;
                         goto out;
                 }
-                value = GF_CALLOC (size + 1, sizeof (char), gf_common_mt_char);
+                value = GF_CALLOC (size + 1, sizeof (char),
+                                   gf_common_mt_char);
                 if (!value) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to allocate "
-                                "memory for getxattr on %s (key: %s)",
-                                loc->name, name);
+                        gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                "allocate memory for getxattr on %s "
+                                "(key: %s)", loc->name, name);
                         op_ret = -1;
                         op_errno = ENOMEM;
                         goto out;
@@ -1684,35 +1753,38 @@ svs_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name,
 
                 size = glfs_h_getxattrs (fs, object, name, value, size);
                 if (size == -1) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to get the "
-                                "xattr %s for entry %s", name, loc->name);
+                        gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                "get the xattr %s for entry %s", name,
+                                loc->name);
                         op_ret = -1;
                         op_errno = errno;
                         goto out;
                 }
                 value[size] = '\0';
 
-                dict = dict_new ();
-                if (!dict) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to allocate "
-                                "dict");
-                        op_ret = -1;
-                        op_errno = ENOMEM;
-                        goto out;
-                }
-
-                op_ret = dict_set_dynptr (dict, (char *)name, value, size);
-                if (op_ret < 0) {
-                        op_errno = -op_ret;
-                        gf_log (this->name, GF_LOG_ERROR, "dict set operation "
-                                "for %s for the key %s failed.", loc->path,
-                                name);
+                if (name) {
+                        op_ret = dict_set_dynptr (dict, (char *)name, value,
+                                                  size);
+                        if (op_ret < 0) {
+                                op_errno = -op_ret;
+                                gf_log (this->name, GF_LOG_ERROR, "dict set "
+                                        "operation for %s for the key %s "
+                                        "failed.", loc->path, name);
+                                GF_FREE (value);
+                                value = NULL;
+                                goto out;
+                        }
+                } else {
+                        op_ret = svs_add_xattrs_to_dict (this, dict, value,
+                                                         size);
+                        if (op_ret == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                        "add the xattrs from the list to dict");
+                                op_errno = ENOMEM;
+                                goto out;
+                        }
                         GF_FREE (value);
-                        goto out;
                 }
-
-                op_ret = 0;
-                op_errno = 0;
         }
 
 out:
@@ -1721,6 +1793,9 @@ out:
 
         STACK_UNWIND_STRICT (getxattr, frame, op_ret, op_errno, dict, NULL);
 
+        if (dict)
+                dict_unref (dict);
+
         return 0;
 }
 
@@ -1761,8 +1836,11 @@ svs_fgetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, const char *name,
         }
 
         glfd = sfd->fd;
-        /* Fake success is sent if the getxattr is on entry point directory
-           or the inode is SNAP_VIEW_ENTRY_POINT_INODE
+        /* EINVAL is sent if the getxattr is on entry point directory
+           or the inode is SNAP_VIEW_ENTRY_POINT_INODE. Entry point is
+           a virtual directory on which setxattr operations are not
+           allowed. If getxattr has to be faked as success, then a value
+           for the name of the xattr has to be sent which we dont have.
         */
         if (inode_ctx->type == SNAP_VIEW_ENTRY_POINT_INODE) {
                 op_ret = -1;
@@ -1771,53 +1849,98 @@ svs_fgetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, const char *name,
         }
 
         if (inode_ctx->type == SNAP_VIEW_VIRTUAL_INODE) {
-                size = glfs_fgetxattr (glfd, name, NULL, 0);
-                if (size == -1) {
-                        gf_log (this->name, GF_LOG_ERROR, "getxattr on %s "
-                                "failed (key: %s)", uuid_utoa (fd->inode->gfid),
-                                name);
-                        op_ret = -1;
-                        op_errno = errno;
-                        goto out;
-                }
-                value = GF_CALLOC (size + 1, sizeof (char), gf_common_mt_char);
-                if (!value) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to allocate "
-                                "memory for getxattr on %s (key: %s)",
-                                uuid_utoa (fd->inode->gfid), name);
-                        op_ret = -1;
-                        op_errno = ENOMEM;
-                        goto out;
-                }
-
-                size = glfs_fgetxattr (glfd, name, value, size);
-                if (size == -1) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to get the "
-                                "xattr %s for inode %s", name,
-                                uuid_utoa (fd->inode->gfid));
-                        op_ret = -1;
-                        op_errno = errno;
-                        goto out;
-                }
-                value[size] = '\0';
-
                 dict = dict_new ();
                 if (!dict) {
-                        gf_log (this->name, GF_LOG_ERROR, "failed to allocate "
-                                "dict");
+                        gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                "allocate  dict");
                         op_ret = -1;
                         op_errno = ENOMEM;
                         goto out;
                 }
 
-                op_ret = dict_set_dynptr (dict, (char *)name, value, size);
-                if (op_ret < 0) {
-                        op_errno = -op_ret;
-                        gf_log (this->name, GF_LOG_ERROR, "dict set operation "
-                                "for gfid %s for the key %s failed.",
-                                uuid_utoa (fd->inode->gfid), name);
+                if (name) {
+                        size = glfs_fgetxattr (glfd, name, NULL, 0);
+                        if (size == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "getxattr on "
+                                        "%s failed (key: %s)",
+                                        uuid_utoa (fd->inode->gfid), name);
+                                op_ret = -1;
+                                op_errno = errno;
+                                goto out;
+                        }
+                        value = GF_CALLOC (size + 1, sizeof (char),
+                                           gf_common_mt_char);
+                        if (!value) {
+                                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                        "allocate memory for getxattr on %s "
+                                        "(key: %s)",
+                                        uuid_utoa (fd->inode->gfid), name);
+                                op_ret = -1;
+                                op_errno = ENOMEM;
+                                goto out;
+                        }
+
+                        size = glfs_fgetxattr (glfd, name, value, size);
+                        if (size == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                        "get the xattr %s for inode %s", name,
+                                        uuid_utoa (fd->inode->gfid));
+                                op_ret = -1;
+                                op_errno = errno;
+                                goto out;
+                        }
+                        value[size] = '\0';
+
+                        op_ret = dict_set_dynptr (dict, (char *)name, value,
+                                                  size);
+                        if (op_ret < 0) {
+                                op_errno = -op_ret;
+                                gf_log (this->name, GF_LOG_ERROR, "dict set "
+                                        "operation for gfid %s for the key %s "
+                                        "failed.",
+                                        uuid_utoa (fd->inode->gfid), name);
+                                GF_FREE (value);
+                                goto out;
+                        }
+                } else {
+                        size = glfs_flistxattr (glfd, NULL, 0);
+                        if (size == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "listxattr "
+                                        "on %s failed",
+                                        uuid_utoa (fd->inode->gfid));
+                                goto out;
+                        }
+
+                        value = GF_CALLOC (size + 1, sizeof (char),
+                                           gf_common_mt_char);
+                        if (!value) {
+                                op_ret = -1;
+                                op_errno = ENOMEM;
+                                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                        "allocate buffer for xattr list (%s)",
+                                        uuid_utoa (fd->inode->gfid));
+                                goto out;
+                        }
+
+                        size = glfs_flistxattr (glfd, value, size);
+                        if (size == -1) {
+                                op_ret = -1;
+                                op_errno = errno;
+                                gf_log (this->name, GF_LOG_ERROR, "listxattr "
+                                        "on %s failed",
+                                        uuid_utoa (fd->inode->gfid));
+                                goto out;
+                        }
+
+                        op_ret = svs_add_xattrs_to_dict (this, dict, value,
+                                                         size);
+                        if (op_ret == -1) {
+                                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                        "add the xattrs from the list to dict");
+                                op_errno = ENOMEM;
+                                goto out;
+                        }
                         GF_FREE (value);
-                        goto out;
                 }
 
                 op_ret = 0;
@@ -1830,6 +1953,9 @@ out:
 
         STACK_UNWIND_STRICT (fgetxattr, frame, op_ret, op_errno, dict, NULL);
 
+        if (dict)
+                dict_unref (dict);
+
         return 0;
 }