afr/stripe: collect pathinfo xattrs from all childs

Signed-off-by: Venky Shankar <venky@gluster.com>
Signed-off-by: Anand Avati <avati@gluster.com>

BUG: 3046 (getxattr for afr should returns realpath from all childs)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=3046

7 files changed, 357 insertions, 17 deletions

diff --git a/xlators/cluster/afr/src/afr-inode-read.c b/xlators/cluster/afr/src/afr-inode-read.c
index 431b734753c..742d3687ceb 100644
--- a/xlators/cluster/afr/src/afr-inode-read.c
+++ b/xlators/cluster/afr/src/afr-inode-read.c
@@ -654,6 +654,110 @@ afr_getxattr_unwind (call_frame_t *frame,
 }
 
 int32_t
+afr_getxattr_pathinfo_cbk (call_frame_t *frame, void *cookie,
+                           xlator_t *this, int32_t op_ret, int32_t op_errno,
+                           dict_t *dict)
+{
+        afr_local_t *local             = NULL;
+        int32_t      callcnt           = 0;
+        int          ret               = 0;
+        char        *pathinfo          = NULL;
+        char        *pathinfo_serz     = NULL;
+        char        pathinfo_cky[1024] = {0,};
+        dict_t      *xattr             = NULL;
+        long         cky               = 0;
+        int32_t      padding           = 0;
+        int32_t      tlen              = 0;
+
+        if (!frame || !frame->local || !this) {
+                gf_log (this->name, GF_LOG_ERROR, "possible NULL deref");
+                goto out;
+        }
+
+        local = frame->local;
+        cky = (long) cookie;
+
+        LOCK (&frame->lock);
+                {
+                        callcnt = --local->call_count;
+
+                        if (!dict || (op_ret < 0))
+                                goto out;
+
+                        if (!local->dict)
+                                local->dict = dict_new ();
+
+                        if (local->dict) {
+                                ret = dict_get_str (dict, GF_XATTR_PATHINFO_KEY, &pathinfo);
+                                if (ret)
+                                        goto out;
+
+                                pathinfo = gf_strdup (pathinfo);
+
+                                snprintf (pathinfo_cky, 1024, "%s-%ld", GF_XATTR_PATHINFO_KEY, cky);
+                                ret = dict_set_dynstr (local->dict, pathinfo_cky, pathinfo);
+                                if (ret) {
+                                        gf_log (this->name, GF_LOG_ERROR, "Cannot set pathinfo cookie key");
+                                        goto out;
+                                }
+
+                                local->cont.getxattr.pathinfo_len += strlen (pathinfo) + 1;
+                        }
+                }
+ out:
+        UNLOCK (&frame->lock);
+
+        if (!callcnt) {
+                if (!local->cont.getxattr.pathinfo_len)
+                        goto unwind;
+
+                xattr = dict_new ();
+                if (!xattr)
+                        goto unwind;
+
+                /* extra bytes for decorations (brackets and <>'s) */
+                padding = strlen (this->name) + strlen (AFR_PATHINFO_HEADER) + 4;
+                local->cont.getxattr.pathinfo_len += (padding + 2);
+
+                pathinfo_serz = GF_CALLOC (local->cont.getxattr.pathinfo_len, sizeof (char),
+                                           gf_common_mt_char);
+
+                if (!pathinfo_serz)
+                        goto unwind;
+
+                /* the xlator info */
+                sprintf (pathinfo_serz, "(<"AFR_PATHINFO_HEADER"%s> ", this->name);
+
+                /* actual series of pathinfo */
+                ret = dict_serialize_value_with_delim (local->dict, pathinfo_serz + strlen (pathinfo_serz),
+                                                       &tlen, ' ');
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Error serializing dictionary");
+                        goto unwind;
+                }
+
+                /* closing part */
+                *(pathinfo_serz + padding + tlen) = ')';
+                *(pathinfo_serz + padding + tlen + 1) = '\0';
+
+                ret = dict_set_dynstr (xattr, GF_XATTR_PATHINFO_KEY, pathinfo_serz);
+                if (ret)
+                        gf_log (this->name, GF_LOG_ERROR, "Cannot set pathinfo key in dict");
+
+        unwind:
+                AFR_STACK_UNWIND (getxattr, frame, op_ret, op_errno, xattr);
+
+                if (local->dict)
+                        dict_unref (local->dict);
+
+                if (xattr)
+                        dict_unref (xattr);
+        }
+
+        return ret;
+}
+
+int32_t
 afr_getxattr (call_frame_t *frame, xlator_t *this,
               loc_t *loc, const char *name)
 {
@@ -683,7 +787,7 @@ afr_getxattr (call_frame_t *frame, xlator_t *this,
 
         loc_copy (&local->loc, loc);
         if (name)
-                local->cont.getxattr.name       = gf_strdup (name);
+                local->cont.getxattr.name = gf_strdup (name);
 
 
         if (name) {
@@ -724,6 +828,24 @@ afr_getxattr (call_frame_t *frame, xlator_t *this,
                         return 0;
                 }
 
+                /*
+                 * if we are doing getxattr with pathinfo as the key then we
+                 * collect information from all childs
+                 */
+                if (strncmp (name, GF_XATTR_PATHINFO_KEY,
+                             strlen (GF_XATTR_PATHINFO_KEY)) == 0) {
+
+                        local->call_count = priv->child_count;
+                        for (i = 0; i < priv->child_count; i++) {
+                                STACK_WIND_COOKIE (frame, afr_getxattr_pathinfo_cbk,
+                                                   (void *) (long) i,
+                                                   children[i], children[i]->fops->getxattr,
+                                                   loc, name);
+                        }
+
+                        return 0;
+                }
+
                 if (*priv->vol_uuid) {
                         if ((match_uuid_local (name, priv->vol_uuid) == 0)
                             && (-1 == frame->root->pid)) {
diff --git a/xlators/cluster/afr/src/afr.h b/xlators/cluster/afr/src/afr.h
index 973e8b58fed..abfaaac0e7b 100644
--- a/xlators/cluster/afr/src/afr.h
+++ b/xlators/cluster/afr/src/afr.h
@@ -33,6 +33,7 @@
 #include "libxlator.h"
 
 #define AFR_XATTR_PREFIX "trusted.afr"
+#define AFR_PATHINFO_HEADER "REPLICATE:"
 
 struct _pump_private;
 
@@ -395,6 +396,7 @@ typedef struct _afr_local {
                 struct {
                         char *name;
                         int last_tried;
+                        long pathinfo_len;
                 } getxattr;
 
                 struct {
diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 18bb65d3098..c07de138029 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -1680,28 +1680,48 @@ int
 dht_pathinfo_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                            int op_ret, int op_errno, dict_t *xattr)
 {
-        dht_local_t  *local = NULL;
-        int ret = 0;
-        int flag = 0;
-        int this_call_cnt = 0;
-        char *value_got = NULL;
-        char  layout_buf[8192] = {0,};
-        char  xattr_buf[8192 + 1024] = {0,};
-        dict_t *dict = NULL;
+        dht_local_t *local         = NULL;
+        int          ret           = 0;
+        int          flag          = 0;
+        int          this_call_cnt = 0;
+        char        *value_got     = NULL;
+        char  layout_buf[8192]     = {0,};
+        char        *xattr_buf     = NULL;
+        dict_t      *dict          = NULL;
+        int32_t      alloc_len     = 0;
+        int32_t      plen          = 0;
 
         local = frame->local;
 
         if (op_ret != -1) {
                 ret = dict_get_str (xattr, GF_XATTR_PATHINFO_KEY, &value_got);
                 if (!ret) {
+                        alloc_len = strlen (value_got);
+
+                        /**
+                         * allocate the buffer:- we allocate 10 bytes extra in case we need to
+                         * append ' Link: ' in the buffer for another STACK_WIND
+                         */
                         if (!local->pathinfo)
-                                local->pathinfo = GF_CALLOC (8192, sizeof (char),
-                                                             gf_common_mt_char);
-                        if (local->pathinfo)
+                                local->pathinfo = GF_CALLOC (alloc_len + strlen (DHT_PATHINFO_HEADER) + 10,
+                                                             sizeof (char), gf_common_mt_char);
+
+                        if (local->pathinfo) {
+                                plen = strlen (local->pathinfo);
+                                if (plen) {
+                                        alloc_len += plen;
+                                        local->pathinfo = GF_REALLOC (local->pathinfo,
+                                                                      alloc_len);
+                                        if (!local->pathinfo)
+                                                goto out;
+                                }
+
                                 strcat (local->pathinfo, value_got);
+                        }
                 }
         }
 
+ out:
         this_call_cnt = dht_frame_return (frame);
         if (is_last_call (this_call_cnt)) {
                 if (local->layout->cnt > 1) {
@@ -1712,19 +1732,23 @@ dht_pathinfo_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
                 dict = dict_new ();
 
+                /* we would need max-to-max this many bytes to create pathinfo string */
+                alloc_len += (2 * strlen (this->name)) + strlen (layout_buf) + 40;
+                xattr_buf = GF_CALLOC (alloc_len, sizeof (char), gf_common_mt_char);
+
                 if (flag && local->pathinfo)
-                        snprintf (xattr_buf, 9216, "((%s %s) (%s-layout %s))",
+                        snprintf (xattr_buf, alloc_len, "((<"DHT_PATHINFO_HEADER"%s> %s) (%s-layout %s))",
                                   this->name, local->pathinfo, this->name,
                                   layout_buf);
                 else if (local->pathinfo)
-                        snprintf (xattr_buf, 9216, "(%s %s)",
+                        snprintf (xattr_buf, alloc_len, "(<"DHT_PATHINFO_HEADER"%s> %s)",
                                   this->name, local->pathinfo);
                 else if (flag)
-                        snprintf (xattr_buf, 9216, "(%s-layout %s)",
+                        snprintf (xattr_buf, alloc_len, "(%s-layout %s)",
                                   this->name, layout_buf);
 
-                ret = dict_set_str (dict, GF_XATTR_PATHINFO_KEY,
-                                    xattr_buf);
+                ret = dict_set_dynstr (dict, GF_XATTR_PATHINFO_KEY,
+                                       xattr_buf);
 
                 if (local->pathinfo)
                         GF_FREE (local->pathinfo);
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index 4b5fff04ccb..461add16897 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -31,6 +31,7 @@
 #define GF_XATTR_FIX_LAYOUT_KEY   "trusted.distribute.fix.layout"
 #define GF_DHT_LOOKUP_UNHASHED_ON   1
 #define GF_DHT_LOOKUP_UNHASHED_AUTO 2
+#define DHT_PATHINFO_HEADER "DISTRIBUTE:"
 
 #include <fnmatch.h>
 
diff --git a/xlators/cluster/stripe/src/stripe-mem-types.h b/xlators/cluster/stripe/src/stripe-mem-types.h
index 2d2757e86a8..c376a1fc93f 100644
--- a/xlators/cluster/stripe/src/stripe-mem-types.h
+++ b/xlators/cluster/stripe/src/stripe-mem-types.h
@@ -34,6 +34,7 @@ enum gf_stripe_mem_types_ {
         gf_stripe_mt_xlator_t,
         gf_stripe_mt_stripe_private_t,
         gf_stripe_mt_stripe_options,
+        gf_stripe_mt_xattr_sort_t,
         gf_stripe_mt_end
 };
 #endif
diff --git a/xlators/cluster/stripe/src/stripe.c b/xlators/cluster/stripe/src/stripe.c
index 07c77c9dfc7..39047196120 100644
--- a/xlators/cluster/stripe/src/stripe.c
+++ b/xlators/cluster/stripe/src/stripe.c
@@ -4246,6 +4246,166 @@ out:
         return 0;
 }
 
+int32_t
+stripe_pathinfo_aggregate (char *buffer, stripe_local_t *local, int32_t *total)
+{
+        int32_t              i     = 0;
+        int32_t              ret   = -1;
+        int32_t              len   = 0;
+        char                *sbuf  = NULL;
+        stripe_xattr_sort_t *xattr = NULL;
+
+        if (!buffer || !local || !local->xattr_list)
+                goto out;
+
+        sbuf = buffer;
+
+        for (i = 0; i < local->nallocs; i++) {
+                xattr = local->xattr_list + i;
+                len = xattr->pathinfo_len;
+
+                if (len && xattr && xattr->pathinfo) {
+                        memcpy (buffer, xattr->pathinfo, len);
+                        buffer += len;
+                        *buffer++ = ' ';
+                }
+        }
+
+        *--buffer = '\0';
+        if (total)
+                *total = buffer - sbuf;
+        ret = 0;
+
+ out:
+        return ret;
+}
+
+int32_t
+stripe_free_pathinfo_str (stripe_local_t *local)
+{
+        int32_t              i     = 0;
+        int32_t              ret   = -1;
+        stripe_xattr_sort_t *xattr = NULL;
+
+        if (!local || !local->xattr_list)
+                goto out;
+
+        for (i = 0; i < local->nallocs; i++) {
+                xattr = local->xattr_list + i;
+
+                if (xattr && xattr->pathinfo)
+                        GF_FREE (xattr->pathinfo);
+        }
+
+        ret = 0;
+ out:
+        return ret;
+}
+
+int32_t
+stripe_getxattr_pathinfo_cbk (call_frame_t *frame, void *cookie,
+                             xlator_t *this, int32_t op_ret, int32_t op_errno,
+                             dict_t *dict) {
+        stripe_local_t      *local         = NULL;
+        int32_t              callcnt       = 0;
+        int32_t              ret           = -1;
+        long                 cky           = 0;
+        char                *pathinfo      = NULL;
+        char                *pathinfo_serz = NULL;
+        int32_t              padding       = 0;
+        int32_t              tlen          = 0;
+        char stripe_size_str[20]           = {0,};
+        stripe_xattr_sort_t *xattr         = NULL;
+        dict_t              *stripe_xattr  = NULL;
+
+        if (!frame || !frame->local || !this) {
+                gf_log (this->name, GF_LOG_ERROR, "Possible NULL deref");
+                return ret;
+        }
+
+        local = frame->local;
+        cky = (long) cookie;
+
+        LOCK (&frame->lock);
+        {
+                callcnt = --local->wind_count;
+
+                if (!dict || (op_ret < 0))
+                        goto out;
+
+                if (!local->xattr_list)
+                        local->xattr_list = (stripe_xattr_sort_t *) GF_CALLOC (local->nallocs,
+                                                                               sizeof (stripe_xattr_sort_t),
+                                                                               gf_stripe_mt_xattr_sort_t);
+
+                if (local->xattr_list) {
+                        ret = dict_get_str (dict, GF_XATTR_PATHINFO_KEY, &pathinfo);
+                        if (ret)
+                                goto out;
+
+                        xattr = local->xattr_list + (int32_t) cky;
+
+                        pathinfo = gf_strdup (pathinfo);
+                        xattr->pos = cky;
+                        xattr->pathinfo = pathinfo;
+                        xattr->pathinfo_len = strlen (pathinfo);
+
+                        local->xattr_total_len += strlen (pathinfo) + 1;
+                }
+        }
+ out:
+        UNLOCK (&frame->lock);
+
+        if (!callcnt) {
+                if (!local->xattr_total_len)
+                        goto unwind;
+
+                stripe_xattr = dict_new ();
+                if (!stripe_xattr)
+                        goto unwind;
+
+                snprintf (stripe_size_str, 20, "%ld", local->stripe_size);
+
+                /* extra bytes for decorations (brackets and <>'s) */
+                padding = strlen (this->name) + strlen (STRIPE_PATHINFO_HEADER)
+                        + strlen (stripe_size_str) + 7;
+                local->xattr_total_len += (padding + 2);
+
+                pathinfo_serz = GF_CALLOC (local->xattr_total_len, sizeof (char),
+                                           gf_common_mt_char);
+                if (!pathinfo_serz)
+                        goto unwind;
+
+                /* xlator info */
+                sprintf (pathinfo_serz, "(<"STRIPE_PATHINFO_HEADER"%s:[%s]> ", this->name, stripe_size_str);
+
+                ret = stripe_pathinfo_aggregate (pathinfo_serz + padding, local, &tlen);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Cannot aggregate pathinfo list");
+                        goto unwind;
+                }
+
+                *(pathinfo_serz + padding + tlen) = ')';
+                *(pathinfo_serz + padding + tlen + 1) = '\0';
+
+                ret = dict_set_dynstr (stripe_xattr, GF_XATTR_PATHINFO_KEY, pathinfo_serz);
+                if (ret)
+                        gf_log (this->name, GF_LOG_ERROR, "Cannot set pathinfo key in dict");
+
+        unwind:
+                STRIPE_STACK_UNWIND (getxattr, frame, op_ret, op_errno, stripe_xattr);
+
+                ret = stripe_free_pathinfo_str (local);
+
+                if (local->xattr_list)
+                        GF_FREE (local->xattr_list);
+
+                if (stripe_xattr)
+                        dict_unref (stripe_xattr);
+        }
+
+        return ret;
+}
 
 int32_t
 stripe_getxattr (call_frame_t *frame, xlator_t *this,
@@ -4317,6 +4477,24 @@ stripe_getxattr (call_frame_t *frame, xlator_t *this,
                 return 0;
         }
 
+        if (name && (strncmp (name, GF_XATTR_PATHINFO_KEY,
+                              strlen (GF_XATTR_PATHINFO_KEY)) == 0)) {
+                local->stripe_size = stripe_get_matching_bs (loc->path,
+                                                             priv->pattern,
+                                                             priv->block_size);
+                local->nallocs = local->wind_count = priv->child_count;
+
+                for (i = 0, trav = this->children; i < priv->child_count; i++,
+                     trav = trav->next) {
+                        STACK_WIND_COOKIE (frame, stripe_getxattr_pathinfo_cbk,
+                                           (void *) (long) i, trav->xlator,
+                                           trav->xlator->fops->getxattr,
+                                           loc, name);
+                }
+
+                return 0;
+        }
+
         if (name &&(*priv->vol_uuid)) {
                 if ((match_uuid_local (name, priv->vol_uuid) == 0)
                     && (-1 == frame->root->pid)) {
diff --git a/xlators/cluster/stripe/src/stripe.h b/xlators/cluster/stripe/src/stripe.h
index 6934244a7a6..a6b9375b451 100644
--- a/xlators/cluster/stripe/src/stripe.h
+++ b/xlators/cluster/stripe/src/stripe.h
@@ -37,6 +37,8 @@
 #include <fnmatch.h>
 #include <signal.h>
 
+#define STRIPE_PATHINFO_HEADER "STRIPE:"
+
 
 #define STRIPE_STACK_UNWIND(fop, frame, params ...) do {           \
                 stripe_local_t *__local = NULL;                    \
@@ -62,6 +64,12 @@
                 }                                         \
         } while (0)
 
+typedef struct stripe_xattr_sort {
+        int32_t  pos;
+        int32_t  pathinfo_len;
+        char    *pathinfo;
+} stripe_xattr_sort_t;
+
 /**
  * struct stripe_options : This keeps the pattern and the block-size
  *     information, which is used for striping on a file.
@@ -165,6 +173,10 @@ struct stripe_local {
         /* For File I/O fops */
         dict_t              *dict;
 
+        stripe_xattr_sort_t *xattr_list;
+        int32_t              xattr_total_len;
+        int32_t              nallocs;
+
         struct marker_str    marker;
 
         /* General usage */