glusterd: Do not allow remove-brick commit if the bricks are not decommissioned

Change-Id: Ibf47ea848bbb7ae37ccf91c40e5fe0e2338438b7
BUG: 1027171
Signed-off-by: Vijaikumar M <vmallika@redhat.com>
Reviewed-on: http://review.gluster.org/6233
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Krishnan Parthasarathi <kparthas@redhat.com>

2 files changed, 115 insertions, 24 deletions

diff --git a/tests/bugs/bug-1027171.t b/tests/bugs/bug-1027171.t
new file mode 100644
index 00000000000..c1f4bd809fe
--- /dev/null
+++ b/tests/bugs/bug-1027171.t
@@ -0,0 +1,53 @@
+#!/bin/bash
+
+#Test case: Do not allow commit if the bricks are not decommissioned
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+
+cleanup;
+
+#Basic checks
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume info
+
+#Create a Distributed volume
+TEST $CLI volume create $V0 $H0:$B0/${V0}{1..2};
+TEST $CLI volume start $V0
+
+#Remove bricks and commit without starting
+function remove_brick_commit_status {
+        $CLI volume remove-brick $V0 \
+        $H0:$B0/${V0}2 commit 2>&1 |grep -oE "success|decommissioned"
+}
+EXPECT "decommissioned"  remove_brick_commit_status;
+
+TEST $CLI volume stop $V0
+TEST $CLI volume delete $V0
+TEST ! $CLI volume info $V0
+
+#Create a Distributed-Replicate volume
+TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{1..4};
+TEST $CLI volume start $V0
+
+#Try to reduce replica count with start option
+function remove_brick_start_status {
+        $CLI volume remove-brick $V0 replica 1 \
+        $H0:$B0/${V0}1  $H0:$B0/${V0}3 start 2>&1 |grep -oE "success|failed"
+}
+EXPECT "failed"  remove_brick_start_status;
+
+#Remove bricks with commit option
+function remove_brick_commit_status2 {
+        $CLI volume remove-brick $V0 replica 1 \
+        $H0:$B0/${V0}1  $H0:$B0/${V0}3 commit 2>&1  |
+        grep -oE "success|decommissioned"
+}
+EXPECT "decommissioned"  remove_brick_commit_status2;
+
+TEST $CLI volume stop $V0
+TEST $CLI volume delete $V0
+TEST ! $CLI volume info $V0
+
+cleanup;
diff --git a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
index df6087659ef..6c316af886d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
@@ -1383,16 +1383,20 @@ out:
 int
 glusterd_op_stage_remove_brick (dict_t *dict, char **op_errstr)
 {
-        int                 ret         = -1;
-        char               *volname     = NULL;
-        glusterd_volinfo_t *volinfo     = NULL;
-        char               *errstr      = NULL;
-        int32_t             brick_count = 0;
-        char                msg[2048]   = {0,};
-        int32_t             flag        = 0;
-        gf1_op_commands     cmd         = GF_OP_CMD_NONE;
-        char               *task_id_str = NULL;
-        xlator_t           *this        = NULL;
+        int                     ret         = -1;
+        char                   *volname     = NULL;
+        glusterd_volinfo_t     *volinfo     = NULL;
+        char                   *errstr      = NULL;
+        int32_t                 brick_count = 0;
+        char                    msg[2048]   = {0,};
+        int32_t                 flag        = 0;
+        gf1_op_commands         cmd         = GF_OP_CMD_NONE;
+        char                   *task_id_str = NULL;
+        xlator_t               *this        = NULL;
+        int                     i           = 1;
+        char                    key[256]    = {0,};
+        char                   *brick       = NULL;
+        glusterd_brickinfo_t   *brickinfo   = NULL;
 
         this = THIS;
         GF_ASSERT (this);
@@ -1426,11 +1430,26 @@ glusterd_op_stage_remove_brick (dict_t *dict, char **op_errstr)
 
         ret = dict_get_int32 (dict, "command", &flag);
         if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "Unable to get brick count");
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to get brick command");
                 goto out;
         }
         cmd = flag;
 
+        ret = dict_get_int32 (dict, "count", &brick_count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get brick count");
+                goto out;
+        }
+
+        ret = 0;
+        if (volinfo->brick_count == brick_count) {
+                errstr = gf_strdup ("Deleting all the bricks of the "
+                                    "volume is not allowed");
+                ret = -1;
+                goto out;
+        }
+
         ret = -1;
         switch (cmd) {
         case GF_OP_CMD_NONE:
@@ -1510,25 +1529,44 @@ glusterd_op_stage_remove_brick (dict_t *dict, char **op_errstr)
                                             "is in progress");
                         goto out;
                 }
+
+                /* Do not allow commit if the bricks are not decommissioned */
+                for ( i = 1; i <= brick_count; i++ ) {
+                        snprintf (key, sizeof (key), "brick%d", i);
+                        ret = dict_get_str (dict, key, &brick);
+                        if (ret) {
+                                snprintf (msg, sizeof (msg),
+                                          "Unable to get %s", key);
+                                errstr = gf_strdup (msg);
+                                goto out;
+                        }
+
+                        ret =
+                        glusterd_volume_brickinfo_get_by_brick(brick, volinfo,
+                                                               &brickinfo);
+                        if (ret) {
+                                snprintf (msg, sizeof (msg), "Incorrect brick "
+                                          "%s for volume %s", brick, volname);
+                                errstr = gf_strdup (msg);
+                                goto out;
+                        }
+                        if ( !brickinfo->decommissioned ) {
+                                snprintf (msg, sizeof (msg), "Brick %s "
+                                          "is not decommissioned. "
+                                          "Use start or force option",
+                                          brick);
+                                errstr = gf_strdup (msg);
+                                ret = -1;
+                                goto out;
+                        }
+                }
+
                 break;
 
         case GF_OP_CMD_COMMIT_FORCE:
                 break;
         }
-
-        ret = dict_get_int32 (dict, "count", &brick_count);
-        if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "Unable to get brick count");
-                goto out;
-        }
-
         ret = 0;
-        if (volinfo->brick_count == brick_count) {
-                errstr = gf_strdup ("Deleting all the bricks of the "
-                                    "volume is not allowed");
-                ret = -1;
-                goto out;
-        }
 
 out:
         gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);