dht: heal directory path if the directory is not present

After a successful nameless lookup if the directory is not
present on any of the subvol, then we will get the path of
the directory and will recursively send a named lookp on
each parent directory.

This will help particularly for the scenarios like add brick
and attach-tier.

Change-Id: I64c2118a5ab03bbaa59b0dfc62babdf4472a92a3
BUG: 1272949
Signed-off-by: Mohammed Rafi KC <rkavunga@redhat.com>
Reviewed-on: http://review.gluster.org/12376
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Reviewed-by: N Balachandran <nbalacha@redhat.com>
Reviewed-by: Raghavendra G <rgowdapp@redhat.com>

4 files changed, 249 insertions, 10 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 79ba77bf16c..6af55d5d42e 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -212,17 +212,22 @@ out:
         return ret;
 }
 
-
 int
 dht_discover_complete (xlator_t *this, call_frame_t *discover_frame)
 {
-        dht_local_t     *local = NULL;
-        call_frame_t    *main_frame = NULL;
-        int              op_errno = 0;
-        int              ret = -1;
-        dht_layout_t    *layout = NULL;
-        dht_conf_t      *conf = NULL;
+        dht_local_t     *local           = NULL;
+        dht_local_t     *heal_local      = NULL;
+        call_frame_t    *main_frame      = NULL;
+        call_frame_t    *heal_frame      = NULL;
+        int              op_errno        = 0;
+        int              ret             = -1;
+        dht_layout_t    *layout          = NULL;
+        dht_conf_t      *conf            = NULL;
         uint32_t         vol_commit_hash = 0;
+        xlator_t        *source          = NULL;
+        int              heal_path       = 0;
+        int              i               = 0;
+        loc_t            loc             = {0 };
 
         local = discover_frame->local;
         layout = local->layout;
@@ -301,6 +306,63 @@ dht_discover_complete (xlator_t *this, call_frame_t *discover_frame)
                 }
         }
 
+        if (IA_ISDIR (local->stbuf.ia_type)) {
+                for (i = 0; i < layout->cnt; i++) {
+                       if (!source && !layout->list[i].err)
+                                source = layout->list[i].xlator;
+                        if (layout->list[i].err == ENOENT ||
+                            layout->list[i].err == ESTALE) {
+                                heal_path = 1;
+                        }
+                        if (source && heal_path)
+                                break;
+                }
+        }
+        if (source && heal_path) {
+                gf_uuid_copy (loc.gfid, local->gfid);
+                if (gf_uuid_is_null (loc.gfid)) {
+                        goto done;
+                }
+
+                if (local->inode)
+                        loc.inode = inode_ref (local->inode);
+                else
+                        goto done;
+
+               heal_frame = create_frame (this, this->ctx->pool);
+               if (heal_frame) {
+                        heal_local = dht_local_init (heal_frame, &loc,
+                                                     NULL, 0);
+                        if (!heal_local)
+                                goto cleanup;
+
+                        gf_uuid_copy (heal_local->gfid, local->gfid);
+                        heal_frame->cookie = source;
+                        heal_local->xattr = dict_ref (local->xattr);
+                        heal_local->stbuf = local->stbuf;
+                        heal_local->postparent = local->postparent;
+                        heal_local->inode = inode_ref (loc.inode);
+                        heal_local->main_frame = main_frame;
+                        FRAME_SU_DO (heal_frame, dht_local_t);
+                        ret = synctask_new (this->ctx->env,
+                                            dht_heal_full_path,
+                                            dht_heal_full_path_done,
+                                            heal_frame, heal_frame);
+                        if (!ret) {
+                                loc_wipe (&loc);
+                                return 0;
+                        }
+                        /*
+                         * Failed to spawn the synctask. Returning
+                         * with out doing heal.
+                         */
+cleanup:
+                        loc_wipe (&loc);
+                        DHT_STACK_DESTROY (heal_frame);
+                }
+
+        }
+done:
         DHT_STACK_UNWIND (lookup, main_frame, local->op_ret, local->op_errno,
                           local->inode, &local->stbuf, local->xattr,
                           &local->postparent);
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index 6483b2e86d7..392d2cf163a 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -1106,6 +1106,12 @@ int
 dht_layout_sort (dht_layout_t *layout);
 
 int
+dht_heal_full_path (void *data);
+
+int
+dht_heal_full_path_done (int op_ret, call_frame_t *frame, void *data);
+
+int
 dht_layout_missing_dirs (dht_layout_t *layout);
 
 int
diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index 6d00ce68909..49f087b3ec9 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -1917,3 +1917,174 @@ out:
 
         return -1;
 }
