rebalance: handshake_event_notify to make fsd talk to glusterd

Event_notify can be used by others to communicate with glusterd.
A cbk event is also added for future use.

req has a op, and dict.
rsp has op_ret, op_errno, and dict.

With this, rebalance process can update the status before
exiting.

Signed-off-by: shishir gowda <shishirng@gluster.com>
Change-Id: If5c0ec00514eb3a109a790b2ea273317611e4562
BUG: 807126
Reviewed-on: http://review.gluster.com/3013
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Amar Tumballi <amarts@redhat.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

13 files changed, 311 insertions, 48 deletions

diff --git a/glusterfsd/src/glusterfsd-mgmt.c b/glusterfsd/src/glusterfsd-mgmt.c
index dac92d67f..4f01871b2 100644
--- a/glusterfsd/src/glusterfsd-mgmt.c
+++ b/glusterfsd/src/glusterfsd-mgmt.c
@@ -67,6 +67,12 @@ mgmt_cbk_spec (void *data)
         return 0;
 }
 
+
+int
+mgmt_cbk_event (void *data)
+{
+        return 0;
+}
 struct iobuf *
 glusterfs_serialize_reply (rpcsvc_request_t *req, void *arg,
                            struct iovec *outmsg, xdrproc_t xdrproc)
@@ -1246,6 +1252,8 @@ out:
 
 rpcclnt_cb_actor_t gluster_cbk_actors[] = {
         [GF_CBK_FETCHSPEC] = {"FETCHSPEC", GF_CBK_FETCHSPEC, mgmt_cbk_spec },
+        [GF_CBK_EVENT_NOTIFY] = {"EVENTNOTIFY", GF_CBK_EVENT_NOTIFY,
+                                 mgmt_cbk_event},
 };
 
 
@@ -1279,6 +1287,7 @@ char *clnt_handshake_procs[GF_HNDSK_MAXVALUE] = {
         [GF_HNDSK_SETVOLUME]    = "SETVOLUME",
         [GF_HNDSK_GETSPEC]      = "GETSPEC",
         [GF_HNDSK_PING]         = "PING",
+        [GF_HNDSK_EVENT_NOTIFY] = "EVENTNOTIFY",
 };
 
 rpc_clnt_prog_t clnt_handshake_prog = {
@@ -1639,6 +1648,77 @@ glusterfs_volfile_fetch (glusterfs_ctx_t *ctx)
         return ret;
 }
 
