storage/posix: Add virtual xattr to fetch path from gfid

The gfid2path infra stores the "pargfid/bname" as on xattr
value for each non directory entry. Hardlinks would have a
separate xattr. This xattr key is internal and is not
exposed to applications.  A virtual xattr is exposed for
the applications to fetch the path from gfid.

Internal xattr:
trusted.gfid2path.<xxhash>

Virtual xattr:
glusterfs.gfidtopath

getfattr -h -n glusterfs.gfidtopath /<aux-mnt>/.gfid/<gfid>

If there are hardlinks, it returns all the paths separated
by ':'.

A volume set option is introduced to change the delimiter
to required string of max length 7.

gluster vol set gfid2path-separator ":::"

Updates: #139
Change-Id: Ie3b0c3fd8bd5333c4a27410011e608333918c02a
Signed-off-by: Kotresh HR <khiremat@redhat.com>
Reviewed-on: https://review.gluster.org/17785
Smoke: Gluster Build System <jenkins@build.gluster.org>
CentOS-regression: Gluster Build System <jenkins@build.gluster.org>
Reviewed-by: Krutika Dhananjay <kdhananj@redhat.com>

9 files changed, 424 insertions, 0 deletions

diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h
index fd659293504..bb925c3d78d 100644
--- a/libglusterfs/src/glusterfs.h
+++ b/libglusterfs/src/glusterfs.h
@@ -121,6 +121,7 @@
 #define GF_XATTR_LINKINFO_KEY   "trusted.distribute.linkinfo"
 #define GFID_XATTR_KEY          "trusted.gfid"
 #define PGFID_XATTR_KEY_PREFIX  "trusted.pgfid."
+#define GFID2PATH_VIRT_XATTR_KEY  "glusterfs.gfidtopath"
 #define GFID2PATH_XATTR_KEY_PREFIX  "trusted.gfid2path."
 #define GFID2PATH_XATTR_KEY_PREFIX_LENGTH 18
 #define VIRTUAL_GFID_XATTR_KEY_STR  "glusterfs.gfid.string"
diff --git a/tests/gfid2path/get-gfid-to-path.t b/tests/gfid2path/get-gfid-to-path.t
new file mode 100644
index 00000000000..71c2c2fab70
--- /dev/null
+++ b/tests/gfid2path/get-gfid-to-path.t
@@ -0,0 +1,75 @@
+#!/bin/bash
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+. $(dirname $0)/../afr.rc
+
+cleanup;
+
+TEST glusterd
+TEST pidof glusterd
+
+## Create a 1*2 volume
+TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{0,1}
+EXPECT "$V0" volinfo_field $V0 'Volume Name';
+EXPECT 'Created' volinfo_field $V0 'Status';
+
+## Start the volume
+TEST $CLI volume start $V0
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT "Y" glustershd_up_status
+EXPECT_WITHIN $CHILD_UP_TIMEOUT "1" afr_child_up_status_in_shd $V0 0
+EXPECT_WITHIN $CHILD_UP_TIMEOUT "1" afr_child_up_status_in_shd $V0 1
+
+## enable gfid2path
+TEST $CLI volume set $V0 gfid2path enable
+
+## Mount the volume
+TEST $GFS --volfile-server=$H0 --aux-gfid-mount --volfile-id=$V0 $M0;
+
+root_gfid="00000000-0000-0000-0000-000000000001"
+
+#Check for ROOT
+EXPECT "/" get_gfid2path $M0/.gfid/$root_gfid
+
+#CREATE
+fname=$M0/file1
+touch $fname;
+
+#Get gfid of file1
+gfid=$(getfattr -h --only-values -n glusterfs.gfid.string $M0/file1)
+
+#Get path from virt xattr
+EXPECT "/file1" get_gfid2path $M0/.gfid/$gfid
+
+#Create hardlink and get path
+ln $fname $M0/hl_file1
+EXPECT "/file1" get_gfid2path $M0/.gfid/$gfid
+EXPECT "/hl_file1" get_gfid2path $M0/.gfid/$gfid
+
+#Rename and get path
+mv $fname $M0/rn_file1
+EXPECT "/hl_file1" get_gfid2path $M0/.gfid/$gfid
+EXPECT "/rn_file1" get_gfid2path $M0/.gfid/$gfid
+
+#Create symlink and get path
+ln -s $fname $M0/sym_file1
+gfid=$(getfattr -h --only-values -n glusterfs.gfid.string $M0/sym_file1)
+EXPECT "/sym_file1" get_gfid2path $M0/.gfid/$gfid
+
+#Create dir and get path
+mkdir -p $M0/dir1/dir2
+gfid=$(getfattr -h --only-values -n glusterfs.gfid.string $M0/dir1/dir2)
+EXPECT "/dir1/dir2" get_gfid2path $M0/.gfid/$gfid
+
+#Create file under dir2 and get path
+touch $M0/dir1/dir2/file1
+gfid=$(getfattr -h --only-values -n glusterfs.gfid.string $M0/dir1/dir2/file1)
+EXPECT "/dir1/dir2/file1" get_gfid2path $M0/.gfid/$gfid
+
+#Create hardlink under dir2 and get path
+ln $M0/dir1/dir2/file1 $M0/dir1/hl_file1
+gfid=$(getfattr -h --only-values -n glusterfs.gfid.string $M0/dir1/dir2/file1)
+EXPECT "/dir1/dir2/file1" get_gfid2path $M0/.gfid/$gfid
+EXPECT "/dir1/hl_file1" get_gfid2path $M0/.gfid/$gfid
+
+cleanup;
diff --git a/tests/volume.rc b/tests/volume.rc
index 402bb9dbf19..369e5a706c6 100644
--- a/tests/volume.rc
+++ b/tests/volume.rc
@@ -359,6 +359,11 @@ function get_text_xattr {
         getfattr -h -d -m. -e text $path 2>/dev/null | grep -a $key | cut -f2 -d'='
 }
 
