cluster/afr: [Feature] Command implementation to get heal-count

Currently to know the number of files to be healed, either user
has to go to backend and check the number of entries present in
indices/xattrop directory. But if a volume consists of large
number of bricks, going to each backend and counting the number
of entries is a time-taking task. Otherwise user can give
gluster volume heal vol-name info command but with this
approach if no. of entries are very hugh in the indices/
xattrop directory, it will comsume time.

So as a feature, new command is implemented.

Command 1: gluster volume heal vn statistics heal-count
This command will get the number of entries present in
every brick of a volume. The output displays only entries
count.

Command 2: gluster volume heal vn statistics heal-count
           replica 192.168.122.1:/home/user/brickname

           Here if we are concerned with just one replica.
So providing any one of the brick of a replica will get
the number of entries to be healed for that replica only.

Example:
Replicate volume with replica count 2.

Backend status:
--------------
[root@dhcp-0-17 xattrop]# ls -lia | wc -l
1918

NOTE: Out of 1918, 2 entries are <xattrop-gfid> dummy
entries so actual no. of entries to be healed are
1916.

[root@dhcp-0-17 xattrop]# pwd
/home/user/2ty/.glusterfs/indices/xattrop

Command output:
--------------
Gathering count of entries to be healed on volume volume3 has been successful

Brick 192.168.122.1:/home/user/22iu
Status: Brick is Not connected
Entries count is not available

Brick 192.168.122.1:/home/user/2ty
Number of entries: 1916

Change-Id: I72452f3de50502dc898076ec74d434d9e77fd290
BUG: 1015990
Signed-off-by: Venkatesh Somyajulu <vsomyaju@redhat.com>
Reviewed-on: http://review.gluster.org/6044
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

2 files changed, 316 insertions, 13 deletions

diff --git a/xlators/features/index/src/index.c b/xlators/features/index/src/index.c
index 259e8ec01de..9253120f3f2 100644
--- a/xlators/features/index/src/index.c
+++ b/xlators/features/index/src/index.c
@@ -15,8 +15,10 @@
 #include "index.h"
 #include "options.h"
 #include "glusterfs3-xdr.h"
+#include "syncop.h"
 
 #define XATTROP_SUBDIR "xattrop"
+#define BASE_INDICES_HOLDER_SUBDIR "base_indices_holder"
 
 call_stub_t *
 __index_dequeue (struct list_head *callstubs)
@@ -242,20 +244,40 @@ check_delete_stale_index_file (xlator_t *this, char *filename)
 {
         int             ret = 0;
         struct stat     st = {0};
+        struct stat     base_index_st = {0};
         char            filepath[PATH_MAX] = {0};
+        char            filepath_under_base_indices_holder[PATH_MAX] = {0};
         index_priv_t    *priv = NULL;
 
         priv = this->private;
+        if (priv->to_be_healed_states != synced_state)
+                return;
+
         make_file_path (priv->index_basepath, XATTROP_SUBDIR,
                         filename, filepath, sizeof (filepath));
+
+        make_file_path (priv->index_basepath, BASE_INDICES_HOLDER_SUBDIR,
+                        filename, filepath_under_base_indices_holder,
+                        sizeof (filepath_under_base_indices_holder));
+
+
+        ret = stat (filepath_under_base_indices_holder, &base_index_st);
+        if (ret) {
+                gf_log (THIS->name, GF_LOG_ERROR, "Base index is not created"
+                        "under index/base_indices_holder");
+                return;
+        }
+
         ret = stat (filepath, &st);
-        if (!ret && st.st_nlink == 1)
+        if (!ret && st.st_nlink == 2) {
                 unlink (filepath);
+                unlink (filepath_under_base_indices_holder);
+        }
 }
 
 static int
 index_fill_readdir (fd_t *fd, DIR *dir, off_t off,
-                    size_t size, gf_dirent_t *entries)
+                    size_t size, gf_dirent_t *entries, readdir_directory type)
 {
         off_t     in_case = -1;
         size_t    filled = 0;
@@ -298,7 +320,8 @@ index_fill_readdir (fd_t *fd, DIR *dir, off_t off,
                 }
 
                 if (!strncmp (entry->d_name, XATTROP_SUBDIR"-",
-                              strlen (XATTROP_SUBDIR"-"))) {
+                              strlen (XATTROP_SUBDIR"-")) &&
+                              (type == INDEX_XATTROP)) {
                         check_delete_stale_index_file (this, entry->d_name);
                         continue;
                 }