+int32_t
+mgmt_event_notify_cbk (struct rpc_req *req, struct iovec *iov, int count,
+                  void *myframe)
+{
+        gf_event_notify_rsp      rsp   = {0,};
+        call_frame_t            *frame = NULL;
+        glusterfs_ctx_t         *ctx = NULL;
+        int                      ret   = 0;
+
+        frame = myframe;
+        ctx = frame->this->ctx;
+
+        if (-1 == req->rpc_status) {
+                ret = -1;
+                goto out;
+        }
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_event_notify_rsp);
+        if (ret < 0) {
+                gf_log (frame->this->name, GF_LOG_ERROR, "XDR decoding error");
+                ret   = -1;
+                goto out;
+        }
+
+        if (-1 == rsp.op_ret) {
+                gf_log (frame->this->name, GF_LOG_ERROR,
+                        "failed to get the rsp from server");
+                ret = -1;
+                goto out;
+        }
+out:
+        if (rsp.dict.dict_val)
+                free (rsp.dict.dict_val); //malloced by xdr
+        return ret;
+
+}
+int32_t
+glusterfs_rebalance_event_notify (dict_t *dict)
+{
+        glusterfs_ctx_t         *ctx = NULL;
+        gf_event_notify_req      req = {0,};
+        int32_t                  ret = -1;
+        cmd_args_t              *cmd_args = NULL;
+        call_frame_t            *frame = NULL;
+
+        ctx = glusterfs_ctx_get ();
+        cmd_args = &ctx->cmd_args;
+
+        frame = create_frame (THIS, ctx->pool);
+
+        req.op = GF_EN_DEFRAG_STATUS;
+
+        if (dict) {
+                ret = dict_set_str (dict, "volname", cmd_args->volfile_id);
+                if (ret)
+                        gf_log ("", GF_LOG_ERROR, "failed to set volname");
+
+                ret = dict_allocate_and_serialize (dict, &req.dict.dict_val,
+                                                (size_t *)&req.dict.dict_len);
+        }
+
+        ret = mgmt_submit_request (&req, frame, ctx, &clnt_handshake_prog,
+                                   GF_HNDSK_EVENT_NOTIFY, NULL,
+                                   (xdrproc_t)xdr_gf_event_notify_req);
+
+        if (req.dict.dict_val)
+                GF_FREE (req.dict.dict_val);
+
+        STACK_DESTROY (frame->root);
+        return ret;
+}
 
 static int
 mgmt_rpc_notify (struct rpc_clnt *rpc, void *mydata, rpc_clnt_event_t event,
diff --git a/libglusterfs/src/xlator.h b/libglusterfs/src/xlator.h
index bad328f6c..2663b041a 100644
--- a/libglusterfs/src/xlator.h
+++ b/libglusterfs/src/xlator.h
@@ -881,4 +881,5 @@ int is_gf_log_command (xlator_t *trans, const char *name, char *value);
 int glusterd_check_log_level (const char *value);
 int xlator_volopt_dynload (char *xlator_type, void **dl_handle,
                            volume_opt_list_t *vol_opt_handle);
+int32_t glusterfs_rebalance_event_notify (dict_t *dict);
 #endif /* _XLATOR_H */
diff --git a/rpc/rpc-lib/src/protocol-common.h b/rpc/rpc-lib/src/protocol-common.h
index cccd1e7f8..ceb87955c 100644
--- a/rpc/rpc-lib/src/protocol-common.h
+++ b/rpc/rpc-lib/src/protocol-common.h
@@ -74,6 +74,7 @@ enum gf_handshake_procnum {
         GF_HNDSK_GETSPEC,
         GF_HNDSK_PING,
         GF_HNDSK_SET_LK_VER,
+        GF_HNDSK_EVENT_NOTIFY,
         GF_HNDSK_MAXVALUE,
 };
 
@@ -118,6 +119,7 @@ enum gf_cbk_procnum {
         GF_CBK_NULL = 0,
         GF_CBK_FETCHSPEC,
         GF_CBK_INO_FLUSH,
+        GF_CBK_EVENT_NOTIFY,
         GF_CBK_MAXVALUE,
 };
 
@@ -200,8 +202,14 @@ typedef enum {
         GF_AFR_OP_HEAL_FAILED_FILES,
         GF_AFR_OP_SPLIT_BRAIN_FILES
 } gf_xl_afr_op_t ;
+
+enum gf_hdsk_event_notify_op {
+        GF_EN_DEFRAG_STATUS,
+        GF_EN_MAX,
+};
+
 #define GLUSTER_HNDSK_PROGRAM    14398633 /* Completely random */
-#define GLUSTER_HNDSK_VERSION    1   /* 0.0.1 */
+#define GLUSTER_HNDSK_VERSION    2   /* 0.0.1 */
 
 #define GLUSTER_PMAP_PROGRAM     34123456
 #define GLUSTER_PMAP_VERSION     1
diff --git a/rpc/xdr/src/glusterfs3-xdr.c b/rpc/xdr/src/glusterfs3-xdr.c
index 5dffda3e0..dd8281ee2 100644
--- a/rpc/xdr/src/glusterfs3-xdr.c
+++ b/rpc/xdr/src/glusterfs3-xdr.c
@@ -1892,3 +1892,31 @@ xdr_gf_set_lk_ver_req (XDR *xdrs, gf_set_lk_ver_req *objp)
 		 return FALSE;
 	return TRUE;
 }
