glusterd: remove-brick:Allow simultaneous removal of multiple subvolumes.

Currently, remove-brick supports removal of only one distributed
stripe/ replica pair at a time. Fix it to support removal of multiple
pairs. This is consistent with add-brick behaviour which supports adding
multiple stripe/replica pairs simultaneously.

Removal is successful irrespective of the order of the bricks given at
the CLI, as long as the bricks are from the same subvolume(s).

Change-Id: I7c11c1235ce07b124155978b9d48d0ea65396103
BUG: 974007
Signed-off-by: Ravishankar N <ravishankar@redhat.com>
Reviewed-on: http://review.gluster.org/5210
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Krishnan Parthasarathi <kparthas@redhat.com>

4 files changed, 144 insertions, 37 deletions

diff --git a/tests/bugs/bug-974007.t b/tests/bugs/bug-974007.t
new file mode 100644
index 00000000000..c8c1c862b33
--- /dev/null
+++ b/tests/bugs/bug-974007.t
@@ -0,0 +1,52 @@
+#!/bin/bash
+
+#Test case: Create a distributed replicate volume, and remove multiple
+#replica pairs in a single remove-brick command.
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+
+cleanup;
+
+#Basic checks
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume info
+
+#Create a 3X2 distributed-replicate volume
+TEST $CLI volume create $V0 replica 2 $H0:$B0/${V0}{1..6};
+TEST $CLI volume start $V0
+
+# Mount FUSE and create files
+TEST glusterfs -s $H0 --volfile-id $V0 $M0
+TEST touch $M0/file{1..10}
+
+# Remove bricks from two sub-volumes to make it a 1x2 vol.
+# Bricks in question are given in a random order but from the same subvols.
+function remove_brick_start_status {
+        $CLI volume remove-brick $V0 \
+        $H0:$B0/${V0}6  $H0:$B0/${V0}1 \
+        $H0:$B0/${V0}2  $H0:$B0/${V0}5 start 2>&1 |grep -oE "success|failed"
+}
+EXPECT "success"  remove_brick_start_status;
+
+# Wait for rebalance to complete
+EXPECT_WITHIN 10 "completed" remove_brick_status_completed_field "$V0" "$H0:$B0/${V0}6 $H0:$B0/${V0}1 $H0:$B0/${V0}2 $H0:$B0/${V0}5"
+
+# Check commit status
+function remove_brick_commit_status {
+        $CLI volume remove-brick $V0 \
+         $H0:$B0/${V0}6  $H0:$B0/${V0}1 \
+         $H0:$B0/${V0}2  $H0:$B0/${V0}5 commit 2>&1 |grep -oE "success|failed"
+}
+EXPECT "success" remove_brick_commit_status;
+
+# Check the volume type
+EXPECT "Replicate" echo `$CLI volume info |grep Type |awk '{print $2}'`
+
+TEST umount $M0
+TEST $CLI volume stop $V0
+TEST $CLI volume delete $V0;
+TEST ! $CLI volume info $V0;
+
+cleanup;
diff --git a/tests/volume.rc b/tests/volume.rc
index 02892d7158b..470fe9a7c3b 100644
--- a/tests/volume.rc
+++ b/tests/volume.rc
@@ -31,7 +31,9 @@ function rebalance_status_field {
 }
 
 function remove_brick_status_completed_field {
-        $CLI volume remove-brick $V0 $H0:$B0/r2d2_{4,5} status | awk '{print $7}' | sed -n 3p
+        local vol=$1
+        local brick_list=$2
+        $CLI volume remove-brick $vol $brick_list status | awk '{print $7}' | sed -n 3p
 }
 
 function get_mount_process_pid {
diff --git a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
index 45927e23583..efb0941ec2c 100644
--- a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
@@ -546,6 +546,72 @@ glusterd_handle_add_brick (rpcsvc_request_t *req)
         return glusterd_big_locked_handler (req, __glusterd_handle_add_brick);
 }
 
