glusterd: handle replace-brick in paused state.

This change ensures that glusterd retains 'state' information
of an ongoing replace brick operation even if it went down midway.

Change-Id: I697c2f27a96b3200c7865bd1d3a1de9692ce0da1
BUG: 3252
Reviewed-on: http://review.gluster.com/6
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

3 files changed, 96 insertions, 27 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index 230c29c47d3..780738df23d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -4056,11 +4056,6 @@ glusterd_op_replace_brick (dict_t *dict, dict_t *rsp_dict)
                                 "Failed to generate nfs volume file");
 		}
 
-		ret = glusterd_store_volinfo (volinfo,
-                                              GLUSTERD_VOLINFO_VER_AC_INCREMENT);
-
-		if (ret)
-			goto out;
 
 		ret = glusterd_fetchspec_notify (THIS);
                 glusterd_set_rb_status (volinfo, GF_RB_STATUS_NONE);
@@ -4129,6 +4124,16 @@ glusterd_op_replace_brick (dict_t *dict, dict_t *rsp_dict)
                         "received status - doing nothing");
                 ctx = glusterd_op_get_ctx (GD_OP_REPLACE_BRICK);
                 if (ctx) {
+                        if (glusterd_is_rb_paused (volinfo)) {
+                                ret = dict_set_str (ctx, "status-reply",
+                                                 "replace brick has been paused");
+                                if (ret)
+                                        gf_log (THIS->name, GF_LOG_ERROR,
+                                                "failed to set pump status"
+                                                "in ctx");
+                                goto out;
+                        }
+
                         ret = rb_do_operation_status (volinfo, src_brickinfo,
                                                       dst_brickinfo);
                         if (ret)
@@ -4142,9 +4147,12 @@ glusterd_op_replace_brick (dict_t *dict, dict_t *rsp_dict)
                 ret = -1;
                 goto out;
         }
-
+        if (!ret && replace_op != GF_REPLACE_OP_STATUS)
+		ret = glusterd_store_volinfo (volinfo,
+                                              GLUSTERD_VOLINFO_VER_AC_INCREMENT);
         if (ret)
-                goto out;
+                gf_log (THIS->name, GF_LOG_ERROR, "Couldn't store"
+                        " replace brick operation's state");
 
 out:
         return ret;
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.c b/xlators/mgmt/glusterd/src/glusterd-store.c
index 110d459dca3..e1bde63772a 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.c
+++ b/xlators/mgmt/glusterd/src/glusterd-store.c
@@ -567,8 +567,8 @@ glusterd_volume_exclude_options_write (int fd, glusterd_volinfo_t *volinfo)
         GF_ASSERT (fd > 0);
         GF_ASSERT (volinfo);
 
-        char                    buf[4096] = {0,};
-        int32_t                 ret = -1;
+        char                    buf[PATH_MAX+20] = {0,};
+        int32_t                 ret              = -1;
 
         snprintf (buf, sizeof (buf), "%d", volinfo->type);
         ret = glusterd_store_save_value (fd, GLUSTERD_STORE_KEY_VOL_TYPE, buf);
@@ -614,6 +614,28 @@ glusterd_volume_exclude_options_write (int fd, glusterd_volinfo_t *volinfo)
         if (ret)
                 goto out;
 