+
+bool_t
+xdr_gf_event_notify_req (XDR *xdrs, gf_event_notify_req *objp)
+{
+	register int32_t *buf;
+        buf = NULL;
+
+	 if (!xdr_int (xdrs, &objp->op))
+		 return FALSE;
+	 if (!xdr_bytes (xdrs, (char **)&objp->dict.dict_val, (u_int *) &objp->dict.dict_len, ~0))
+		 return FALSE;
+	return TRUE;
+}
+
+bool_t
+xdr_gf_event_notify_rsp (XDR *xdrs, gf_event_notify_rsp *objp)
+{
+	register int32_t *buf;
+        buf = NULL;
+
+	 if (!xdr_int (xdrs, &objp->op_ret))
+		 return FALSE;
+	 if (!xdr_int (xdrs, &objp->op_errno))
+		 return FALSE;
+	 if (!xdr_bytes (xdrs, (char **)&objp->dict.dict_val, (u_int *) &objp->dict.dict_len, ~0))
+		 return FALSE;
+	return TRUE;
+}
diff --git a/rpc/xdr/src/glusterfs3-xdr.h b/rpc/xdr/src/glusterfs3-xdr.h
index 124b21e22..ed29deea2 100644
--- a/rpc/xdr/src/glusterfs3-xdr.h
+++ b/rpc/xdr/src/glusterfs3-xdr.h
@@ -1081,6 +1081,25 @@ struct gf_set_lk_ver_req {
 };
 typedef struct gf_set_lk_ver_req gf_set_lk_ver_req;
 
+struct gf_event_notify_req {
+	int op;
+	struct {
+		u_int dict_len;
+		char *dict_val;
+	} dict;
+};
+typedef struct gf_event_notify_req gf_event_notify_req;
+
+struct gf_event_notify_rsp {
+	int op_ret;
+	int op_errno;
+	struct {
+		u_int dict_len;
+		char *dict_val;
+	} dict;
+};
+typedef struct gf_event_notify_rsp gf_event_notify_rsp;
+
 /* the xdr functions */
 
 #if defined(__STDC__) || defined(__cplusplus)
@@ -1172,6 +1191,8 @@ extern  bool_t xdr_gfs3_dirplist (XDR *, gfs3_dirplist*);
 extern  bool_t xdr_gfs3_readdirp_rsp (XDR *, gfs3_readdirp_rsp*);
 extern  bool_t xdr_gf_set_lk_ver_rsp (XDR *, gf_set_lk_ver_rsp*);
 extern  bool_t xdr_gf_set_lk_ver_req (XDR *, gf_set_lk_ver_req*);
+extern  bool_t xdr_gf_event_notify_req (XDR *, gf_event_notify_req*);
+extern  bool_t xdr_gf_event_notify_rsp (XDR *, gf_event_notify_rsp*);
 
 #else /* K&R C */
 extern bool_t xdr_gf_statfs ();
@@ -1262,6 +1283,8 @@ extern bool_t xdr_gfs3_dirplist ();
 extern bool_t xdr_gfs3_readdirp_rsp ();
 extern bool_t xdr_gf_set_lk_ver_rsp ();
 extern bool_t xdr_gf_set_lk_ver_req ();
+extern bool_t xdr_gf_event_notify_req ();
+extern bool_t xdr_gf_event_notify_rsp ();
 
 #endif /* K&R C */
 
diff --git a/rpc/xdr/src/glusterfs3-xdr.x b/rpc/xdr/src/glusterfs3-xdr.x
index 64c2d2061..ad261423d 100644
--- a/rpc/xdr/src/glusterfs3-xdr.x
+++ b/rpc/xdr/src/glusterfs3-xdr.x
@@ -683,3 +683,14 @@ struct gf_set_lk_ver_req {
        string uid<>;
        int lk_ver;
 };
+
+struct gf_event_notify_req {
+	int op;
+	opaque dict<>;
+};
+
+struct gf_event_notify_rsp {
+	int op_ret;
+	int op_errno;
+	opaque dict<>;
+};
diff --git a/xlators/cluster/dht/src/dht-rebalance.c b/xlators/cluster/dht/src/dht-rebalance.c
index 8b32b3f69..b3787f5cb 100644
--- a/xlators/cluster/dht/src/dht-rebalance.c
+++ b/xlators/cluster/dht/src/dht-rebalance.c
@@ -24,6 +24,7 @@
 #endif
 
 #include "dht-common.h"