+static int
+subvol_matcher_init (int **subvols, int count)
+{
+        int ret = -1;
+
+        *subvols = GF_CALLOC (count, sizeof(int), gf_gld_mt_int);
+        if (*subvols)
+                ret = 0;
+
+        return ret;
+}
+
+static void
+subvol_matcher_update (int *subvols, glusterd_volinfo_t *volinfo,
+                       glusterd_brickinfo_t *brickinfo)
+{
+        glusterd_brickinfo_t *tmp        = NULL;
+        int32_t               sub_volume = 0;
+        int                   pos        = 0;
+
+        list_for_each_entry (tmp, &volinfo->bricks, brick_list) {
+
+                if (strcmp (tmp->hostname, brickinfo->hostname) ||
+                    strcmp (tmp->path, brickinfo->path)) {
+                        pos++;
+                        continue;
+                }
+                gf_log (THIS->name, GF_LOG_DEBUG, LOGSTR_FOUND_BRICK,
+                                        brickinfo->hostname, brickinfo->path,
+                                        volinfo->volname);
+                sub_volume = (pos / volinfo->dist_leaf_count);
+                subvols[sub_volume]++;
+                break;
+        }
+
+}
+
+static int
+subvol_matcher_verify (int *subvols, glusterd_volinfo_t *volinfo, char *err_str,
+                       size_t err_len, char *vol_type)
+{
+        int i = 0;
+        int ret = 0;
+
+       do {
+
+                if (subvols[i] % volinfo->dist_leaf_count == 0) {
+                        continue;
+                } else {
+                        ret = -1;
+                        snprintf (err_str, err_len,
+                                "Bricks not from same subvol for %s", vol_type);
+                        gf_log (THIS->name, GF_LOG_ERROR, "%s", err_str);
+                        break;
+                }
+        } while (++i < volinfo->subvol_count);
+
+        return ret;
+}
+
+static void
+subvol_matcher_destroy (int *subvols)
+{
+        GF_FREE (subvols);
+}
+
 int
 __glusterd_handle_remove_brick (rpcsvc_request_t *req)
 {
@@ -559,10 +625,7 @@ __glusterd_handle_remove_brick (rpcsvc_request_t *req)
         int                       i                = 1;
         glusterd_volinfo_t       *volinfo          = NULL;
         glusterd_brickinfo_t     *brickinfo        = NULL;
-        int32_t                   pos              = 0;
-        int32_t                   sub_volume       = 0;
-        int32_t                   sub_volume_start = 0;
-        int32_t                   sub_volume_end   = 0;
+        int                      *subvols          = NULL;
         glusterd_brickinfo_t     *tmp              = NULL;
         char                      err_str[2048]    = {0};
         gf_cli_rsp                rsp              = {0,};
@@ -738,6 +801,14 @@ __glusterd_handle_remove_brick (rpcsvc_request_t *req)
         }
 
         strcpy (brick_list, " ");
+
+        if ((volinfo->type != GF_CLUSTER_TYPE_NONE) &&
+            (volinfo->subvol_count > 1)) {
+                ret = subvol_matcher_init (&subvols, volinfo->subvol_count);
+                if (ret)
+                        goto out;
+        }
+
         while ( i <= count) {
                 snprintf (key, sizeof (key), "brick%d", i);
                 ret = dict_get_str (dict, key, &brick);
@@ -805,38 +876,18 @@ __glusterd_handle_remove_brick (rpcsvc_request_t *req)
                         goto out;
                 }
 
-                pos = 0;
-                list_for_each_entry (tmp, &volinfo->bricks, brick_list) {
-
-                        if (strcmp (tmp->hostname,brickinfo->hostname) ||
-                            strcmp (tmp->path, brickinfo->path)) {
-                                pos++;
-                                continue;
-                        }
+                /* Find which subvolume the brick belongs to */
+                subvol_matcher_update (subvols, volinfo, brickinfo);
+        }
 
-                        gf_log (this->name, GF_LOG_DEBUG, LOGSTR_FOUND_BRICK,
-                                brickinfo->hostname, brickinfo->path,
-                                volinfo->volname);
-                        if (!sub_volume && (volinfo->dist_leaf_count > 1)) {
-                                sub_volume = (pos / volinfo->dist_leaf_count) + 1;
-                                sub_volume_start = (volinfo->dist_leaf_count *
-                                                    (sub_volume - 1));
-                                sub_volume_end = (volinfo->dist_leaf_count *
-                                                  sub_volume) - 1;
-                        } else {
-                                if (pos < sub_volume_start ||
-                                    pos >sub_volume_end) {
-                                        ret = -1;
-                                        snprintf (err_str, sizeof (err_str),
-                                                  "Bricks not from same subvol "
-                                                  "for %s", vol_type);
-                                        gf_log (this->name, GF_LOG_ERROR,
-                                                "%s", err_str);
-                                        goto out;
-                                }
-                        }
-                        break;
-                }
+        /* Check if the bricks belong to the same subvolumes.*/
+        if ((volinfo->type != GF_CLUSTER_TYPE_NONE) &&
+            (volinfo->subvol_count > 1)) {
+                ret = subvol_matcher_verify (subvols, volinfo,
+                                             err_str, sizeof(err_str),
+                                             vol_type);
+                if (ret)
+                        goto out;
         }
 
         ret = glusterd_op_begin_synctask (req, GD_OP_REMOVE_BRICK, dict);
@@ -859,6 +910,7 @@ out:
         }
 
         GF_FREE (brick_list);
+        subvol_matcher_destroy (subvols);
         free (cli_req.dict.dict_val); //its malloced by xdr
 
         return ret;
diff --git a/xlators/mgmt/glusterd/src/glusterd-mem-types.h b/xlators/mgmt/glusterd/src/glusterd-mem-types.h
index 98216e28ab0..51057c27472 100644
--- a/xlators/mgmt/glusterd/src/glusterd-mem-types.h
+++ b/xlators/mgmt/glusterd/src/glusterd-mem-types.h
@@ -66,7 +66,8 @@ typedef enum gf_gld_mem_types_ {
         gf_gld_mt_hooks_stub_t                  = gf_common_mt_end + 50,
         gf_gld_mt_hooks_priv_t                  = gf_common_mt_end + 51,
         gf_gld_mt_mop_commit_req_t              = gf_common_mt_end + 52,
-        gf_gld_mt_end                           = gf_common_mt_end + 53,
+        gf_gld_mt_int                           = gf_common_mt_end + 53,
+        gf_gld_mt_end                           = gf_common_mt_end + 54,
 } gf_gld_mem_types_t;
 #endif