+function get_gfid2path {
+        local path=$1
+        getfattr -h --only-values -n glusterfs.gfidtopath $path 2>/dev/null
+}
+
 function get_xattr_key {
         local key=$1
         local path=$2
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index 92ff4ba2e60..d336d383986 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -2795,6 +2795,11 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .voltype     = "storage/posix",
           .op_version  = GD_OP_VERSION_3_12_0,
         },
+        { .option      = "gfid2path-separator",
+          .key         = "storage.gfid2path-separator",
+          .voltype     = "storage/posix",
+          .op_version  = GD_OP_VERSION_3_12_0,
+        },
         { .key         = "storage.reserve",
           .voltype     = "storage/posix",
           .op_version  = GD_OP_VERSION_3_12_0,
diff --git a/xlators/storage/posix/src/posix-gfid-path.c b/xlators/storage/posix/src/posix-gfid-path.c
index 7529f559fc7..2834c2fc4fd 100644
--- a/xlators/storage/posix/src/posix-gfid-path.c
+++ b/xlators/storage/posix/src/posix-gfid-path.c
@@ -13,6 +13,9 @@
 #include "syscall.h"
 #include "logging.h"
 #include "posix-messages.h"
+#include "posix-mem-types.h"
+#include "posix-gfid-path.h"
+#include "posix.h"
 
 int32_t
 posix_set_gfid2path_xattr (xlator_t *this, const char *path, uuid_t pgfid,
@@ -96,3 +99,184 @@ posix_is_gfid2path_xattr (const char *name)
                 return _gf_false;
         }
 }
