glusterd: Implementing volume stop in mgmt v3

Change-Id: I8f9c594cf56331d54eb4884335699744685ef20d
fixes: bz#1560441
Signed-off-by: Sanju Rakonde <srakonde@redhat.com>

2 files changed, 66 insertions, 1 deletions

diff --git a/xlators/mgmt/glusterd/src/glusterd-mgmt.c b/xlators/mgmt/glusterd/src/glusterd-mgmt.c
index a1596f0944d..19eefab6f3c 100644
--- a/xlators/mgmt/glusterd/src/glusterd-mgmt.c
+++ b/xlators/mgmt/glusterd/src/glusterd-mgmt.c
@@ -189,6 +189,15 @@ gd_mgmt_v3_pre_validate_fn (glusterd_op_t op, dict_t *dict,
                         goto out;
                 }
                 break;
+        case GD_OP_STOP_VOLUME:
+                ret = glusterd_op_stage_stop_volume (dict, op_errstr);
+                if (ret) {
+                        gf_msg (this->name, GF_LOG_WARNING, 0,
+                                GD_MSG_PRE_VALIDATION_FAIL,
+                                "Volume stop prevalidation failed.");
+                        goto out;
+                }
+                break;
         case GD_OP_TIER_START_STOP:
         case GD_OP_TIER_STATUS:
         case GD_OP_DETACH_TIER_STATUS:
@@ -327,6 +336,17 @@ gd_mgmt_v3_commit_fn (glusterd_op_t op, dict_t *dict,
                         break;
 
                 }
+                case GD_OP_STOP_VOLUME:
+                {
+                        ret = glusterd_op_stop_volume (dict);
+                        if (ret) {
+                                gf_msg (this->name, GF_LOG_ERROR, 0,
+                                        GD_MSG_COMMIT_OP_FAIL,
+                                        "Volume stop commit failed.");
+                                goto out;
+                        }
+                        break;
+                }
                 case GD_OP_RESET_BRICK:
                 {
                         ret = glusterd_op_reset_brick (dict, rsp_dict);
@@ -506,6 +526,34 @@ gd_mgmt_v3_post_validate_fn (glusterd_op_t op, int32_t op_ret, dict_t *dict,
                         }
                         break;
                }
+               case GD_OP_STOP_VOLUME:
+               {
+                        ret = dict_get_str (dict, "volname", &volname);
+                        if (ret) {
+                                gf_msg ("glusterd", GF_LOG_ERROR, 0,
+                                        GD_MSG_DICT_GET_FAILED, "Unable to get"
+                                        " volume name");
+                                goto out;
+                        }
+
+                        ret = glusterd_volinfo_find (volname, &volinfo);
+                        if (ret) {
+                                gf_msg ("glusterd", GF_LOG_ERROR, EINVAL,
+                                        GD_MSG_VOL_NOT_FOUND, "Unable to "
+                                        "allocate memory");
+                                goto out;
+                        }
+                        break;
+
+                        if (volinfo->type == GF_CLUSTER_TYPE_TIER) {
+                                svc = &(volinfo->tierd.svc);
+                                ret = svc->manager (svc, volinfo,
+                                                    PROC_START_NO_WAIT);
+                                if (ret)
+                                        goto out;
+                        }
+                        break;
+               }
                case GD_OP_ADD_TIER_BRICK:
                {
                         ret = dict_get_str (dict, "volname", &volname);
@@ -826,6 +874,7 @@ glusterd_pre_validate_aggr_rsp_dict (glusterd_op_t op,
                                 "response dictionaries.");
                         goto out;
                 }
+        case GD_OP_STOP_VOLUME:
         case GD_OP_TIER_STATUS:
         case GD_OP_DETACH_TIER_STATUS:
         case GD_OP_TIER_START_STOP:
@@ -1136,6 +1185,7 @@ glusterd_mgmt_v3_build_payload (dict_t **req, char **op_errstr, dict_t *dict,
                 dict_copy (dict, req_dict);
                 break;
         case GD_OP_START_VOLUME:
+        case GD_OP_STOP_VOLUME:
         case GD_OP_ADD_BRICK:
         case GD_OP_REPLACE_BRICK:
         case GD_OP_RESET_BRICK:
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
index 216b403867c..4001230358c 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
@@ -588,10 +588,13 @@ __glusterd_handle_cli_stop_volume (rpcsvc_request_t *req)
         glusterd_op_t                   cli_op = GD_OP_STOP_VOLUME;
         xlator_t                        *this = NULL;
         char                            err_str[2048] = {0,};
+        glusterd_conf_t                 *conf = NULL;
 
         this = THIS;
         GF_ASSERT (this);
         GF_ASSERT (req);
+        conf = this->private;
+        GF_ASSERT (conf);
 
         ret = xdr_to_generic (req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
         if (ret < 0) {
@@ -633,7 +636,18 @@ __glusterd_handle_cli_stop_volume (rpcsvc_request_t *req)
         gf_msg_debug (this->name, 0, "Received stop vol req "
                 "for volume %s", dup_volname);
 
-        ret = glusterd_op_begin_synctask (req, GD_OP_STOP_VOLUME, dict);
+        if (conf->op_version <= GD_OP_VERSION_4_1_0) {
+                gf_msg_debug (this->name, 0, "The cluster is operating at "
+                          "version less than or equal to %d. Volume start "
+                          "falling back to syncop framework.",
+                          GD_OP_VERSION_4_1_0);
+                ret = glusterd_op_begin_synctask (req, GD_OP_STOP_VOLUME,
+                                                  dict);
+        } else {
+                ret = glusterd_mgmt_v3_initiate_all_phases (req,
+                                                            GD_OP_STOP_VOLUME,
+                                                            dict);
+        }
 
 out:
         free (cli_req.dict.dict_val); //its malloced by xdr
@@ -2802,6 +2816,7 @@ glusterd_op_stop_volume (dict_t *dict)
                 goto out;
         }
 out:
+        gf_msg_trace (this->name, 0, "returning %d ", ret);
         return ret;
 }