glusterd, cli: Task id's for async tasks

This patch introduces task-id's for async tasks like rebalance, remove-brick and
replace-brick. An id is generated for each task when it is started and displayed
to the user in cli output. The status of running tasks is also included in the
output of "volume status" along with its id, so that a user can easily track the
progress of an async task.

Also,
 * added tests for this feature into the regression test suite.
 * added a python script for creating files, 'create-files.py', courtesy
   Vijaykumar Koppad (vkoppad@redhat.com) into the test suite.

Change-Id: Ib0c0d12e0d6c8f72ace48d303d7ff3102157e876
BUG: 857330
Signed-off-by: Kaushal M <kaushal@redhat.com>
Reviewed-on: http://review.gluster.org/3942
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Anand Avati <avati@redhat.com>

2 files changed, 269 insertions, 45 deletions

diff --git a/cli/src/cli-rpc-ops.c b/cli/src/cli-rpc-ops.c
index e772137ed3e..5b8abcdbc3c 100644
--- a/cli/src/cli-rpc-ops.c
+++ b/cli/src/cli-rpc-ops.c
@@ -1140,6 +1140,7 @@ gf_cli_defrag_volume_cbk (struct rpc_req *req, struct iovec *iov,
         uint64_t                 failures = 0;
         double                   elapsed = 0;
         char                    *size_str = NULL;
+        char                    *task_id_str = NULL;
 
         if (-1 == req->rpc_status) {
                 goto out;
@@ -1189,15 +1190,24 @@ gf_cli_defrag_volume_cbk (struct rpc_req *req, struct iovec *iov,
                 }
         }
 
-        if (!((cmd == GF_DEFRAG_CMD_STOP) || (cmd == GF_DEFRAG_CMD_STATUS))) {
-                /* All other possibility is about starting a volume */
-                if (rsp.op_ret && strcmp (rsp.op_errstr, ""))
+        if (!((cmd == GF_DEFRAG_CMD_STOP) || (cmd == GF_DEFRAG_CMD_STATUS)) &&
+             !(global_state->mode & GLUSTER_MODE_XML)) {
+                /* All other possibilites are about starting a rebalance */
+                ret = dict_get_str (dict, GF_REBALANCE_TID_KEY, &task_id_str);
+                if (rsp.op_ret && strcmp (rsp.op_errstr, "")) {
                         snprintf (msg, sizeof (msg), "%s", rsp.op_errstr);
-                else
-                        snprintf (msg, sizeof (msg),
-                                  "Starting rebalance on volume %s has been %s",
-                                  volname, (rsp.op_ret) ? "unsuccessful":
-                                  "successful");
+                } else {
+                        if (!rsp.op_ret) {
+                                snprintf (msg, sizeof (msg),
+                                          "Starting rebalance on volume %s has "
+                                          "been successful.\nID: %s", volname,
+                                          task_id_str);
+                        } else {
+                                snprintf (msg, sizeof (msg),
+                                          "Starting rebalance on volume %s has "
+                                          "been unsuccessful.", volname);
+                        }
+                }
                 goto done;
         }
 
@@ -1739,6 +1749,8 @@ gf_cli_remove_brick_cbk (struct rpc_req *req, struct iovec *iov,
         char                           *cmd_str = "unknown";
         cli_local_t                    *local = NULL;
         call_frame_t                   *frame = NULL;
+        char                           *task_id_str = NULL;
+        dict_t                         *rsp_dict = NULL;
 
         if (-1 == req->rpc_status) {
                 goto out;
@@ -1759,10 +1771,32 @@ gf_cli_remove_brick_cbk (struct rpc_req *req, struct iovec *iov,
                  goto out;
         }
 
-        switch (cmd) {
+        if (rsp.dict.dict_len) {
+                rsp_dict = dict_new ();
+                if (!rsp_dict) {
+                        ret = -1;
+                        goto out;
+                }
+
+                ret = dict_unserialize (rsp.dict.dict_val, rsp.dict.dict_len,
+                                        &rsp_dict);
+                if (ret) {
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "Failed to unserialize rsp_dict");
+                        goto out;
+                }
+        }
 
+        switch (cmd) {
         case GF_OP_CMD_START:
                 cmd_str = "start";
+
+                ret = dict_get_str (rsp_dict, GF_REMOVE_BRICK_TID_KEY, &task_id_str);
+                if (ret) {
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "remove-brick-id is not present in dict");
+                        goto out;
+                }
                 break;
         case GF_OP_CMD_COMMIT:
                 cmd_str = "commit";
@@ -1784,7 +1818,7 @@ gf_cli_remove_brick_cbk (struct rpc_req *req, struct iovec *iov,
                           (rsp.op_ret) ? "unsuccessful": "successful");
 
         if (global_state->mode & GLUSTER_MODE_XML) {
-                ret = cli_xml_output_vol_remove_brick (_gf_false, NULL,
+                ret = cli_xml_output_vol_remove_brick (_gf_false, rsp_dict,
                                                        rsp.op_ret, rsp.op_errno,
                                                        rsp.op_errstr);
                 if (ret)
@@ -1793,10 +1827,14 @@ gf_cli_remove_brick_cbk (struct rpc_req *req, struct iovec *iov,
                 goto out;
         }
 
-        if (rsp.op_ret)
-                cli_err ("volume remove-brick: failed: %s", rsp.op_errstr);
-        else
-                cli_out ("volume remove-brick: success");
+        if (rsp.op_ret) {
+                cli_err ("volume remove-brick %s: failed: %s", cmd_str,
+                         rsp.op_errstr);
+        } else {
+                cli_out ("volume remove-brick %s: success", cmd_str);
+                if (GF_OP_CMD_START == cmd)
+                        cli_out ("ID: %s", task_id_str);
+        }
 
         ret = rsp.op_ret;
 
@@ -1825,7 +1863,8 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
         gf1_cli_replace_op               replace_op       = 0;
         char                            *rb_operation_str = NULL;
         dict_t                          *rsp_dict         = NULL;
-        char                             msg[1024]         = {0,};
+        char                             msg[1024]        = {0,};
+        char                            *task_id_str      = NULL;
 
         if (-1 == req->rpc_status) {
                 goto out;
@@ -1850,33 +1889,48 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
                 goto out;
         }
 
+        if (rsp.dict.dict_len) {
+                /* Unserialize the dictionary */
+                rsp_dict  = dict_new ();
+
+                ret = dict_unserialize (rsp.dict.dict_val,
+                                rsp.dict.dict_len,
+                                &rsp_dict);
+                if (ret < 0) {
+                        gf_log ("glusterd", GF_LOG_ERROR,
+                                        "failed to "
+                                        "unserialize rsp buffer to dictionary");
+                        goto out;
+                }
+        }
+
         switch (replace_op) {
         case GF_REPLACE_OP_START:
-                if (rsp.op_ret)
-                        rb_operation_str = "replace-brick failed to start";
-                else
-                        rb_operation_str = "replace-brick started successfully";
+                if (rsp.op_ret) {
+                        rb_operation_str = gf_strdup ("replace-brick failed to"
+                                                      " start");
+                } else {
+                        ret = dict_get_str (rsp_dict, GF_REPLACE_BRICK_TID_KEY,
+                                            &task_id_str);
+                        if (ret) {
+                                gf_log ("cli", GF_LOG_ERROR, "Failed to get "
+                                        "\"replace-brick-id\" from dict");
+                                goto out;
+                        }
+                        ret = gf_asprintf (&rb_operation_str,
+                                           "replace-brick started successfully"
+                                           "\nID: %s", task_id_str);
+                        if (ret < 0)
+                                goto out;
+                }
                 break;
 
         case GF_REPLACE_OP_STATUS:
 
-                if (rsp.op_ret || ret)
-                        rb_operation_str = "replace-brick status unknown";
-                else {
-                        if (rsp.dict.dict_len) {
-                                /* Unserialize the dictionary */
-                                rsp_dict  = dict_new ();
-
-                                ret = dict_unserialize (rsp.dict.dict_val,
-                                                        rsp.dict.dict_len,
-                                                        &rsp_dict);
-                                if (ret < 0) {
-                                        gf_log ("glusterd", GF_LOG_ERROR,
-                                                "failed to "
-                                                "unserialize req-buffer to dictionary");
-                                        goto out;
-                                }
-                        }
+                if (rsp.op_ret || ret) {
+                        rb_operation_str = gf_strdup ("replace-brick status "
+                                                      "unknown");
+                } else {
                         ret = dict_get_str (rsp_dict, "status-reply",
                                             &status_reply);
                         if (ret) {
@@ -1885,23 +1939,27 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
                                 goto out;
                         }
 
-                        rb_operation_str = status_reply;
+                        rb_operation_str = gf_strdup (status_reply);
                 }
 
                 break;
 
         case GF_REPLACE_OP_PAUSE:
                 if (rsp.op_ret)
-                        rb_operation_str = "replace-brick pause failed";
+                        rb_operation_str = gf_strdup ("replace-brick pause "
+                                                      "failed");
                 else
-                        rb_operation_str = "replace-brick paused successfully";
+                        rb_operation_str = gf_strdup ("replace-brick paused "
+                                                      "successfully");
                 break;
 
         case GF_REPLACE_OP_ABORT:
                 if (rsp.op_ret)
-                        rb_operation_str = "replace-brick abort failed";
+                        rb_operation_str = gf_strdup ("replace-brick abort "
+                                                      "failed");
                 else
-                        rb_operation_str = "replace-brick aborted successfully";
+                        rb_operation_str = gf_strdup ("replace-brick aborted "
+                                                      "successfully");
                 break;
 
         case GF_REPLACE_OP_COMMIT:
@@ -1922,9 +1980,11 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
 
 
                 if (rsp.op_ret || ret)
-                        rb_operation_str = "replace-brick commit failed";
+                        rb_operation_str = gf_strdup ("replace-brick commit "
+                                                      "failed");
                 else
-                        rb_operation_str = "replace-brick commit successful";
+                        rb_operation_str = gf_strdup ("replace-brick commit "
+                                                      "successful");
 
                 break;
 
@@ -1935,7 +1995,7 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
         }
 
         if (rsp.op_ret && (strcmp (rsp.op_errstr, ""))) {
-                rb_operation_str = rsp.op_errstr;
+                rb_operation_str = gf_strdup (rsp.op_errstr);
         }
 
         gf_log ("cli", GF_LOG_INFO, "Received resp to replace brick");
@@ -1959,6 +2019,17 @@ gf_cli_replace_brick_cbk (struct rpc_req *req, struct iovec *iov,
         ret = rsp.op_ret;
 
 out:
+        if (frame)
+                frame->local = NULL;
+
+        if (local) {
+                dict_unref (local->dict);
+                cli_local_wipe (local);
+        }
+
+        if (rb_operation_str)
+                GF_FREE (rb_operation_str);
+
         cli_cmd_broadcast_response (ret);
         free (rsp.dict.dict_val);
         if (rsp_dict)
@@ -3030,7 +3101,7 @@ gf_cli_remove_brick (call_frame_t *frame, xlator_t *this,
                                        GLUSTER_CLI_REMOVE_BRICK, this,
                                        cli_rpc_prog, NULL);
         } else {
-                /* Need rebalance status to e sent :-) */
+                /* Need rebalance status to be sent :-) */
                 req_dict = dict_new ();
                 if (!req_dict) {
                         ret = -1;
@@ -5272,6 +5343,55 @@ out:
         return;
 }
 
+
+static void
+cli_print_volume_tasks (dict_t *dict) {
+        int             ret = -1;
+        int             tasks = 0;
+        char            *op = 0;
+        char            *task_id_str = NULL;
+        int             status = 0;
+        char            key[1024] = {0,};
+        int             i = 0;
+
+        ret = dict_get_int32 (dict, "tasks", &tasks);
+        if (ret) {
+                gf_log ("cli", GF_LOG_ERROR,
+                        "Failed to get tasks count");
+                return;
+        }
+
+        if (tasks == 0) {
+                cli_out ("There are no active volume tasks");
+                return;
+        }
+
+        cli_out ("%15s%40s%15s", "Task", "ID", "Status");
+        cli_out ("%15s%40s%15s", "----", "--", "------");
+        for (i = 0; i < tasks; i++) {
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.type", i);
+                ret = dict_get_str(dict, key, &op);
+                if (ret)
+                        return;
+
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.id", i);
+                ret = dict_get_str (dict, key, &task_id_str);
+                if (ret)
+                        return;
+
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.status", i);
+                ret = dict_get_int32 (dict, key, &status);
+                if (ret)
+                        return;
+
+                cli_out ("%15s%40s%15d", op, task_id_str, status);
+        }
+
+}
+
 static int
 gf_cli_status_cbk (struct rpc_req *req, struct iovec *iov,
                       int count, void *myframe)
@@ -5518,6 +5638,9 @@ gf_cli_status_cbk (struct rpc_req *req, struct iovec *iov,
                 }
         }
         cli_out (" ");
+
+        if ((cmd & GF_CLI_STATUS_MASK) == GF_CLI_STATUS_NONE)
+                cli_print_volume_tasks (dict);
 cont:
         ret = rsp.op_ret;
 
diff --git a/cli/src/cli-xml-output.c b/cli/src/cli-xml-output.c
index ffe32caccdd..0765dc26a2e 100644
--- a/cli/src/cli-xml-output.c
+++ b/cli/src/cli-xml-output.c
@@ -1381,6 +1381,74 @@ out:
 }
 
 int
+cli_xml_output_vol_status_tasks (cli_local_t *local, dict_t *dict) {
+        int                     ret = -1;
+        char                    *task_type = NULL;
+        char                    *task_id_str = NULL;
+        int                     status = 0;
+        int                     tasks = 0;
+        char                    key[1024] = {0,};
+        int                     i = 0;
+
+        /* <tasks> */
+        ret = xmlTextWriterStartElement (local->writer, (xmlChar *)"tasks");
+        XML_RET_CHECK_AND_GOTO (ret, out);
+
+        ret = dict_get_int32 (dict, "tasks", &tasks);
+        if (ret)
+                goto out;
+
+        for (i = 0; i < tasks; i++) {
+                /* <task> */
+                ret = xmlTextWriterStartElement (local->writer,
+                                                 (xmlChar *)"task");
+                XML_RET_CHECK_AND_GOTO (ret, out);
+
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.type", i);
+                ret = dict_get_str (dict, key, &task_type);
+                if (ret)
+                        goto out;
+                ret = xmlTextWriterWriteFormatElement (local->writer,
+                                                       (xmlChar *)"type",
+                                                       "%s", task_type);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.id", i);
+                ret = dict_get_str (dict, key, &task_id_str);
+                if (ret)
+                        goto out;
+                ret = xmlTextWriterWriteFormatElement (local->writer,
+                                                       (xmlChar *)"id",
+                                                       "%s", task_id_str);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+
+                memset (key, 0, sizeof (key));
+                snprintf (key, sizeof (key), "task%d.status", i);
+                ret = dict_get_int32 (dict, key, &status);
+                if (ret)
+                        goto out;
+                ret = xmlTextWriterWriteFormatElement (local->writer,
+                                                       (xmlChar *)"status",
+                                                       "%d", status);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+
+                /* </task> */
+                ret = xmlTextWriterEndElement (local->writer);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+        }
+
+        /* </tasks> */
+        ret = xmlTextWriterEndElement (local->writer);
+
+
+out:
+        gf_log ("cli", GF_LOG_DEBUG, "Returning %d", ret);
+        return ret;
+}
+
+int
 cli_xml_output_vol_status (cli_local_t *local, dict_t *dict)
 {
 #if (HAVE_LIB_XML)
@@ -1507,6 +1575,12 @@ cli_xml_output_vol_status (cli_local_t *local, dict_t *dict)
                 XML_RET_CHECK_AND_GOTO (ret, out);
         }
 
+        if ((cmd & GF_CLI_STATUS_MASK) == GF_CLI_STATUS_NONE) {
+                ret = cli_xml_output_vol_status_tasks (local, dict);
+                if (ret)
+                        goto out;
+        }
+
         /* </volume> */
         ret = xmlTextWriterEndElement (local->writer);
         XML_RET_CHECK_AND_GOTO (ret, out);
@@ -2913,6 +2987,7 @@ cli_xml_output_vol_rebalance (gf_cli_defrag_type op, dict_t *dict, int op_ret,
         int                     ret = -1;
         xmlTextWriterPtr        writer = NULL;
         xmlBufferPtr            buf = NULL;
+        char                    *task_id_str = NULL;
 
         ret = cli_begin_xml_output (&writer, &buf);
         if (ret)
@@ -2926,6 +3001,14 @@ cli_xml_output_vol_rebalance (gf_cli_defrag_type op, dict_t *dict, int op_ret,
         ret = xmlTextWriterStartElement (writer, (xmlChar *)"volRebalance");
         XML_RET_CHECK_AND_GOTO (ret, out);
 
+        ret = dict_get_str (dict, GF_REBALANCE_TID_KEY, &task_id_str);
+        if (ret == 0) {
+                ret = xmlTextWriterWriteFormatElement (writer,
+                                                       (xmlChar *)"task-id",
+                                                       "%s", task_id_str);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+        }
+
         ret = xmlTextWriterWriteFormatElement (writer, (xmlChar *)"op",
                                                "%d", op);
         XML_RET_CHECK_AND_GOTO (ret, out);
@@ -2959,6 +3042,7 @@ cli_xml_output_vol_remove_brick (gf_boolean_t status_op, dict_t *dict,
         int                     ret = -1;
         xmlTextWriterPtr        writer = NULL;
         xmlBufferPtr            buf = NULL;
+        char                    *task_id_str = NULL;
 
         ret = cli_begin_xml_output (&writer, &buf);
         if (ret)
@@ -2972,6 +3056,14 @@ cli_xml_output_vol_remove_brick (gf_boolean_t status_op, dict_t *dict,
         ret = xmlTextWriterStartElement (writer, (xmlChar *)"volRemoveBrick");
         XML_RET_CHECK_AND_GOTO (ret, out);
 
+        ret = dict_get_str (dict, GF_REMOVE_BRICK_TID_KEY, &task_id_str);
+        if (ret == 0) {
+                ret = xmlTextWriterWriteFormatElement (writer,
+                                                       (xmlChar *)"task-id",
+                                                       "%s", task_id_str);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+        }
+
         if (status_op) {
                 ret = cli_xml_output_vol_rebalance_status (writer, dict);
                 if (ret)
@@ -3002,6 +3094,7 @@ cli_xml_output_vol_replace_brick (gf1_cli_replace_op op, dict_t *dict,
         int                     status = 0;
         uint64_t                files = 0;
         char                    *current_file = 0;
+        char                    *task_id_str = NULL;
         xmlTextWriterPtr        writer = NULL;
         xmlBufferPtr            buf = NULL;
 
@@ -3017,6 +3110,14 @@ cli_xml_output_vol_replace_brick (gf1_cli_replace_op op, dict_t *dict,
         ret = xmlTextWriterStartElement (writer, (xmlChar *)"volReplaceBrick");
         XML_RET_CHECK_AND_GOTO (ret, out);
 
+        ret = dict_get_str (dict, GF_REPLACE_BRICK_TID_KEY, &task_id_str);
+        if (ret == 0) {
+                ret = xmlTextWriterWriteFormatElement (writer,
+                                                       (xmlChar *)"task-id",
+                                                       "%s", task_id_str);
+                XML_RET_CHECK_AND_GOTO (ret, out);
+        }
+
         ret = xmlTextWriterWriteFormatElement (writer, (xmlChar *)"op",
                                                "%d", op);
         XML_RET_CHECK_AND_GOTO (ret, out);