mgmt/glusterd: replace-brick validations

Signed-off-by: Pranith Kumar K <pranithk@gluster.com>
Signed-off-by: Vijay Bellur <vijay@dev.gluster.com>

BUG: 1657 (validations for replace-brick while stage op)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=1657

10 files changed, 193 insertions, 81 deletions

diff --git a/cli/src/cli-cmd-parser.c b/cli/src/cli-cmd-parser.c
index 2b7cdd9f6..9de13e56c 100644
--- a/cli/src/cli-cmd-parser.c
+++ b/cli/src/cli-cmd-parser.c
@@ -729,42 +729,45 @@ cli_cmd_volume_replace_brick_parse (const char **words, int wordcount,
         }
 
         delimiter = strchr ((char *)words[3], ':');
-        if (delimiter && delimiter != words[3]
-            && *(delimiter+1) == '/') {
+        if (!delimiter || delimiter == words[3]
+            || *(delimiter+1) != '/') {
+                cli_out ("wrong brick type: %s, use "
+                        "<HOSTNAME>:<export-dir-abs-path>", words[3]);
+                ret = -1;
+                goto out;
+        } else {
                 cli_path_strip_trailing_slashes (delimiter + 1);
-                ret = dict_set_str (dict, "src-brick", (char *)words[3]);
-
-                if (ret)
-                        goto out;
+        }
+        ret = dict_set_str (dict, "src-brick", (char *)words[3]);
 
-                if (wordcount < 5) {
-                        ret = -1;
-                        goto out;
-                }
+        if (ret)
+                goto out;
 
-                delimiter = strchr ((char *)words[4], ':');
-                if (!delimiter || delimiter == words[4]
-                    || *(delimiter+1) != '/') {
-                        cli_out ("wrong brick type: %s, use "
-                                "<HOSTNAME>:<export-dir-abs-path>", words[4]);
-                        ret = -1;
-                        goto out;
-                } else {
-                        cli_path_strip_trailing_slashes (delimiter + 1);
-                }
+        if (wordcount < 5) {
+                ret = -1;
+                goto out;
+        }
 
+        delimiter = strchr ((char *)words[4], ':');
+        if (!delimiter || delimiter == words[4]
+            || *(delimiter+1) != '/') {
+                cli_out ("wrong brick type: %s, use "
+                        "<HOSTNAME>:<export-dir-abs-path>", words[4]);
+                ret = -1;
+                goto out;
+        } else {
+                cli_path_strip_trailing_slashes (delimiter + 1);
+        }
 
-                ret = dict_set_str (dict, "dst-brick", (char *)words[4]);
 
-                if (ret)
-                        goto out;
+        ret = dict_set_str (dict, "dst-brick", (char *)words[4]);
 
-                op_index = 5;
-        } else {
-                op_index = 3;
-        }
+        if (ret)
+                goto out;
 
-        if (wordcount < (op_index + 1)) {
+        op_index = 5;
+        if ((wordcount < (op_index + 1)) ||
+            (wordcount > (op_index + 1))) {
                 ret = -1;
                 goto out;
         }
diff --git a/cli/src/cli-cmd-volume.c b/cli/src/cli-cmd-volume.c
index 28104c187..c83451778 100644
--- a/cli/src/cli-cmd-volume.c
+++ b/cli/src/cli-cmd-volume.c
@@ -611,7 +611,7 @@ void
 cli_cmd_volume_replace_brick_usage ()
 {
         cli_out("Usage: volume replace-brick <VOLNAME> "
-                "(<BRICK> <NEW-BRICK>) start|pause|abort|status");
+                "<BRICK> <NEW-BRICK> start|pause|abort|status");
 }
 
 
diff --git a/cli/src/cli3_1-cops.c b/cli/src/cli3_1-cops.c
index 53b80c556..78683ccab 100644
--- a/cli/src/cli3_1-cops.c
+++ b/cli/src/cli3_1-cops.c
@@ -937,6 +937,9 @@ gf_cli3_1_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
                 break;
         }
 