+inode_t*
+dht_heal_path (xlator_t *this, char *path, inode_table_t *itable)
+{
+        int             ret             = -1;
+        struct iatt     iatt            = {0, };
+        inode_t        *linked_inode    = NULL;
+        loc_t           loc             = {0, };
+        char           *bname           = NULL;
+        char           *save_ptr        = NULL;
+        uuid_t          gfid            = {0, };
+        char           *tmp_path        = NULL;
+
+
+        tmp_path = gf_strdup (path);
+        if (!tmp_path) {
+                goto out;
+        }
+
+        memset (gfid, 0, 16);
+        gfid[15] = 1;
+
+        gf_uuid_copy (loc.pargfid, gfid);
+        loc.parent = inode_ref (itable->root);
+
+        bname = strtok_r (tmp_path, "/",  &save_ptr);
+
+        /* sending a lookup on parent directory,
+         * Eg:  if  path is like /a/b/c/d/e/f/g/
+         * then we will send a lookup on a first and then b,c,d,etc
+         */
+
+        while (bname) {
+                linked_inode = NULL;
+                loc.inode = inode_grep (itable, loc.parent, bname);
+                if (loc.inode == NULL) {
+                        loc.inode = inode_new (itable);
+                        if (loc.inode == NULL) {
+                                ret = -ENOMEM;
+                                goto out;
+                        }
+                } else {
+                        /*
+                         * Inode is already populated in the inode table.
+                         * Which means we already looked up the inde and
+                         * linked with a dentry. So that we will skip
+                         * lookup on this entry, and proceed to next.
+                         */
+                        bname = strtok_r (NULL, "/",  &save_ptr);
+                        inode_unref (loc.parent);
+                        loc.parent = loc.inode;
+                        gf_uuid_copy (loc.pargfid, loc.inode->gfid);
+                        loc.inode = NULL;
+                        continue;
+                }
+
+                loc.name = bname;
+                ret = loc_path (&loc, bname);
+
+                ret = syncop_lookup (this, &loc, &iatt, NULL, NULL, NULL);
+                if (ret) {
+                        gf_msg (this->name, GF_LOG_INFO, -ret,
+                                DHT_MSG_DIR_SELFHEAL_FAILED,
+                                "Healing of path %s failed on subvolume %s for "
+                                "directory %s", path, this->name, bname);
+                        goto out;
+                }
+
+                linked_inode = inode_link (loc.inode, loc.parent, bname, &iatt);
+                if (!linked_inode)
+                        goto out;
+
+                loc_wipe (&loc);
+                gf_uuid_copy (loc.pargfid, linked_inode->gfid);
+                loc.inode = NULL;
+                loc.parent = linked_inode;
+
+                bname = strtok_r (NULL, "/",  &save_ptr);
+        }
+out:
+        inode_ref (linked_inode);
+        loc_wipe (&loc);
+        GF_FREE (tmp_path);
+
+        return linked_inode;
+}
+
+
+int
+dht_heal_full_path (void *data)
+{
+        call_frame_t            *heal_frame     = data;
+        dht_local_t             *local          = NULL;
+        loc_t                    loc            = {0, };
+        dict_t                  *dict           = NULL;
+        char                    *path           = NULL;
+        int                      ret            = -1;
+        xlator_t                *source         = NULL;
+        xlator_t                *this           = NULL;
+        inode_table_t           *itable         = NULL;
+        inode_t                 *inode          = NULL;
+        inode_t                 *tmp_inode      = NULL;
+
+        GF_VALIDATE_OR_GOTO ("DHT", heal_frame, out);
+
+        local = heal_frame->local;
+        this = heal_frame->this;
+        source = heal_frame->cookie;
+        heal_frame->cookie = NULL;
+        gf_uuid_copy (loc.gfid, local->gfid);
+
+        if (local->loc.inode)
+                loc.inode = inode_ref (local->loc.inode);
+        else
+                goto out;
+
+        itable = loc.inode->table;
+        ret = syncop_getxattr (source, &loc, &dict,
+                       GET_ANCESTRY_PATH_KEY, NULL, NULL);
+        if (ret) {
+                gf_msg (this->name, GF_LOG_INFO, -ret,
+                        DHT_MSG_DIR_SELFHEAL_FAILED,
+                        "Failed to get path from subvol %s. Aborting "
+                        "directory healing.", source->name);
+                goto out;
+        }
+
+        ret = dict_get_str (dict, GET_ANCESTRY_PATH_KEY, &path);
+        if (path) {
+                inode = dht_heal_path (this, path, itable);
+                if (inode && inode != local->inode) {
+                        /*
+                         * if inode returned by heal function is different
+                         * from what we passed, which means a racing thread
+                         * already linked a different inode for dentry.
+                         * So we will update our local->inode, so that we can
+                         * retrurn proper inode.
+                         */
+                        tmp_inode = local->inode;
+                        local->inode = inode;
+                        inode_unref (tmp_inode);
+                        tmp_inode = NULL;
+                } else {
+                        inode_unref (inode);
+                }
+        }
+
+out:
+        loc_wipe (&loc);
+        if (dict)
+                dict_unref (dict);
+        return 0;
+}
+
+int
+dht_heal_full_path_done (int op_ret, call_frame_t *heal_frame, void *data)
+{
+
+        call_frame_t            *main_frame       = NULL;
+        dht_local_t             *local            = NULL;
+
+        local = heal_frame->local;
+        main_frame = local->main_frame;
+        local->main_frame = NULL;
+
+        DHT_STACK_UNWIND (lookup, main_frame, 0, 0,
+                          local->inode, &local->stbuf, local->xattr,
+                          &local->postparent);
+
+        DHT_STACK_DESTROY (heal_frame);
+        return 0;
+}
diff --git a/xlators/storage/posix/src/posix.c b/xlators/storage/posix/src/posix.c
index 0cc4f63ae5f..e2fe081f86d 100644
--- a/xlators/storage/posix/src/posix.c
+++ b/xlators/storage/posix/src/posix.c
@@ -3802,14 +3802,14 @@ posix_get_ancestry (xlator_t *this, inode_t *leaf_inode,
 
         priv = this->private;
 
-        if (!priv->update_pgfid_nlinks)
-                goto out;
-
         if (IA_ISDIR (leaf_inode->ia_type)) {
                 ret = posix_get_ancestry_directory (this, leaf_inode,
                                                     head, path, type, op_errno,
                                                     xdata);
         } else  {
+
+                if (!priv->update_pgfid_nlinks)
+                        goto out;
                 ret = posix_get_ancestry_non_directory (this, leaf_inode,
                                                         head, path, type,
                                                         op_errno, xdata);