+#include "xlator.h"
 
 #define GF_DISK_SECTOR_SIZE             512
 #define DHT_REBALANCE_PID               4242 /* Change it if required */
@@ -1414,6 +1415,7 @@ gf_defrag_start_crawl (void *data)
         struct iatt              parent = {0,};
         dict_t                  *fix_layout = NULL;
         dict_t                  *migrate_data = NULL;
+        dict_t                  *status = NULL;
 
         this = data;
         if (!this)
@@ -1482,7 +1484,11 @@ gf_defrag_start_crawl (void *data)
 out:
         LOCK (&defrag->lock);
         {
-                gf_defrag_status_get (defrag, NULL);
+                status = dict_new ();
+                gf_defrag_status_get (defrag, status);
+                glusterfs_rebalance_event_notify (status);
+                if (status)
+                        dict_unref (status);
                 defrag->is_exiting = 1;
         }
         UNLOCK (&defrag->lock);
diff --git a/xlators/mgmt/glusterd/src/glusterd-handshake.c b/xlators/mgmt/glusterd/src/glusterd-handshake.c
index d80cacc8a..ce61ac8bd 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handshake.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handshake.c
@@ -221,10 +221,71 @@ fail:
         return 0;
 }
 
+int32_t
+server_event_notify (rpcsvc_request_t *req)
+{
+        int32_t                 ret             = -1;
+        int32_t                 op_errno        =  0;
+        gf_event_notify_req     args            = {0,};
+        gf_event_notify_rsp     rsp             = {0,};
+        dict_t                 *dict            = NULL;
+        gf_boolean_t            need_rsp        = _gf_true;
 
+        if (!xdr_to_generic (req->msg[0], &args,
+            (xdrproc_t)xdr_gf_event_notify_req)) {
+                req->rpc_err = GARBAGE_ARGS;
+                goto fail;
+        }
+
+        if (args.dict.dict_len) {
+                dict = dict_new ();
+                if (!dict)
+                        return ret;
+                ret = dict_unserialize (args.dict.dict_val,
+                                        args.dict.dict_len, &dict);
+                if (ret) {
+                        gf_log ("", GF_LOG_ERROR, "Failed to unserialize req");
+                        goto fail;
+                }
+        }
+
+        switch (args.op) {
+        case GF_EN_DEFRAG_STATUS:
+                gf_log ("", GF_LOG_INFO,
+                        "recieved defrag status updated");
+                if (dict) {
+                        glusterd_defrag_event_notify_handle (dict);
+                        need_rsp = _gf_false;
+                }
+                break;
+        default:
+                gf_log ("", GF_LOG_ERROR, "Unkown op recieved in in event "
+                        "notify");
+                ret = -1;
+                break;
+        }
+
+fail:
+        rsp.op_ret   = ret;
+
+        if (op_errno)
+                rsp.op_errno = gf_errno_to_error (op_errno);
+
+        if (need_rsp)
+                glusterd_submit_reply (req, &rsp, NULL, 0, NULL,
+                                       (xdrproc_t)xdr_gf_event_notify_rsp);
+        if (dict)
+                dict_unref (dict);
+        if (args.dict.dict_val)
+                free (args.dict.dict_val);//malloced by xdr
+
+        return 0;
+}
 rpcsvc_actor_t gluster_handshake_actors[] = {
         [GF_HNDSK_NULL]      = {"NULL",      GF_HNDSK_NULL,      NULL, NULL, NULL, 0},
         [GF_HNDSK_GETSPEC]   = {"GETSPEC",   GF_HNDSK_GETSPEC,   server_getspec, NULL, NULL, 0},
+        [GF_HNDSK_EVENT_NOTIFY] = {"EVENTNOTIFY", GF_HNDSK_EVENT_NOTIFY, server_event_notify,
+                                    NULL, NULL, 0},
 };
 
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index 80976f2ca..3b6f679ed 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -3253,7 +3253,6 @@ out:
         return ret;
 }
 