+
+static int gf_posix_xattr_enotsup_log;
+
+int32_t
+posix_get_gfid2path (xlator_t *this, inode_t *inode, const char *real_path,
+                     int *op_errno, dict_t *dict)
+{
+        int                    ret                             = 0;
+        char                  *path                            = NULL;
+        ssize_t                size                            = 0;
+        char                  *list                            = NULL;
+        int32_t                list_offset                     = 0;
+        int32_t                i                               = 0;
+        int32_t                j                               = 0;
+        char                  *paths[MAX_GFID2PATH_LINK_SUP]   = {NULL,};
+        char                  *value                           = NULL;
+        size_t                 remaining_size                  = 0;
+        size_t                 bytes                           = 0;
+        char                   keybuffer[4096]                 = {0,};
+        char                   value_buf[8192]                 = {0,};
+        uuid_t                 pargfid                         = {0,};
+        gf_boolean_t           have_val                        = _gf_false;
+        struct posix_private  *priv                            = NULL;
+        char                   pargfid_str[UUID_CANONICAL_FORM_LEN + 1] = {0,};
+
+        priv = this->private;
+
+        if (IA_ISDIR (inode->ia_type)) {
+                ret = posix_resolve_dirgfid_to_path (inode->gfid,
+                                                     priv->base_path,
+                                                     NULL, &path);
+                if (ret < 0) {
+                        ret = -1;
+                        goto err;
+                }
+                ret = dict_set_dynstr (dict, GFID2PATH_VIRT_XATTR_KEY, path);
+                if (ret < 0) {
+                        gf_msg (this->name, GF_LOG_WARNING, -ret,
+                                P_MSG_DICT_SET_FAILED, "could not set "
+                                "value for key (%s)", GFID2PATH_VIRT_XATTR_KEY);
+                        goto err;
+                }
+        } else {
+                have_val = _gf_false;
+                memset (value_buf, '\0', sizeof(value_buf));
+                size = sys_llistxattr (real_path, value_buf,
+                                       sizeof (value_buf) - 1);
+                if (size > 0) {
+                        have_val = _gf_true;
+                } else {
+                        if (errno == ERANGE) {
+                                gf_msg (this->name, GF_LOG_DEBUG, errno,
+                                        P_MSG_XATTR_FAILED,
+                                        "listxattr failed due to overflow of"
+                                        " buffer on %s ", real_path);
+                                size = sys_llistxattr (real_path, NULL, 0);
+                        }
+                        if (size == -1) {
+                                *op_errno = errno;
+                                if ((errno == ENOTSUP) || (errno == ENOSYS)) {
+                                        GF_LOG_OCCASIONALLY (
+                                            gf_posix_xattr_enotsup_log,
+                                            this->name, GF_LOG_WARNING,
+                                            "Extended attributes not "
+                                            "supported (try remounting"
+                                            " brick with 'user_xattr' "
+                                            "flag)");
+                                } else {
+                                        gf_msg (this->name, GF_LOG_ERROR, errno,
+                                                P_MSG_XATTR_FAILED,
+                                                "listxattr failed on %s",
+                                                real_path);
+                                }
+                                goto err;
+                        }
+                        if (size == 0)
+                                goto done;
+                }
+                list = alloca (size);
+                if (!list) {
+                        *op_errno = errno;
+                        goto err;
+                }
+                if (have_val) {
+                        memcpy (list, value_buf, size);
+                } else {
+                        size = sys_llistxattr (real_path, list, size);
+                        if (size < 0) {
+                                ret = -1;
+                                *op_errno = errno;
+                                goto err;
+                        }
+                }
+                remaining_size = size;
+                list_offset = 0;
+                while (remaining_size > 0) {
+                        strncpy (keybuffer, list + list_offset,
+                                 sizeof(keybuffer));
+
+                        if (!posix_is_gfid2path_xattr (keybuffer)) {
+                                goto ignore;
+                        }
+
+                        memset (value_buf, '\0', sizeof(value_buf));
+                        size = sys_lgetxattr (real_path, keybuffer, value_buf,
+                                              sizeof (value_buf) - 1);
+                        if (size == -1) {
+                                ret = -1;
+                                *op_errno = errno;
+                                gf_msg (this->name, GF_LOG_ERROR, errno,
+                                        P_MSG_XATTR_FAILED, "getxattr failed on"
+                                        " %s: key = %s ", real_path, keybuffer);
+                                break;
+                        }
+
+                        /* Parse pargfid from xattr value*/
+                        strncpy (pargfid_str, value_buf, 36);
+                        pargfid_str[36] = '\0';
+                        gf_uuid_parse (pargfid_str, pargfid);
+
+                        /* Convert pargfid to path */
+                        ret = posix_resolve_dirgfid_to_path (pargfid,
+                                                             priv->base_path,
+                                                             &value_buf[37],
+                                                             &paths[i]);
+                        i++;
+
+ignore:
+                        remaining_size -= strlen (keybuffer) + 1;
+                        list_offset += strlen (keybuffer) + 1;
+                } /* while (remaining_size > 0) */
+
+                /* Calculate memory to be allocated */
+                for (j = 0; j < i; j++) {
+                        bytes += strlen(paths[j]);
+                        if (j < i-1)
+                                bytes += strlen(priv->gfid2path_sep);
+                }
+                value = GF_CALLOC (bytes + 1, sizeof(char), gf_posix_mt_char);
+                if (!value) {
+                        ret = -1;
+                        *op_errno = errno;
+                        goto err;
+                }
+
+                for (j = 0; j < i; j++) {
+                        strcat (value, paths[j]);
+                        if (j != i - 1)
+                                strcat (value, priv->gfid2path_sep);
+                }
+                value[bytes] = '\0';
+
+                ret = dict_set_dynptr (dict, GFID2PATH_VIRT_XATTR_KEY,
+                                       value, bytes);
+                if (ret < 0) {
+                        *op_errno = -ret;
+                        gf_msg (this->name, GF_LOG_ERROR, *op_errno,
+                                P_MSG_DICT_SET_FAILED, "dict set operation "
+                                "on %s for the key %s failed.",
+                                real_path, GFID2PATH_VIRT_XATTR_KEY);
+                        GF_FREE (value);
+                        goto err;
+                }
+        }
+
+done:
+        for (j = 0; j < i; j++) {
+                if (paths[j])
+                        GF_FREE (paths[j]);
+        }
+        ret = 0;
+        return ret;
+err:
+        if (path)
+                GF_FREE (path);
+        for (j = 0; j < i; j++) {
+                if (paths[j])
+                        GF_FREE (paths[j]);
+        }
+        return ret;
+}
diff --git a/xlators/storage/posix/src/posix-gfid-path.h b/xlators/storage/posix/src/posix-gfid-path.h
index b1a23752e8f..ac3d03e14c3 100644
--- a/xlators/storage/posix/src/posix-gfid-path.h
+++ b/xlators/storage/posix/src/posix-gfid-path.h
@@ -24,4 +24,7 @@ posix_remove_gfid2path_xattr (xlator_t *, const char *, uuid_t,
                               const char *);
 gf_boolean_t
 posix_is_gfid2path_xattr (const char *name);