@@ -337,16 +360,175 @@ out:
 }
 
 int
+sync_base_indices (void *index_priv)
+{
+        index_priv_t    *priv = NULL;
+        DIR             *dir_base_holder  = NULL;
+        DIR             *xattrop_dir = NULL;
+        struct dirent   *entry = NULL;
+        char            base_indices_holder[PATH_MAX] = {0};
+        char            xattrop_directory[PATH_MAX] = {0};
+        char            base_index_path[PATH_MAX] = {0};
+        char            xattrop_index_path[PATH_MAX] = {0};
+        int             ret = 0;
+
+        priv = index_priv;
+
+        snprintf (base_indices_holder, PATH_MAX, "%s/%s", priv->index_basepath,
+                  BASE_INDICES_HOLDER_SUBDIR);
+        snprintf (xattrop_directory, PATH_MAX, "%s/%s", priv->index_basepath,
+                  XATTROP_SUBDIR);
+
+        if ((dir_base_holder = opendir(base_indices_holder)) == NULL) {
+                ret = -1;
+                goto out;
+        }
+        if ((xattrop_dir = opendir (xattrop_directory)) == NULL) {
+                ret = -1;
+                goto out;
+        }
+
+        priv->to_be_healed_states = sync_started;
+        while ((entry = readdir(xattrop_dir)) != NULL) {
+                if (!strcmp (entry->d_name, ".") ||
+                    !strcmp (entry->d_name, "..")) {
+                        continue;
+                }
+                if (strncmp (entry->d_name, XATTROP_SUBDIR"-",
+                    strlen (XATTROP_SUBDIR"-"))) {
+                        continue;
+                }
+                if (!strncmp (entry->d_name, XATTROP_SUBDIR"-",
+                    strlen (XATTROP_SUBDIR"-"))) {
+
+                    snprintf (xattrop_index_path, PATH_MAX, "%s/%s",
+                              xattrop_directory, entry->d_name);
+
+                    snprintf (base_index_path, PATH_MAX, "%s/%s",
+                              base_indices_holder, entry->d_name);
+
+                    ret = link (xattrop_index_path, base_index_path);
+                    if (ret && errno != EEXIST)
+                        goto out;
+
+                }
+        }
+        ret = closedir (xattrop_dir);
+        if (ret)
+                goto out;
+        ret = closedir (dir_base_holder);
+        if (ret)
+                goto out;
+
+        ret = 0;
+out:
+        return ret;
+
+}
+
+int
+base_indices_syncing_done (int ret, call_frame_t *frame, void *data)
+{
+        index_priv_t         *priv = NULL;
+        priv = data;
+
+        if (!priv)
+                goto out;
+
+        if (ret) {
+                priv->to_be_healed_states = sync_not_started;
+        } else {
+                priv->to_be_healed_states = synced_state;
+        }
+
+        STACK_DESTROY (frame->root);
+
+out:
+        return 0;
+}
+
+int
+sync_base_indices_from_xattrop (xlator_t *this)
+{
+
+        index_priv_t          *priv = NULL;
+        char                  base_indices_holder[PATH_MAX] =  {0};
+        int                   ret = 0;
+        struct stat           st = {0};
+        DIR                   *dir = NULL;
+        struct dirent         *entry = NULL;
+        call_frame_t          *frame = NULL;
+
+        priv = this->private;
+
+        if (priv->to_be_healed_states != sync_not_started) {
+                ret = -1;
+                goto out;
+        }
+
+        snprintf (base_indices_holder, PATH_MAX, "%s/%s", priv->index_basepath,
+                  BASE_INDICES_HOLDER_SUBDIR);
+
+        ret = stat (base_indices_holder, &st);
+
+        if (ret && (errno != ENOENT)) {
+                goto out;
+        } else if (errno == ENOENT) {
+                ret = index_dir_create (this, BASE_INDICES_HOLDER_SUBDIR);
+                if (ret)
+                        goto out;
+        } else {
+                if ((dir = opendir (base_indices_holder)) == NULL) {
+                        ret = -1;
+                        goto out;
+                }
+                while ((entry = readdir (dir)) != NULL) {
+                        if (!strcmp (entry->d_name, ".") ||
+                            !strcmp (entry->d_name,"..")) {
+                                continue;
+                        }
+                        ret = unlink (entry->d_name);
+                        if (ret)
+                                goto out;
+                }
+                closedir (dir);
+        }
+
+        /*At this point of time we have index/base_indicies_holder directory
+         *is with no entries*/
+
+        frame = create_frame (this, this->ctx->pool);
+        if (!frame) {
+                ret = -1;
+                goto out;
+        }
+        set_lk_owner_from_ptr (&frame->root->lk_owner, frame->root);
+
+        frame->root->pid = LOW_PRIO_PROC_PID;
+
+        ret = synctask_new (this->ctx->env, sync_base_indices,
+                            base_indices_syncing_done,frame, priv);
+
+
+
+out:
+        return ret;
+
+}
+
+int
 index_add (xlator_t *this, uuid_t gfid, const char *subdir)
 {
         int32_t           op_errno = 0;
         char              gfid_path[PATH_MAX] = {0};
         char              index_path[PATH_MAX] = {0};
+        char              base_path[PATH_MAX] = {0};
         int               ret = 0;
         uuid_t            index = {0};
         index_priv_t      *priv = NULL;
         struct stat       st = {0};
         int               fd = 0;
+        int               index_created = 0;
 
         priv = this->private;
         GF_ASSERT_AND_GOTO_WITH_ERROR (this->name, !uuid_is_null (gfid),
@@ -364,9 +546,11 @@ index_add (xlator_t *this, uuid_t gfid, const char *subdir)
         ret = link (index_path, gfid_path);
         if (!ret || (errno == EEXIST))  {
                 ret = 0;
+                index_created = 1;
                 goto out;
         }
 
+
         op_errno = errno;
         if (op_errno == ENOENT) {
                 ret = index_dir_create (this, subdir);
@@ -398,10 +582,36 @@ index_add (xlator_t *this, uuid_t gfid, const char *subdir)
                         "add to index (%s)", uuid_utoa (gfid),
                         strerror (errno));
                 goto out;
+        } else {
+                index_created = 1;
+        }
+
+        if (priv->to_be_healed_states != sync_not_started) {
+                 make_index_path (priv->index_basepath,
+                                  GF_BASE_INDICES_HOLDER_GFID,
+                                  index, base_path, sizeof (base_path));
+                 ret = link (index_path, base_path);
+                 if (ret)
+                         goto out;
         }
 
         ret = 0;
 out:
+        /*If base_indices_holder is not created: create and sync
+         *If directory is present: delete contents and start syncing
+         *If syncing is in progress :No need to do any thing
+         *If syncing is done: No need to do anything*/
+        if (!ret) {
+                switch (priv->to_be_healed_states) {
+                        case sync_not_started:
+                                ret = sync_base_indices_from_xattrop (this);
+                                break;
+                        case sync_started:
+                        case synced_state:
+                                /*No need to do anything*/
+                                break;
+                }
+        }
         return ret;
 }
 