-
 int
 glusterd_defrag_volume_node_rsp (dict_t *req_dict, dict_t *rsp_dict,
                                  dict_t *op_ctx)
@@ -3261,16 +3260,11 @@ glusterd_defrag_volume_node_rsp (dict_t *req_dict, dict_t *rsp_dict,
         int                             ret = 0;
         char                            *volname = NULL;
         glusterd_volinfo_t              *volinfo = NULL;
-        uint64_t                        files = 0;
-        uint64_t                        size = 0;
-        uint64_t                        lookup = 0;
-        gf_defrag_status_t              status = GF_DEFRAG_STATUS_NOT_STARTED;
         char                            key[256] = {0,};
         int32_t                         i = 0;
         char                            buf[1024] = {0,};
         char                            *node_str = NULL;
         glusterd_conf_t                 *priv = NULL;
-        uint64_t                        failures = 0;
 
         priv = THIS->private;
         GF_ASSERT (req_dict);
@@ -3286,48 +3280,16 @@ glusterd_defrag_volume_node_rsp (dict_t *req_dict, dict_t *rsp_dict,
         if (ret)
                 goto out;
 
-        if (!rsp_dict)
-                goto populate;
-
-        ret = dict_get_uint64 (rsp_dict, "files", &files);
-        if (ret)
-                gf_log (THIS->name, GF_LOG_TRACE,
-                        "failed to get file count");
-
-        ret = dict_get_uint64 (rsp_dict, "size", &size);
-        if (ret)
-                gf_log (THIS->name, GF_LOG_TRACE,
-                        "failed to get size of xfer");
-
-        ret = dict_get_uint64 (rsp_dict, "lookups", &lookup);
-        if (ret)
-                gf_log (THIS->name, GF_LOG_TRACE,
-                        "failed to get lookedup file count");
-        ret = dict_get_int32 (rsp_dict, "status", (int32_t *)&status);
-        if (ret)
-                gf_log (THIS->name, GF_LOG_TRACE,
-                        "failed to get status");
-
-        ret = dict_get_uint64 (rsp_dict, "failures", &failures);
-        if (ret)
-                gf_log (THIS->name, GF_LOG_TRACE,
-                        "failed to get failure count");
-
-        if (files)
-                volinfo->rebalance_files = files;
-        if (size)
-                volinfo->rebalance_data = size;
-        if (lookup)
-                volinfo->lookedup_files = lookup;
-        if (failures)
-                volinfo->rebalance_failures = failures;
+        if (rsp_dict) {
+                ret = glusterd_defrag_volume_status_update (volinfo,
+                                                            rsp_dict);
+        }
 
         if (!op_ctx) {
                 dict_copy (rsp_dict, op_ctx);
                 goto out;
         }
 
-populate:
         ret = dict_get_int32 (op_ctx, "count", &i);
         i++;
 
@@ -3364,12 +3326,10 @@ populate:
         if (ret)
                 gf_log (THIS->name, GF_LOG_ERROR,
                         "failed to set lookedup file count");
-        if (!status)
-                status = volinfo->defrag_status;
 
         memset (key, 0 , 256);
         snprintf (key, 256, "status-%d", i);
-        ret = dict_set_int32 (op_ctx, key, status);
+        ret = dict_set_int32 (op_ctx, key, volinfo->defrag_status);
         if (ret)
                 gf_log (THIS->name, GF_LOG_ERROR,
                         "failed to set status");
diff --git a/xlators/mgmt/glusterd/src/glusterd-rebalance.c b/xlators/mgmt/glusterd/src/glusterd-rebalance.c
index fddf6699f..9ed6af667 100644
--- a/xlators/mgmt/glusterd/src/glusterd-rebalance.c
+++ b/xlators/mgmt/glusterd/src/glusterd-rebalance.c
@@ -677,3 +677,30 @@ out:
 
         return ret;
 }
+
+int32_t
+glusterd_defrag_event_notify_handle (dict_t *dict)
+{
+        glusterd_volinfo_t      *volinfo = NULL;
+        char                    *volname = NULL;
+        int32_t                  ret     = -1;
+
+        ret = dict_get_str (dict, "volname", &volname);
+        if (ret) {
+                gf_log ("", GF_LOG_ERROR, "Failed to get volname");
+                return ret;
+        }
+
+        ret = glusterd_volinfo_find (volname, &volinfo);
+        if (ret) {
+                gf_log ("", GF_LOG_ERROR, "Failed to get volinfo for %s"
+                        , volname);
+                return ret;
+        }
+
+        ret = glusterd_defrag_volume_status_update (volinfo, dict);
+
+        if (ret)
+                gf_log ("", GF_LOG_ERROR, "Failed to update status");
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.c b/xlators/mgmt/glusterd/src/glusterd-utils.c
index 73770bb76..6e3a16710 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.c
@@ -5418,3 +5418,56 @@ glusterd_validate_volume_id (dict_t *op_dict, glusterd_volinfo_t *volinfo)
 out:
         return ret;
 }
+
+int
+glusterd_defrag_volume_status_update (glusterd_volinfo_t *volinfo,
+                                      dict_t *rsp_dict)
+{
+        int                             ret = 0;
+        uint64_t                        files = 0;
+        uint64_t                        size = 0;
+        uint64_t                        lookup = 0;
+        gf_defrag_status_t              status = GF_DEFRAG_STATUS_NOT_STARTED;
+        uint64_t                        failures = 0;
+        xlator_t                       *this = NULL;
+
+        this = THIS;
+
+        ret = dict_get_uint64 (rsp_dict, "files", &files);
+        if (ret)
+                gf_log (this->name, GF_LOG_TRACE,
+                        "failed to get file count");
+
+        ret = dict_get_uint64 (rsp_dict, "size", &size);
+        if (ret)
+                gf_log (this->name, GF_LOG_TRACE,
+                        "failed to get size of xfer");
+
+        ret = dict_get_uint64 (rsp_dict, "lookups", &lookup);
+        if (ret)
+                gf_log (this->name, GF_LOG_TRACE,
+                        "failed to get lookedup file count");
+
+        ret = dict_get_int32 (rsp_dict, "status", (int32_t *)&status);
+        if (ret)
+                gf_log (this->name, GF_LOG_TRACE,
+                        "failed to get status");
+
+        ret = dict_get_uint64 (rsp_dict, "failures", &failures);
+        if (ret)
+                gf_log (this->name, GF_LOG_TRACE,
+                        "failed to get failure count");
+
+        if (files)
+                volinfo->rebalance_files = files;
+        if (size)
+                volinfo->rebalance_data = size;
+        if (lookup)
+                volinfo->lookedup_files = lookup;
+        if (status)
+                volinfo->defrag_status = status;
+        if (failures)
+                volinfo->rebalance_failures = failures;
+
+        return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-utils.h b/xlators/mgmt/glusterd/src/glusterd-utils.h
index 602dc780f..0213f0a1d 100644
--- a/xlators/mgmt/glusterd/src/glusterd-utils.h
+++ b/xlators/mgmt/glusterd/src/glusterd-utils.h
@@ -434,4 +434,8 @@ glusterd_is_local_brick (xlator_t *this, glusterd_volinfo_t *volinfo,
                          glusterd_brickinfo_t *brickinfo);
 int
 glusterd_validate_volume_id (dict_t *op_dict, glusterd_volinfo_t *volinfo);
+
+int
+glusterd_defrag_volume_status_update (glusterd_volinfo_t *volinfo,
+                                      dict_t *rsp_dict);
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index 7484fc695..a4c414be7 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -642,5 +642,6 @@ int glusterd_op_gsync_args_get (dict_t *dict, char **op_errstr,
 /* Synctask part */
 int32_t glusterd_op_begin_synctask (rpcsvc_request_t *req, glusterd_op_t op,
                                     void *dict);
-
+int32_t
+glusterd_defrag_event_notify_handle (dict_t *dict);
 #endif