+int32_t
+posix_get_gfid2path (xlator_t *this, inode_t *inode, const char *real_path,
+                     int *op_errno, dict_t *dict);
 #endif /* _POSIX_GFID_PATH_H */
diff --git a/xlators/storage/posix/src/posix-helpers.c b/xlators/storage/posix/src/posix-helpers.c
index 1530b1192c5..1a05d65161a 100644
--- a/xlators/storage/posix/src/posix-helpers.c
+++ b/xlators/storage/posix/src/posix-helpers.c
@@ -2379,6 +2379,102 @@ posix_fdget_objectsignature (int fd, dict_t *xattr)
         return -EINVAL;
 }
 
+/*
+ * posix_resolve_dirgfid_to_path:
+ *       It converts given dirgfid to path by doing recursive readlinks at the
+ *  backend. If bname is given, it suffixes bname to dir path to form the
+ *  complete path else it doesn't. It allocates memory for the path and is
+ *  caller's responsibility to free the same. If bname is NULL and pargfid
+ *  is ROOT, then it returns "/"
+ **/
+
+int32_t
+posix_resolve_dirgfid_to_path (const uuid_t dirgfid, const char *brick_path,
+                               const char *bname, char **path)
+{
+        char             *linkname                  = NULL;
+        char             *dir_handle                = NULL;
+        char             *pgfidstr                  = NULL;
+        char             *saveptr                   = NULL;
+        ssize_t           len                       = 0;
+        int               ret                       = 0;
+        uuid_t            tmp_gfid                  = {0, };
+        uuid_t            pargfid                   = {0, };
+        char              gpath[PATH_MAX]           = {0,};
+        char              result[PATH_MAX]          = {0,};
+        char              result1[PATH_MAX]         = {0,};
+        char             *dir_name                  = NULL;
+        char              pre_dir_name[PATH_MAX]    = {0,};
+        xlator_t         *this                      = NULL;
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        gf_uuid_copy (pargfid, dirgfid);
+        if (!path || gf_uuid_is_null (pargfid)) {
+                ret = -1;
+                goto out;
+        }
+
+        if (__is_root_gfid (pargfid)) {
+                if (bname) {
+                        snprintf (result, PATH_MAX, "/%s", bname);
+                        *path = gf_strdup (result);
+                } else {
+                        *path = gf_strdup ("/");
+                }
+                return ret;
+        }
+
+        dir_handle = alloca (PATH_MAX);
+        linkname   = alloca (PATH_MAX);
+        (void) snprintf (gpath, PATH_MAX, "%s/.glusterfs/", brick_path);
+
+        while (!(__is_root_gfid (pargfid))) {
+                snprintf (dir_handle, PATH_MAX, "%s/%02x/%02x/%s", gpath,
+                          pargfid[0], pargfid[1], uuid_utoa (pargfid));
+
+                len = sys_readlink (dir_handle, linkname, PATH_MAX);
+                if (len < 0) {
+                        gf_msg (this->name, GF_LOG_ERROR, errno,
+                                P_MSG_READLINK_FAILED,
+                                "could not read the "
+                                "link from the gfid handle %s", dir_handle);
+                        ret = -1;
+                        goto out;
+                }
+
+                linkname[len] = '\0';
+
+                pgfidstr = strtok_r (linkname + strlen("../../00/00/"), "/",
+                                     &saveptr);
+                dir_name = strtok_r (NULL, "/", &saveptr);
+
+                if (strlen(pre_dir_name) != 0) { /* Remove '/' at the end */
+                        snprintf (result, PATH_MAX, "%s/%s", dir_name,
+                                  pre_dir_name);
+                } else {
+                        snprintf (result, PATH_MAX, "%s", dir_name);
+                }
+
+                strncpy (pre_dir_name, result, sizeof(pre_dir_name));
+
+                gf_uuid_parse (pgfidstr, tmp_gfid);
+                gf_uuid_copy (pargfid, tmp_gfid);
+        }
+
+        if (bname) {
+                snprintf (result1, PATH_MAX, "/%s/%s", result, bname);
+        } else {
+                snprintf (result1, PATH_MAX, "/%s", result);
+        }
+
+        *path = gf_strdup (result1);
+
+out:
+        return ret;
+}
+
 posix_inode_ctx_t *
 __posix_inode_ctx_get (inode_t *inode, xlator_t *this)
 {
diff --git a/xlators/storage/posix/src/posix.c b/xlators/storage/posix/src/posix.c
index ceb9d448025..a04d830dd69 100644
--- a/xlators/storage/posix/src/posix.c
+++ b/xlators/storage/posix/src/posix.c
@@ -4734,6 +4734,23 @@ posix_getxattr (call_frame_t *frame, xlator_t *this,
                 goto done;
         }
 
+        if (loc->inode && name &&
+            (strcmp (name, GFID2PATH_VIRT_XATTR_KEY) == 0)) {
+                if (!priv->gfid2path) {
+                        op_errno = ENOATTR;
+                        op_ret = -1;
+                        goto out;
+                }
+                ret = posix_get_gfid2path (this, loc->inode, real_path,
+                                           &op_errno, dict);
+                if (ret < 0) {
+                        op_ret = -1;
+                        goto out;
+                }
+                size = ret;
+                goto done;
+        }
+
         if (loc->inode && name
             && (strcmp (name, GET_ANCESTRY_PATH_KEY) == 0)) {
                 int type = POSIX_ANCESTRY_PATH;
@@ -6989,7 +7006,19 @@ posix_set_owner (xlator_t *this, uid_t uid, gid_t gid)
 
         return ret;
 }
+static int
+set_gfid2path_separator (struct posix_private *priv, const char *str)
+{
+        int str_len = 0;
 
+        str_len = strlen(str);
+        if (str_len > 0 && str_len < 8) {
+                strcpy (priv->gfid2path_sep, str);
+                return 0;
+        }
+
+        return -1;
+}
 
 static int
 set_batch_fsync_mode (struct posix_private *priv, const char *str)
@@ -7036,6 +7065,7 @@ reconfigure (xlator_t *this, dict_t *options)
         int32_t               uid = -1;
         int32_t               gid = -1;
 	char                 *batch_fsync_mode_str = NULL;
+    char                 *gfid2path_sep = NULL;
 
 	priv = this->private;
 
@@ -7056,6 +7086,14 @@ reconfigure (xlator_t *this, dict_t *options)
 		goto out;
 	}
 