@@ -737,8 +947,18 @@ index_getxattr_wrapper (call_frame_t *frame, xlator_t *this,
                 goto done;
         }
 
-        ret = dict_set_static_bin (xattr, (char*)name, priv->xattrop_vgfid,
-                                   sizeof (priv->xattrop_vgfid));
+        if (!strcmp (name, GF_XATTROP_INDEX_GFID)) {
+
+                ret = dict_set_static_bin (xattr, (char*)name,
+                                           priv->xattrop_vgfid,
+                                           sizeof (priv->xattrop_vgfid));
+
+        } else if (!strcmp (name, GF_BASE_INDICES_HOLDER_GFID)) {
+
+                ret = dict_set_static_bin (xattr, (char*)name,
+                                           priv->base_indices_holder_vgfid,
+                                      sizeof (priv->base_indices_holder_vgfid));
+        }
         if (ret) {
                 ret = -ENOMEM;
                 gf_log (THIS->name, GF_LOG_ERROR, "xattrop index "
@@ -782,6 +1002,15 @@ index_lookup_wrapper (call_frame_t *frame, xlator_t *this,
         } else if (!uuid_compare (loc->pargfid, priv->xattrop_vgfid)) {
                 make_file_path (priv->index_basepath, XATTROP_SUBDIR,
                                 loc->name, path, sizeof (path));
+        } else if (!uuid_compare (loc->gfid,priv->base_indices_holder_vgfid)){
+                make_index_dir_path (priv->index_basepath,
+                                     BASE_INDICES_HOLDER_SUBDIR, path,
+                                    sizeof (path));
+                is_dir = _gf_true;
+        } else if (!uuid_compare (loc->pargfid, priv->base_indices_holder_vgfid)) {
+                make_file_path (priv->index_basepath,
+                                BASE_INDICES_HOLDER_SUBDIR,loc->name, path,
+                                sizeof (path));
         }
 
         ret = lstat (path, &lstatbuf);
@@ -803,10 +1032,14 @@ index_lookup_wrapper (call_frame_t *frame, xlator_t *this,
         }
 
         iatt_from_stat (&stbuf, &lstatbuf);
-        if (is_dir)
+        if (is_dir && !uuid_compare (loc->gfid, priv->xattrop_vgfid)) {
                 uuid_copy (stbuf.ia_gfid, priv->xattrop_vgfid);
-        else
+        } else if (is_dir &&
+                !uuid_compare (loc->gfid, priv->base_indices_holder_vgfid)) {
+                uuid_copy (stbuf.ia_gfid, priv->base_indices_holder_vgfid);
+        } else {
                 uuid_generate (stbuf.ia_gfid);
+        }
         stbuf.ia_ino = -1;
         op_ret = 0;
 done:
@@ -818,6 +1051,44 @@ done:
 }
 
 int32_t
+base_indices_readdir_wrapper (call_frame_t *frame, xlator_t *this,
+                              fd_t *fd, size_t size, off_t off, dict_t *xdata)
+{
+        index_priv_t    *priv = NULL;
+        char            base_indices_holder[PATH_MAX] = {0};
+        DIR             *dir = NULL;
+        int32_t         op_ret = -1;
+        int32_t         op_errno = 0;
+        int             count = 0;
+        gf_dirent_t     entries;
+
+        priv = this->private;
+
+        make_index_dir_path (priv->index_basepath, BASE_INDICES_HOLDER_SUBDIR,
+                             base_indices_holder, sizeof (base_indices_holder));
+
+        dir = opendir (base_indices_holder);
+        if (!dir) {
+                op_errno = EINVAL;
+                goto done;
+        }
+
+
+        INIT_LIST_HEAD (&entries.list);
+
+        count = index_fill_readdir (fd, dir, off, size, &entries,
+                                    BASE_INDICES_HOLDER);
+        /* pick ENOENT to indicate EOF */
+        op_errno = errno;
+        op_ret = count;
+        closedir (dir);
+done:
+        STACK_UNWIND_STRICT (readdir, frame, op_ret, op_errno, &entries, xdata);
+        gf_dirent_free (&entries);
+        return 0;
+}
+
+int32_t
 index_readdir_wrapper (call_frame_t *frame, xlator_t *this,
                        fd_t *fd, size_t size, off_t off, dict_t *xdata)
 {
@@ -848,7 +1119,8 @@ index_readdir_wrapper (call_frame_t *frame, xlator_t *this,
                 goto done;
         }
 
-        count = index_fill_readdir (fd, dir, off, size, &entries);
+        count = index_fill_readdir (fd, dir, off, size, &entries,
+                                    INDEX_XATTROP);
 
         /* pick ENOENT to indicate EOF */
         op_errno = errno;
@@ -915,7 +1187,10 @@ index_getxattr (call_frame_t *frame, xlator_t *this,
 {
         call_stub_t     *stub = NULL;
 
-        if (!name || strcmp (GF_XATTROP_INDEX_GFID, name))
+        if (!name)
+                goto out;
+        if (strcmp (GF_XATTROP_INDEX_GFID, name) &&
+            strcmp (GF_BASE_INDICES_HOLDER_GFID, name))
                 goto out;
 
         stub = fop_getxattr_stub (frame, index_getxattr_wrapper, loc, name,
@@ -942,7 +1217,9 @@ index_lookup (call_frame_t *frame, xlator_t *this,
         priv = this->private;
 
         if (uuid_compare (loc->gfid, priv->xattrop_vgfid) &&
-            uuid_compare (loc->pargfid, priv->xattrop_vgfid))
+            uuid_compare (loc->pargfid, priv->xattrop_vgfid) &&
+            uuid_compare (loc->gfid, priv->base_indices_holder_vgfid) &&
+            uuid_compare (loc->pargfid, priv->base_indices_holder_vgfid))
                 goto normal;
 
         stub = fop_lookup_stub (frame, index_lookup_wrapper, loc, xattr_req);
@@ -968,10 +1245,19 @@ index_readdir (call_frame_t *frame, xlator_t *this,
         index_priv_t    *priv = NULL;
 
         priv = this->private;
-        if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid))
+        if (uuid_compare (fd->inode->gfid, priv->xattrop_vgfid) &&
+            uuid_compare (fd->inode->gfid, priv->base_indices_holder_vgfid))
                 goto out;
-        stub = fop_readdir_stub (frame, index_readdir_wrapper, fd, size, off,
-                                 xdata);
+
+        if (!uuid_compare (fd->inode->gfid, priv->xattrop_vgfid)) {
+                stub = fop_readdir_stub (frame, index_readdir_wrapper, fd, size, 
+                                         off, xdata);
+        } else if (!uuid_compare (fd->inode->gfid,
+                                  priv->base_indices_holder_vgfid)) {
+                stub = fop_readdir_stub (frame, base_indices_readdir_wrapper,
+                                         fd, size, off, xdata);
+        }
+
         if (!stub) {
                 STACK_UNWIND_STRICT (readdir, frame, -1, ENOMEM, NULL, NULL);
                 return 0;
@@ -1075,6 +1361,9 @@ init (xlator_t *this)
         GF_OPTION_INIT ("index-base", priv->index_basepath, path, out);
         uuid_generate (priv->index);
         uuid_generate (priv->xattrop_vgfid);
+        /*base_indices_holder is a directory which contains hard links to
+         * all base indices inside indices/xattrop directory*/
+        uuid_generate (priv->base_indices_holder_vgfid);
         INIT_LIST_HEAD (&priv->callstubs);
 
         this->private = priv;
diff --git a/xlators/features/index/src/index.h b/xlators/features/index/src/index.h
index 661dcdbc417..d6dcb1c23b4 100644
--- a/xlators/features/index/src/index.h
+++ b/xlators/features/index/src/index.h
@@ -36,14 +36,28 @@ typedef struct index_fd_ctx {
         DIR *dir;
 } index_fd_ctx_t;
 
+typedef enum {
+        sync_not_started,
+        sync_started,
+        synced_state,
+} to_be_healed_states_t;
+
+typedef enum {
+        INDEX_XATTROP,
+        BASE_INDICES_HOLDER,
+} readdir_directory;
+
 typedef struct index_priv {
         char *index_basepath;
         uuid_t index;
         gf_lock_t lock;
         uuid_t xattrop_vgfid;//virtual gfid of the xattrop index dir
+        uuid_t base_indices_holder_vgfid; //virtual gfid of the
+                                          //to_be_healed_xattrop directory
         struct list_head callstubs;
         pthread_mutex_t mutex;
         pthread_cond_t  cond;
+        to_be_healed_states_t to_be_healed_states;
 } index_priv_t;
 
 #define INDEX_STACK_UNWIND(fop, frame, params ...)      \