+        snprintf (buf, sizeof (buf), "%d", volinfo->rb_status);
+        ret = glusterd_store_save_value (fd, GLUSTERD_STORE_KEY_RB_STATUS,
+                                         buf);
+        if (ret)
+                goto out;
+
+        if (volinfo->rb_status > GF_RB_STATUS_NONE) {
+                snprintf (buf, sizeof (buf), "%s:%s", volinfo->src_brick->hostname,
+                          volinfo->src_brick->path);
+                ret = glusterd_store_save_value (fd, GLUSTERD_STORE_KEY_RB_SRC_BRICK,
+                                                 buf);
+                if (ret)
+                        goto out;
+
+                snprintf (buf, sizeof (buf), "%d:%s:%s", volinfo->dst_brick->port,
+                          volinfo->dst_brick->hostname, volinfo->dst_brick->path);
+                ret = glusterd_store_save_value (fd, GLUSTERD_STORE_KEY_RB_DST_BRICK,
+                                                 buf);
+                if (ret)
+                        goto out;
+        }
+
 out:
         if (ret)
                 gf_log ("", GF_LOG_ERROR, "Unable to write volume values"
@@ -1504,16 +1526,18 @@ out:
 int32_t
 glusterd_store_retrieve_volume (char    *volname)
 {
-        int32_t                 ret = -1;
-        glusterd_volinfo_t      *volinfo = NULL;
-        glusterd_store_iter_t   *iter = NULL;
-        char                    *key = NULL;
-        char                    *value = NULL;
-        char                    volpath[PATH_MAX] = {0,};
-        glusterd_conf_t         *priv = NULL;
-        char                    path[PATH_MAX] = {0,};
-        int                     exists = 0;
-        glusterd_store_op_errno_t op_errno = GD_STORE_SUCCESS;
+        int32_t                   ret                   = -1;
+        glusterd_volinfo_t        *volinfo              = NULL;
+        glusterd_store_iter_t     *iter                 = NULL;
+        char                      *key                  = NULL;
+        char                      *value                = NULL;
+        char                      volpath[PATH_MAX]     = {0,};
+        glusterd_conf_t           *priv                 = NULL;
+        char                      path[PATH_MAX]        = {0,};
+        char                      dst_brick[PATH_MAX]   = {0, };
+        int                       exists                = 0;
+        int                       dst_port              = -1;
+        glusterd_store_op_errno_t op_errno              = GD_STORE_SUCCESS;
 
         ret = glusterd_volinfo_new (&volinfo);
 
@@ -1547,36 +1571,70 @@ glusterd_store_retrieve_volume (char    *volname)
                               strlen (GLUSTERD_STORE_KEY_VOL_TYPE))) {
                         volinfo->type = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_COUNT,
-                            strlen (GLUSTERD_STORE_KEY_VOL_COUNT))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_COUNT))) {
                         volinfo->brick_count = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_STATUS,
-                            strlen (GLUSTERD_STORE_KEY_VOL_STATUS))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_STATUS))) {
                         volinfo->status = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_VERSION,
-                            strlen (GLUSTERD_STORE_KEY_VOL_VERSION))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_VERSION))) {
                         volinfo->version = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_PORT,
-                            strlen (GLUSTERD_STORE_KEY_VOL_PORT))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_PORT))) {
                         volinfo->port = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_SUB_COUNT,
-                            strlen (GLUSTERD_STORE_KEY_VOL_SUB_COUNT))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_SUB_COUNT))) {
                         volinfo->sub_count = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_STRIPE_CNT,
-                            strlen (GLUSTERD_STORE_KEY_VOL_STRIPE_CNT))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_STRIPE_CNT))) {
                         volinfo->stripe_count = atoi (value);
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_TRANSPORT,
-                            strlen (GLUSTERD_STORE_KEY_VOL_TRANSPORT))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_TRANSPORT))) {
                         volinfo->transport_type = atoi (value);
                         volinfo->nfs_transport_type = volinfo->transport_type;
                         if (volinfo->transport_type == GF_TRANSPORT_BOTH_TCP_RDMA) {
                                 volinfo->nfs_transport_type = GF_DEFAULT_NFS_TRANSPORT;
                         }
                 } else if (!strncmp (key, GLUSTERD_STORE_KEY_VOL_ID,
-                            strlen (GLUSTERD_STORE_KEY_VOL_ID))) {
+                                     strlen (GLUSTERD_STORE_KEY_VOL_ID))) {
                         ret = uuid_parse (value, volinfo->volume_id);
                         if (ret)
                                 gf_log ("", GF_LOG_WARNING,
                                         "failed to parse uuid");
+
+                } else if (!strncmp (key, GLUSTERD_STORE_KEY_RB_STATUS,
+                                     strlen (GLUSTERD_STORE_KEY_RB_STATUS))) {
+                        glusterd_set_rb_status (volinfo, atoi (value));
+
+                } else if (volinfo->rb_status > GF_RB_STATUS_NONE &&
+                           !strncmp (key, GLUSTERD_STORE_KEY_RB_SRC_BRICK,
+                                     strlen (GLUSTERD_STORE_KEY_RB_SRC_BRICK))) {
+                        ret = glusterd_brickinfo_from_brick (value,
+                                                             &volinfo->src_brick);
+                        if (ret) {
+                                gf_log ("", GF_LOG_ERROR, "Unable to create"
+                                        " src brickinfo");
+                                goto out;
+                        }
+
+                } else if (volinfo->rb_status > GF_RB_STATUS_NONE &&
+                           !strncmp (key, GLUSTERD_STORE_KEY_RB_DST_BRICK,
+                                     strlen (GLUSTERD_STORE_KEY_RB_DST_BRICK))) {
+                        sscanf (value, "%d:%s", &dst_port, dst_brick);
+                        if (dst_port == -1 || dst_brick[0] == 0) {
+                                gf_log ("", GF_LOG_ERROR, "replace brick: "
+                                        "dst brick info is invalid");
+                                goto out;
+                        }
+                        ret = glusterd_brickinfo_from_brick (dst_brick,
+                                                             &volinfo->dst_brick);
+                        if (ret) {
+                                gf_log ("", GF_LOG_ERROR, "Unable to create"
+                                        " dst brickinfo");
+                                goto out;
+                        }
+                        volinfo->dst_brick->port = dst_port;
+
                 } else if (strstr (key, "slave")) {
                         ret = dict_set_dynstr (volinfo->gsync_slaves, key,
                                                 gf_strdup (value));
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.h b/xlators/mgmt/glusterd/src/glusterd-store.h
index 811347b396a..460a315f470 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.h
+++ b/xlators/mgmt/glusterd/src/glusterd-store.h
@@ -55,6 +55,9 @@ typedef enum glusterd_store_ver_ac_{
 #define GLUSTERD_STORE_KEY_VOL_VERSION    "version"
 #define GLUSTERD_STORE_KEY_VOL_TRANSPORT  "transport-type"
 #define GLUSTERD_STORE_KEY_VOL_ID         "volume-id"
+#define GLUSTERD_STORE_KEY_RB_STATUS      "rb_status"
+#define GLUSTERD_STORE_KEY_RB_SRC_BRICK   "rb_src"
+#define GLUSTERD_STORE_KEY_RB_DST_BRICK   "rb_dst"
 
 #define GLUSTERD_STORE_KEY_BRICK_HOSTNAME "hostname"
 #define GLUSTERD_STORE_KEY_BRICK_PATH     "path"