+	GF_OPTION_RECONF ("gfid2path-separator", gfid2path_sep, options,
+                          str, out);
+        if (set_gfid2path_separator (priv, gfid2path_sep) != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, 0, P_MSG_INVALID_ARGUMENT,
+                        "Length of separator exceeds 7: %s", gfid2path_sep);
+                goto out;
+        }
+
 #ifdef GF_DARWIN_HOST_OS
 
         char   *xattr_user_namespace_mode_str = NULL;
@@ -7249,6 +7287,7 @@ init (xlator_t *this)
         int32_t               uid           = -1;
         int32_t               gid           = -1;
 	char                 *batch_fsync_mode_str;
+        char                 *gfid2path_sep = NULL;
 
         dir_data = dict_get (this->options, "directory");
 
@@ -7750,6 +7789,13 @@ init (xlator_t *this)
 		goto out;
 	}
 
+	GF_OPTION_INIT ("gfid2path-separator", gfid2path_sep, str, out);
+        if (set_gfid2path_separator (_private, gfid2path_sep) != 0) {
+                gf_msg (this->name, GF_LOG_ERROR, 0, P_MSG_INVALID_ARGUMENT,
+                        "Length of separator exceeds 7: %s", gfid2path_sep);
+                goto out;
+        }
+
 #ifdef GF_DARWIN_HOST_OS
 
         char  *xattr_user_namespace_mode_str = NULL;