+        if (rsp.op_ret && (strcmp (rsp.op_errstr, ""))) {
+                rb_operation_str = rsp.op_errstr;
+        }
 
         gf_log ("cli", GF_LOG_NORMAL, "Received resp to replace brick");
         cli_out ("%s",
diff --git a/rpc/xdr/src/cli1-xdr.c b/rpc/xdr/src/cli1-xdr.c
index 93f7e7681..ca2bc09f3 100644
--- a/rpc/xdr/src/cli1-xdr.c
+++ b/rpc/xdr/src/cli1-xdr.c
@@ -94,7 +94,7 @@ bool_t
 xdr_gf1_cli_probe_rsp (XDR *xdrs, gf1_cli_probe_rsp *objp)
 {
 
-        register int32_t *buf;
+	register int32_t *buf;
 
 	if (xdrs->x_op == XDR_ENCODE) {
 		buf = XDR_INLINE (xdrs, 3 * BYTES_PER_XDR_UNIT);
@@ -452,6 +452,8 @@ xdr_gf1_cli_replace_brick_rsp (XDR *xdrs, gf1_cli_replace_brick_rsp *objp)
 		 return FALSE;
 	 if (!xdr_int (xdrs, &objp->op_errno))
 		 return FALSE;
+	 if (!xdr_string (xdrs, &objp->op_errstr, ~0))
+		 return FALSE;
 	 if (!xdr_string (xdrs, &objp->volname, ~0))
 		 return FALSE;
 	 if (!xdr_string (xdrs, &objp->status, ~0))
diff --git a/rpc/xdr/src/cli1-xdr.h b/rpc/xdr/src/cli1-xdr.h
index c6d8e8bf5..0225acefe 100644
--- a/rpc/xdr/src/cli1-xdr.h
+++ b/rpc/xdr/src/cli1-xdr.h
@@ -275,6 +275,7 @@ typedef struct gf1_cli_replace_brick_req gf1_cli_replace_brick_req;
 struct gf1_cli_replace_brick_rsp {
 	int op_ret;
 	int op_errno;
+	char *op_errstr;
 	char *volname;
 	char *status;
 };
diff --git a/rpc/xdr/src/cli1.x b/rpc/xdr/src/cli1.x
index 3f37f6d75..ba4288be5 100644
--- a/rpc/xdr/src/cli1.x
+++ b/rpc/xdr/src/cli1.x
@@ -188,6 +188,7 @@ struct gf1_cli_get_vol_rsp {
  struct gf1_cli_replace_brick_rsp {
         int     op_ret;
         int     op_errno;
+        string  op_errstr<>;
         string  volname<>;
         string  status<>;
 }  ;
diff --git a/xlators/mgmt/glusterd/src/glusterd-handler.c b/xlators/mgmt/glusterd/src/glusterd-handler.c
index 0c33b6590..4d9a662b3 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handler.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handler.c
@@ -1126,8 +1126,6 @@ glusterd_handle_create_volume (rpcsvc_request_t *req)
         gf1_cli_create_vol_rsp  rsp         = {0,};
         glusterd_conf_t        *priv        = NULL;
         int                     err_ret     = 0;
-        glusterd_brickinfo_t   *tmpbrkinfo  = NULL;
-        glusterd_volinfo_t     *volinfo     = NULL;
         xlator_t               *this        = NULL;
         char                   *free_ptr    = NULL;
         char                   *trans_type  = NULL;
@@ -1137,10 +1135,10 @@ glusterd_handle_create_volume (rpcsvc_request_t *req)
         GF_ASSERT (req);
 
         this = THIS;
-        priv = this->private;
-
         GF_ASSERT(this);
 
+        priv = this->private;
+
         if (!gf_xdr_to_cli_create_vol_req (req->msg[0], &cli_req)) {
                 //failed to decode msg;
                 req->rpc_err = GARBAGE_ARGS;
@@ -1250,22 +1248,12 @@ glusterd_handle_create_volume (rpcsvc_request_t *req)
                         goto out;
                 }
 brick_validation:
-                list_for_each_entry (volinfo, &priv->volumes, vol_list) {
-
-                        list_for_each_entry (tmpbrkinfo, &volinfo->bricks,
-                                             brick_list) {
-
-                                if ((!strcmp(brickinfo->hostname, tmpbrkinfo->
-                                    hostname) && !strcmp(brickinfo->path,
-                                    tmpbrkinfo->path))) {
-                                        snprintf(err_str, 1048, "Brick %s already"
-                                                " in use", brick);
-                                        gf_log ("glusterd", GF_LOG_ERROR, "%s",
-                                                err_str);
-                                        err_ret = 1;
-                                        goto out;
-                                }
-                        }
+                err_ret = glusterd_is_exisiting_brick (brickinfo->hostname,
+                                                       brickinfo->path);
+                if (err_ret) {
+                        snprintf(err_str, 1048, "Brick: %s already in use",
+                                 brick);
+                        goto out;
                 }
         }
         ret = glusterd_create_volume (req, dict);
@@ -1406,18 +1394,16 @@ glusterd_handle_add_brick (rpcsvc_request_t *req)
         char                            err_str[1048];
         gf1_cli_add_brick_rsp           rsp = {0,};
         glusterd_volinfo_t              *volinfo = NULL;
-        glusterd_brickinfo_t            *tmpbrkinfo = NULL;
         int32_t                         err_ret = 0;
-        glusterd_volinfo_t              *tmpvolinfo = NULL;
         glusterd_conf_t                 *priv = NULL;
         xlator_t                        *this = NULL;
         char                            *free_ptr = NULL;
 
         this = THIS;
-        priv = this->private;
-
         GF_ASSERT(this);
 
+        priv = this->private;
+
         GF_ASSERT (req);
 
         if (!gf_xdr_to_cli_add_brick_req (req->msg[0], &cli_req)) {
@@ -1544,22 +1530,12 @@ brick_val:
                         goto out;
                 }
 brick_validation:
-                list_for_each_entry (tmpvolinfo, &priv->volumes, vol_list) {
-
-                        list_for_each_entry (tmpbrkinfo, &tmpvolinfo->bricks,
-                                             brick_list) {
-
-                                if ((!strcmp(brickinfo->hostname, tmpbrkinfo->
-                                    hostname) && !strcmp(brickinfo->path,
-                                    tmpbrkinfo->path))) {
-                                        snprintf(err_str, 1048, "Brick %s already"
-                                                "in use", brick);
-                                        gf_log ("glusterd", GF_LOG_ERROR, "%s",
-                                                err_str);
-                                        err_ret = 1;
-                                        goto out;
-                                }
-                        }
+                err_ret = glusterd_is_exisiting_brick (brickinfo->hostname,
+                                                       brickinfo->path);
+                if (err_ret) {
+                        snprintf(err_str, 1048, "Brick: %s already in use",
+                                 brick);
+                        goto out;
                 }
         }
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index 4506b3a68..b8d2778e3 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -753,16 +753,22 @@ out:
 }
 
 static int
-glusterd_op_stage_replace_brick (gd1_mgmt_stage_op_req *req)
+glusterd_op_stage_replace_brick (gd1_mgmt_stage_op_req *req, char **op_errstr)
 {
         int                                      ret           = 0;
         dict_t                                  *dict          = NULL;
         char                                    *src_brick     = NULL;
         char                                    *dst_brick     = NULL;
         char                                    *volname       = NULL;
+        gf1_cli_replace_op                       replace_op    = 0;
         glusterd_volinfo_t                      *volinfo       = NULL;
         glusterd_brickinfo_t                    *src_brickinfo = NULL;
-        gf_boolean_t                             exists        = _gf_false;
+        char                                    *host          = NULL;
+        char                                    *path          = NULL;
+        char                                    msg[2048]      = {0};
+        char                                    *dup_dstbrick  = NULL;
+        glusterd_peerinfo_t                     *peerinfo = NULL;
+        struct stat                             st_buf = {0,};
 
         GF_ASSERT (req);
 
@@ -804,36 +810,116 @@ glusterd_op_stage_replace_brick (gd1_mgmt_stage_op_req *req)
                 goto out;
         }
 
+        ret = dict_get_int32 (dict, "operation", (int32_t *)&replace_op);
+        if (ret) {
+                gf_log ("", GF_LOG_DEBUG,
+                        "dict get on replace-brick operation failed");
+                goto out;
+        }
+
         ret = glusterd_volinfo_find (volname, &volinfo);
         if (ret) {
-                gf_log ("", GF_LOG_ERROR, "Unable to allocate memory");
+                snprintf (msg, sizeof (msg), "volume: %s does not exist",
+                          volname);
+                *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        if (GLUSTERD_STATUS_STARTED != volinfo->status) {
+                ret = -1;
+                snprintf (msg, sizeof (msg), "volume: %s is not started",
+                          volname);
+                *op_errstr = gf_strdup (msg);
                 goto out;
         }
 
         ret = glusterd_brickinfo_get (src_brick, volinfo,
                                       &src_brickinfo);
         if (ret) {
-                gf_log ("", GF_LOG_DEBUG, "Unable to get src-brickinfo");
+                snprintf (msg, sizeof (msg), "brick: %s does not exist in "
+                          "volume: %s", src_brick, volname);
+                *op_errstr = gf_strdup (msg);
                 goto out;
         }
 
         if (!glusterd_is_local_addr (src_brickinfo->hostname)) {
                 gf_log ("", GF_LOG_DEBUG,
                         "I AM THE SOURCE HOST");
-                exists = glusterd_check_volume_exists (volname);
+        }
 
-                if (!exists) {
-                        gf_log ("", GF_LOG_ERROR, "Volume with name: %s "
-                                "does not exist",
-                                volname);
+        dup_dstbrick = gf_strdup (dst_brick);
+        if (!dup_dstbrick) {
+                ret = -1;
+                gf_log ("", GF_LOG_ERROR, "Memory allocation failed");
+                goto out;
+        }
+        host = strtok (dup_dstbrick, ":");
+        path = strtok (NULL, ":");
+
+        if (!host || !path) {
+                gf_log ("", GF_LOG_ERROR,
+                        "dst brick %s is not of form <HOSTNAME>:<export-dir>",
+                        dst_brick);
+                ret = -1;
+                goto out;
+        }
+        if (glusterd_is_exisiting_brick (host, path)) {
+                snprintf(msg, sizeof(msg), "Brick: %s:%s already in use",
+                         host, path);
+                *op_errstr = gf_strdup (msg);
+                ret = -1;
+                goto out;
+        }
+
+        if (!glusterd_is_local_addr (host)) {
+                ret = stat (path, &st_buf);
+                if (ret == -1) {
+                        snprintf (msg, sizeof (msg) ,"path: %s for brick: %s"
+                        " does not exist", path, dst_brick);
+                        gf_log ("glusterd", GF_LOG_ERROR, "%s", msg);
+                        *op_errstr = gf_strdup (msg);
+                        goto out;
+                }
+                if (!S_ISDIR (st_buf.st_mode)) {
+                        snprintf (msg, sizeof (msg), "Volume name %s, brick"
+                                  ": %s, path %s is not a directory", volname,
+                                  dst_brick, path);
+                        gf_log ("glusterd", GF_LOG_ERROR,
+                                "%s", msg);
+                        *op_errstr = gf_strdup (msg);
+                        ret = -1;
+                        goto out;
+                }
+        } else {
+                ret = glusterd_friend_find (NULL, host, &peerinfo);
+                if (ret) {
+                        snprintf (msg, sizeof (msg), "%s, is not a friend",
+                                  host);
+                        *op_errstr = gf_strdup (msg);
+                        goto out;
+                }
+
+                if (!peerinfo->connected) {
+                        snprintf (msg, sizeof (msg), "%s, is not connected at "
+                                  "the moment", host);
+                        *op_errstr = gf_strdup (msg);
                         ret = -1;
                         goto out;
                 }
-        }
 
+                if (GD_FRIEND_STATE_BEFRIENDED != peerinfo->state.state) {
+                        snprintf (msg, sizeof (msg), "%s, is not befriended "
+                                  "at the moment", host);
+                        *op_errstr = gf_strdup (msg);
+                        ret = -1;
+                        goto out;
+                }
+        }
         ret = 0;
 
 out:
+        if (dup_dstbrick)
+                GF_FREE (dup_dstbrick);
         if (dict)
                 dict_unref (dict);
         gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
@@ -3657,6 +3743,10 @@ glusterd_op_send_cli_response (int32_t op, int32_t op_ret,
                                         rsp.status = "";
                                 rsp.op_ret = op_ret;
                                 rsp.op_errno = op_errno;
+                                if (op_errstr)
+                                        rsp.op_errstr = op_errstr;
+                                else
+                                        rsp.op_errstr = "";
                                 rsp.volname = "";
                                 cli_rsp = &rsp;
                                 sfunc = gf_xdr_serialize_cli_replace_brick_rsp;
@@ -3949,7 +4039,7 @@ glusterd_op_stage_validate (gd1_mgmt_stage_op_req *req, char **op_errstr)
                         break;
 
                 case GD_OP_REPLACE_BRICK:
-                        ret = glusterd_op_stage_replace_brick (req);
+                        ret = glusterd_op_stage_replace_brick (req, op_errstr);
                         break;
 
                 case GD_OP_SET_VOLUME:
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index 358d6ffe4..e3592891b 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -1716,3 +1716,37 @@ glusterd_volume_count_get (void)
         return ret;
 
 }
+
+int
+glusterd_is_exisiting_brick (char *hostname, char *path)
+{
+        glusterd_brickinfo_t            *tmpbrkinfo = NULL;
+        glusterd_volinfo_t              *volinfo     = NULL;
+        glusterd_conf_t                 *priv = NULL;
+        xlator_t                        *this = NULL;
+        int                             ret = 0;
+
+        GF_ASSERT (hostname);
+        GF_ASSERT (path);
+
+        this = THIS;
+        GF_ASSERT (this);
+
+        priv = this->private;
+        list_for_each_entry (volinfo, &priv->volumes, vol_list) {
+
+                list_for_each_entry (tmpbrkinfo, &volinfo->bricks,
+                                     brick_list) {
+
+                        if ((!strcmp(hostname, tmpbrkinfo-> hostname)
+                            && !strcmp(path, tmpbrkinfo->path))) {
+                                gf_log ("glusterd", GF_LOG_ERROR, "Brick %s:%s"
+                                        " already in use", hostname, path);
+                                ret = 1;
+                                goto out;
+                        }
+                }
+        }
+out:
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index e03968dfe..72715a0f5 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -158,4 +158,6 @@ glusterd_volume_count_get (void);
 int32_t
 glusterd_add_volume_to_dict (glusterd_volinfo_t *volinfo,
                              dict_t  *dict, int32_t count);
+int
+glusterd_is_exisiting_brick (char *hostname, char *path);
 #endif