@@ -7950,6 +7996,11 @@ struct volume_options options[] = {
           .default_value = "off",
           .description = "Enable logging metadata for gfid to path conversion"
         },
+        { .key = {"gfid2path-separator"},
+          .type = GF_OPTION_TYPE_STR,
+          .default_value = ":",
+          .description = "Path separator for glusterfs.gfidtopath virt xattr"
+        },
 #if GF_DARWIN_HOST_OS
         { .key = {"xattr-user-namespace-mode"},
           .type = GF_OPTION_TYPE_STR,
diff --git a/xlators/storage/posix/src/posix.h b/xlators/storage/posix/src/posix.h
index d83e2678572..3893cf3f9bd 100644
--- a/xlators/storage/posix/src/posix.h
+++ b/xlators/storage/posix/src/posix.h
@@ -173,6 +173,7 @@ struct posix_private {
 	uint32_t        batch_fsync_delay_usec;
         gf_boolean_t    update_pgfid_nlinks;
         gf_boolean_t    gfid2path;
+        char            gfid2path_sep[8];
 
         /* seconds to sleep between health checks */
         uint32_t        health_check_interval;
@@ -300,6 +301,9 @@ posix_get_ancestry (xlator_t *this, inode_t *leaf_inode,
                     dict_t *xdata);
 int
 posix_handle_georep_xattrs (call_frame_t *, const char *, int *, gf_boolean_t);
+int32_t
+posix_resolve_dirgfid_to_path (const uuid_t dirgfid, const char *brick_path,
+                               const char *bname, char **path);
 void
 posix_gfid_unset (xlator_t *this, dict_t *xdata);