glusterd: op-version handshake implementation

Brings in a new rpc program MGMT_HANDSHAKE, which implements the op-version
handshake. This is required for bringing in the op-version feature as described
in http://www.gluster.org/community/documentation/index.php/Features/Opversion

Change-Id: I4333fd2714dbbd3a2a3fca5862cbb3c56615529e
BUG: 814534
Signed-off-by: Kaushal M <kaushal@redhat.com>
Reviewed-on: http://review.gluster.org/3688
Reviewed-by: Vijay Bellur <vbellur@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>

10 files changed, 882 insertions, 155 deletions

diff --git a/rpc/rpc-lib/src/protocol-common.h b/rpc/rpc-lib/src/protocol-common.h
index 603bda5826c..899844cd44c 100644
--- a/rpc/rpc-lib/src/protocol-common.h
+++ b/rpc/rpc-lib/src/protocol-common.h
@@ -185,6 +185,13 @@ enum glusterd_brick_procnum {
         GLUSTERD_BRICK_MAXVALUE,
 };
 
+enum glusterd_mgmt_hndsk_procnum {
+        GD_MGMT_HNDSK_NULL,
+        GD_MGMT_HNDSK_VERSIONS,
+        GD_MGMT_HNDSK_VERSIONS_ACK,
+        GD_MGMT_HNDSK_MAXVALUE,
+};
+
 typedef enum {
         GF_AFR_OP_INVALID,
         GF_AFR_OP_HEAL_INDEX,
@@ -196,7 +203,7 @@ typedef enum {
 } gf_xl_afr_op_t ;
 
 #define GLUSTER_HNDSK_PROGRAM    14398633 /* Completely random */
-#define GLUSTER_HNDSK_VERSION    2   /* 0.0.1 */
+#define GLUSTER_HNDSK_VERSION    2   /* 0.0.2 */
 
 #define GLUSTER_PMAP_PROGRAM     34123456
 #define GLUSTER_PMAP_VERSION     1
@@ -221,4 +228,8 @@ typedef enum {
 #define GD_BRICK_PROGRAM         4867634 /*Completely random*/
 #define GD_BRICK_VERSION         2
 
+/* OP-VERSION handshake */
+#define GD_MGMT_HNDSK_PROGRAM    1239873 /* Completely random */
+#define GD_MGMT_HNDSK_VERSION    1
+
 #endif /* !_PROTOCOL_COMMON_H */
diff --git a/rpc/xdr/src/glusterfs3-xdr.c b/rpc/xdr/src/glusterfs3-xdr.c
index 9d55f89c2fe..a502b2ea16c 100644
--- a/rpc/xdr/src/glusterfs3-xdr.c
+++ b/rpc/xdr/src/glusterfs3-xdr.c
@@ -1213,32 +1213,6 @@ xdr_gfs3_readdirp_req (XDR *xdrs, gfs3_readdirp_req *objp)
 }
 
 bool_t
-xdr_gf_setvolume_req (XDR *xdrs, gf_setvolume_req *objp)
-{
-	register int32_t *buf;
-        buf = NULL;
-
-	 if (!xdr_bytes (xdrs, (char **)&objp->dict.dict_val, (u_int *) &objp->dict.dict_len, ~0))
-		 return FALSE;
-	return TRUE;
-}
-
-bool_t
-xdr_gf_setvolume_rsp (XDR *xdrs, gf_setvolume_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;
-}
-
-bool_t
 xdr_gfs3_access_req (XDR *xdrs, gfs3_access_req *objp)
 {
 	register int32_t *buf;
@@ -1612,6 +1586,32 @@ xdr_gfs3_rchecksum_rsp (XDR *xdrs, gfs3_rchecksum_rsp *objp)
 }
 
 bool_t
+xdr_gf_setvolume_req (XDR *xdrs, gf_setvolume_req *objp)
+{
+	register int32_t *buf;
+        buf = NULL;
+
+	 if (!xdr_bytes (xdrs, (char **)&objp->dict.dict_val, (u_int *) &objp->dict.dict_len, ~0))
+		 return FALSE;
+	return TRUE;
+}
+
+bool_t
+xdr_gf_setvolume_rsp (XDR *xdrs, gf_setvolume_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;
+}
+
+bool_t
 xdr_gf_getspec_req (XDR *xdrs, gf_getspec_req *objp)
 {
 	register int32_t *buf;
@@ -1644,6 +1644,32 @@ xdr_gf_getspec_rsp (XDR *xdrs, gf_getspec_rsp *objp)
 }
 
 bool_t
+xdr_gf_mgmt_hndsk_req (XDR *xdrs, gf_mgmt_hndsk_req *objp)
+{
+	register int32_t *buf;
+        buf = NULL;
+
+	 if (!xdr_bytes (xdrs, (char **)&objp->hndsk.hndsk_val, (u_int *) &objp->hndsk.hndsk_len, ~0))
+		 return FALSE;
+	return TRUE;
+}
+
+bool_t
+xdr_gf_mgmt_hndsk_rsp (XDR *xdrs, gf_mgmt_hndsk_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->hndsk.hndsk_val, (u_int *) &objp->hndsk.hndsk_len, ~0))
+		 return FALSE;
+	return TRUE;
+}
+
+bool_t
 xdr_gf_log_req (XDR *xdrs, gf_log_req *objp)
 {
 	register int32_t *buf;
diff --git a/rpc/xdr/src/glusterfs3-xdr.h b/rpc/xdr/src/glusterfs3-xdr.h
index a68d1a6784e..0268d1a144a 100644
--- a/rpc/xdr/src/glusterfs3-xdr.h
+++ b/rpc/xdr/src/glusterfs3-xdr.h
@@ -731,24 +731,6 @@ struct gfs3_readdirp_req {
 };
 typedef struct gfs3_readdirp_req gfs3_readdirp_req;
 
-struct gf_setvolume_req {
-	struct {
-		u_int dict_len;
-		char *dict_val;
-	} dict;
-};
-typedef struct gf_setvolume_req gf_setvolume_req;
-
-struct gf_setvolume_rsp {
-	int op_ret;
-	int op_errno;
-	struct {
-		u_int dict_len;
-		char *dict_val;
-	} dict;
-};
-typedef struct gf_setvolume_rsp gf_setvolume_rsp;
-
 struct gfs3_access_req {
 	char gfid[16];
 	u_int mask;
@@ -931,6 +913,24 @@ struct gfs3_rchecksum_rsp {
 };
 typedef struct gfs3_rchecksum_rsp gfs3_rchecksum_rsp;
 
+struct gf_setvolume_req {
+	struct {
+		u_int dict_len;
+		char *dict_val;
+	} dict;
+};
+typedef struct gf_setvolume_req gf_setvolume_req;
+
+struct gf_setvolume_rsp {
+	int op_ret;
+	int op_errno;
+	struct {
+		u_int dict_len;
+		char *dict_val;
+	} dict;
+};
+typedef struct gf_setvolume_rsp gf_setvolume_rsp;
+
 struct gf_getspec_req {
 	u_int flags;
 	char *key;
@@ -952,6 +952,24 @@ struct gf_getspec_rsp {
 };
 typedef struct gf_getspec_rsp gf_getspec_rsp;
 
+struct gf_mgmt_hndsk_req {
+	struct {
+		u_int hndsk_len;
+		char *hndsk_val;
+	} hndsk;
+};
+typedef struct gf_mgmt_hndsk_req gf_mgmt_hndsk_req;
+
+struct gf_mgmt_hndsk_rsp {
+	int op_ret;
+	int op_errno;
+	struct {
+		u_int hndsk_len;
+		char *hndsk_val;
+	} hndsk;
+};
+typedef struct gf_mgmt_hndsk_rsp gf_mgmt_hndsk_rsp;
+
 struct gf_log_req {
 	struct {
 		u_int msg_len;
@@ -1151,8 +1169,6 @@ extern  bool_t xdr_gfs3_opendir_rsp (XDR *, gfs3_opendir_rsp*);
 extern  bool_t xdr_gfs3_fsyncdir_req (XDR *, gfs3_fsyncdir_req*);
 extern  bool_t xdr_gfs3_readdir_req (XDR *, gfs3_readdir_req*);
 extern  bool_t xdr_gfs3_readdirp_req (XDR *, gfs3_readdirp_req*);
-extern  bool_t xdr_gf_setvolume_req (XDR *, gf_setvolume_req*);
-extern  bool_t xdr_gf_setvolume_rsp (XDR *, gf_setvolume_rsp*);
 extern  bool_t xdr_gfs3_access_req (XDR *, gfs3_access_req*);
 extern  bool_t xdr_gfs3_create_req (XDR *, gfs3_create_req*);
 extern  bool_t xdr_gfs3_create_rsp (XDR *, gfs3_create_rsp*);
@@ -1168,8 +1184,12 @@ extern  bool_t xdr_gfs3_fsetattr_req (XDR *, gfs3_fsetattr_req*);
 extern  bool_t xdr_gfs3_fsetattr_rsp (XDR *, gfs3_fsetattr_rsp*);
 extern  bool_t xdr_gfs3_rchecksum_req (XDR *, gfs3_rchecksum_req*);
 extern  bool_t xdr_gfs3_rchecksum_rsp (XDR *, gfs3_rchecksum_rsp*);
+extern  bool_t xdr_gf_setvolume_req (XDR *, gf_setvolume_req*);
+extern  bool_t xdr_gf_setvolume_rsp (XDR *, gf_setvolume_rsp*);
 extern  bool_t xdr_gf_getspec_req (XDR *, gf_getspec_req*);
 extern  bool_t xdr_gf_getspec_rsp (XDR *, gf_getspec_rsp*);
+extern  bool_t xdr_gf_mgmt_hndsk_req (XDR *, gf_mgmt_hndsk_req*);
+extern  bool_t xdr_gf_mgmt_hndsk_rsp (XDR *, gf_mgmt_hndsk_rsp*);
 extern  bool_t xdr_gf_log_req (XDR *, gf_log_req*);
 extern  bool_t xdr_gf_notify_req (XDR *, gf_notify_req*);
 extern  bool_t xdr_gf_notify_rsp (XDR *, gf_notify_rsp*);
@@ -1243,8 +1263,6 @@ extern bool_t xdr_gfs3_opendir_rsp ();
 extern bool_t xdr_gfs3_fsyncdir_req ();
 extern bool_t xdr_gfs3_readdir_req ();
 extern bool_t xdr_gfs3_readdirp_req ();
-extern bool_t xdr_gf_setvolume_req ();
-extern bool_t xdr_gf_setvolume_rsp ();
 extern bool_t xdr_gfs3_access_req ();
 extern bool_t xdr_gfs3_create_req ();
 extern bool_t xdr_gfs3_create_rsp ();
@@ -1260,8 +1278,12 @@ extern bool_t xdr_gfs3_fsetattr_req ();
 extern bool_t xdr_gfs3_fsetattr_rsp ();
 extern bool_t xdr_gfs3_rchecksum_req ();
 extern bool_t xdr_gfs3_rchecksum_rsp ();
+extern bool_t xdr_gf_setvolume_req ();
+extern bool_t xdr_gf_setvolume_rsp ();
 extern bool_t xdr_gf_getspec_req ();
 extern bool_t xdr_gf_getspec_rsp ();
+extern bool_t xdr_gf_mgmt_hndsk_req ();
+extern bool_t xdr_gf_mgmt_hndsk_rsp ();
 extern bool_t xdr_gf_log_req ();
 extern bool_t xdr_gf_notify_req ();
 extern bool_t xdr_gf_notify_rsp ();
diff --git a/rpc/xdr/src/glusterfs3-xdr.x b/rpc/xdr/src/glusterfs3-xdr.x
index ad261423d2e..063f302d933 100644
--- a/rpc/xdr/src/glusterfs3-xdr.x
+++ b/rpc/xdr/src/glusterfs3-xdr.x
@@ -460,15 +460,6 @@ struct   gfs3_finodelk_req {
 }  ;
 
 
- struct gf_setvolume_req {
-        opaque dict<>;
-}  ;
- struct  gf_setvolume_rsp {
-        int    op_ret;
-        int    op_errno;
-        opaque dict<>;
-} ;
-
 struct gfs3_access_req  {
         opaque gfid[16];
 	unsigned int mask;
@@ -590,6 +581,16 @@ struct gfs3_fstat_req {
 }  ;
 
 
+ struct gf_setvolume_req {
+        opaque dict<>;
+}  ;
+ struct  gf_setvolume_rsp {
+        int    op_ret;
+        int    op_errno;
+        opaque dict<>;
+} ;
+
+
  struct gf_getspec_req {
 	unsigned int flags;
 	string     key<>;
@@ -602,10 +603,19 @@ struct gfs3_fstat_req {
         opaque   xdata<>; /* Extra data */
 } ;
 
+ struct gf_mgmt_hndsk_req {
+        opaque   hndsk<>;
+}  ;
+
+ struct  gf_mgmt_hndsk_rsp {
+        int    op_ret;
+        int    op_errno;
+        opaque   hndsk<>;
+} ;
 
  struct   gf_log_req {
-	opaque     msg<>;
-};
+        opaque    msg<>;
+} ;
 
  struct gf_notify_req {
 	unsigned int  flags;
diff --git a/xlators/mgmt/glusterd/src/glusterd-handler.c b/xlators/mgmt/glusterd/src/glusterd-handler.c
index a3106342d6c..7cdad10e219 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handler.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handler.c
@@ -2881,7 +2881,7 @@ glusterd_peer_rpc_notify (struct rpc_clnt *rpc, void *mydata,
                 gf_log (this->name, GF_LOG_DEBUG, "got RPC_CLNT_CONNECT");
                 peerinfo->connected = 1;
 
-                ret = glusterd_peer_handshake (this, rpc, peerctx);
+                ret = glusterd_peer_dump_version (this, rpc, peerctx);
                 if (ret)
                         gf_log ("", GF_LOG_ERROR, "glusterd handshake failed");
                 break;
diff --git a/xlators/mgmt/glusterd/src/glusterd-handshake.c b/xlators/mgmt/glusterd/src/glusterd-handshake.c
index ab38de17c9c..428051d4a51 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handshake.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handshake.c
@@ -21,6 +21,7 @@
 #include "glusterd.h"
 #include "glusterd-utils.h"
 #include "glusterd-op-sm.h"
+#include "glusterd-store.h"
 
 #include "glusterfs3.h"
 #include "protocol-common.h"
@@ -267,34 +268,247 @@ fail:
 
         return 0;
 }
+
+int
+gd_validate_cluster_op_version (xlator_t *this, int cluster_op_version,
+                                char *peerid)
+{
+        int              ret  = -1;
+        glusterd_conf_t *conf = NULL;
+
+        conf = this->private;
+
+        if (cluster_op_version > GD_OP_VERSION_MAX) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "operating version %d is more than the maximum "
+                        "supported (%d) on the machine (as per peer request "
+                        "from %s)", cluster_op_version, GD_OP_VERSION_MAX,
+                        peerid);
+                goto out;
+        }
+
+        if (cluster_op_version < conf->op_version) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "operating version %d is less than the currently "
+                        "running version (%d) on the machine (as per peer "
+                        "request from %s)", cluster_op_version,
+                        conf->op_version, peerid);
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+int
+glusterd_mgmt_hndsk_versions (rpcsvc_request_t *req)
+{
+        dict_t            *dict            = NULL;
+        xlator_t          *this            = NULL;
+        glusterd_conf_t   *conf            = NULL;
+        int                ret             = -1;
+        int                op_errno        = EINVAL;
+        gf_mgmt_hndsk_req  args            = {{0,},};
+        gf_mgmt_hndsk_rsp  rsp             = {0,};
+
+        this = THIS;
+        conf = this->private;
+
+        if (!xdr_to_generic (req->msg[0], &args,
+                             (xdrproc_t)xdr_gf_mgmt_hndsk_req)) {
+                //failed to decode msg;
+                req->rpc_err = GARBAGE_ARGS;
+                goto out;
+        }
+
+        dict = dict_new ();
+        if (!dict)
+                goto out;
+
+        ret = dict_set_int32 (dict, GD_OP_VERSION_KEY, conf->op_version);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "failed to set operating version");
+                rsp.op_ret = ret;
+                goto out;
+        }
+
+        ret = dict_set_int32 (dict, GD_MIN_OP_VERSION_KEY, GD_OP_VERSION_MIN);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "failed to set %s", GD_MIN_OP_VERSION_KEY);
+                rsp.op_ret = ret;
+                goto out;
+        }
+
+        ret = dict_set_int32 (dict, GD_MAX_OP_VERSION_KEY, GD_OP_VERSION_MAX);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "failed to set %s", GD_MAX_OP_VERSION_KEY);
+                rsp.op_ret = ret;
+                goto out;
+        }
+
+        ret = 0;
+
+        GF_PROTOCOL_DICT_SERIALIZE (this, dict, (&rsp.hndsk.hndsk_val),
+                                    rsp.hndsk.hndsk_len, op_errno, out);
+out:
+
+        rsp.op_ret = ret;
+        rsp.op_errno = op_errno;
+
+        glusterd_submit_reply (req, &rsp, NULL, 0, NULL,
+                               (xdrproc_t)xdr_gf_mgmt_hndsk_rsp);
+
+        ret = 0;
+
+        if (dict)
+                dict_unref (dict);
+
+        if (args.hndsk.hndsk_val)
+                free (args.hndsk.hndsk_val);
+
+        if (rsp.hndsk.hndsk_val)
+                GF_FREE (rsp.hndsk.hndsk_val);
+
+        return ret;
+}
+
+int
+glusterd_mgmt_hndsk_versions_ack (rpcsvc_request_t *req)
+{
+        dict_t            *clnt_dict       = NULL;
+        xlator_t          *this            = NULL;
+        glusterd_conf_t   *conf            = NULL;
+        int                ret             = -1;
+        int                op_errno        = EINVAL;
+        int32_t            peer_op_version = 0;
+        gf_mgmt_hndsk_req  args            = {{0,},};
+        gf_mgmt_hndsk_rsp  rsp             = {0,};
+
+        this = THIS;
+        conf = this->private;
+
+        if (!xdr_to_generic (req->msg[0], &args,
+                             (xdrproc_t)xdr_gf_mgmt_hndsk_req)) {
+                //failed to decode msg;
+                req->rpc_err = GARBAGE_ARGS;
+                goto out;
+        }
+
+        GF_PROTOCOL_DICT_UNSERIALIZE (this, clnt_dict, args.hndsk.hndsk_val,
+                                      (args.hndsk.hndsk_len), ret, op_errno,
+                                      out);
+
+        ret = dict_get_int32 (clnt_dict, GD_OP_VERSION_KEY,
+                              &peer_op_version);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "failed to get the op-version key peer=%s",
+                        req->trans->peerinfo.identifier);
+                goto out;
+        }
+
+        ret = gd_validate_cluster_op_version (this, peer_op_version,
+                                              req->trans->peerinfo.identifier);
+        if (ret)
+                goto out;
+
+
+        /* As this is ACK from the Cluster for the versions supported,
+           can set the op-version of 'this' glusterd to the one
+           received. */
+        gf_log (this->name, GF_LOG_INFO, "using the op-version %d",
+                peer_op_version);
+        conf->op_version = peer_op_version;
+        ret = glusterd_store_global_info (this);
+        if (ret)
+                gf_log (this->name, GF_LOG_ERROR, "Failed to store op-version");
+
+out:
+        rsp.op_ret = ret;
+        rsp.op_errno = op_errno;
+
+        glusterd_submit_reply (req, &rsp, NULL, 0, NULL,
+                               (xdrproc_t)xdr_gf_mgmt_hndsk_rsp);
+
+        if (clnt_dict)
+                dict_unref (clnt_dict);
+
+        if (args.hndsk.hndsk_val)
+                free (args.hndsk.hndsk_val);
+
+        return ret;
+}
+
+
 rpcsvc_actor_t gluster_handshake_actors[] = {
-        [GF_HNDSK_NULL]      = {"NULL",      GF_HNDSK_NULL,      NULL, NULL, 0},
-        [GF_HNDSK_GETSPEC]   = {"GETSPEC",   GF_HNDSK_GETSPEC,   server_getspec, NULL, 0},
-        [GF_HNDSK_EVENT_NOTIFY] = {"EVENTNOTIFY", GF_HNDSK_EVENT_NOTIFY, server_event_notify,
-                                    NULL, 0},
+        [GF_HNDSK_NULL]         = {"NULL", GF_HNDSK_NULL, NULL, NULL, 0},
+        [GF_HNDSK_GETSPEC]      = {"GETSPEC", GF_HNDSK_GETSPEC,
+                                   server_getspec, NULL, 0},
+        [GF_HNDSK_EVENT_NOTIFY] = {"EVENTNOTIFY", GF_HNDSK_EVENT_NOTIFY,
+                                   server_event_notify,  NULL, 0},
 };
 
 
 struct rpcsvc_program gluster_handshake_prog = {
-        .progname  = "GlusterFS Handshake",
+        .progname  = "Gluster Handshake",
         .prognum   = GLUSTER_HNDSK_PROGRAM,
         .progver   = GLUSTER_HNDSK_VERSION,
         .actors    = gluster_handshake_actors,
         .numactors = GF_HNDSK_MAXVALUE,
 };
 
+
 char *glusterd_dump_proc[GF_DUMP_MAXVALUE] = {
         [GF_DUMP_NULL] = "NULL",
         [GF_DUMP_DUMP] = "DUMP",
 };
 
 rpc_clnt_prog_t glusterd_dump_prog = {
-        .progname  = "GLUSTERD-DUMP",
-        .prognum   = GLUSTER_DUMP_PROGRAM,
-        .progver   = GLUSTER_DUMP_VERSION,
-        .procnames = glusterd_dump_proc,
+        .progname       = "GLUSTERD-DUMP",
+        .prognum        = GLUSTER_DUMP_PROGRAM,
+        .progver        = GLUSTER_DUMP_VERSION,
+        .procnames      = glusterd_dump_proc,
 };
 
+
+rpcsvc_actor_t glusterd_mgmt_hndsk_actors[] = {
+        [GD_MGMT_HNDSK_NULL]            = {"NULL", GD_MGMT_HNDSK_NULL, NULL,
+                                           NULL, 0},
+        [GD_MGMT_HNDSK_VERSIONS]        = {"MGMT-VERS", GD_MGMT_HNDSK_VERSIONS,
+                                           glusterd_mgmt_hndsk_versions, NULL,
+                                           0},
+        [GD_MGMT_HNDSK_VERSIONS_ACK]    = {"MGMT-VERS-ACK",
+                                           GD_MGMT_HNDSK_VERSIONS_ACK,
+                                           glusterd_mgmt_hndsk_versions_ack,
+                                           NULL, 0},
+};
+
+struct rpcsvc_program glusterd_mgmt_hndsk_prog = {
+        .progname       = "Gluster MGMT Handshake",
+        .prognum        = GD_MGMT_HNDSK_PROGRAM,
+        .progver        = GD_MGMT_HNDSK_VERSION,
+        .actors         = glusterd_mgmt_hndsk_actors,
+        .numactors      = GD_MGMT_HNDSK_MAXVALUE,
+};
+
+char *glusterd_mgmt_hndsk_proc[GD_MGMT_HNDSK_MAXVALUE] = {
+        [GD_MGMT_HNDSK_NULL]          = "NULL",
+        [GD_MGMT_HNDSK_VERSIONS] = "MGMT-VERS",
+        [GD_MGMT_HNDSK_VERSIONS_ACK] = "MGMT-VERS-ACK",
+};
+
+rpc_clnt_prog_t gd_clnt_mgmt_hndsk_prog = {
+        .progname  = "Gluster MGMT Handshake",
+        .prognum   = GD_MGMT_HNDSK_PROGRAM,
+        .progver   = GD_MGMT_HNDSK_VERSION,
+        .procnames = glusterd_mgmt_hndsk_proc,
+};
+
+
 static int
 glusterd_event_connected_inject (glusterd_peerctx_t *peerctx)
 {
@@ -343,12 +557,279 @@ out:
         return ret;
 }
 
+
+int
+gd_validate_peer_op_version (xlator_t *this, glusterd_peerinfo_t *peerinfo,
+                             dict_t *dict, char **errstr)
+{
+        int              ret  = -1;
+        glusterd_conf_t *conf = NULL;
+        int32_t          peer_op_version = 0;
+        int32_t          peer_min_op_version = 0;
+        int32_t          peer_max_op_version = 0;
+
+        if (!dict && !this && !peerinfo)
+                goto out;
+
+        conf = this->private;
+
+        ret = dict_get_int32 (dict, GD_OP_VERSION_KEY, &peer_op_version);
+        if (ret)
+                goto out;
+
+        ret = dict_get_int32 (dict, GD_MAX_OP_VERSION_KEY,
+                              &peer_max_op_version);
+        if (ret)
+                goto out;
+
+        ret = dict_get_int32 (dict, GD_MIN_OP_VERSION_KEY,
+                              &peer_min_op_version);
+        if (ret)
+                goto out;
+
+        ret = -1;
+        /* Check if peer can support our op_version */
+        if ((peer_max_op_version < conf->op_version) ||
+            (peer_min_op_version > conf->op_version)) {
+                ret = gf_asprintf (errstr, "Peer %s does not support required "
+                                   "op-version", peerinfo->hostname);
+                ret = -1;
+                goto out;
+        }
+
+        /* If peer is already operating at a higher op_version reject it.
+         * Cluster cannot be moved to higher op_version to accomodate a peer.
+         */
+        if (peer_op_version > conf->op_version) {
+                ret = gf_asprintf (errstr, "Peer %s is already at a higher "
+                                   "op-version", peerinfo->hostname);
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        gf_log (this->name , GF_LOG_DEBUG, "Peer %s %s", peerinfo->hostname,
+                ((ret < 0) ? "rejected" : "accepted"));
+        return ret;
+}
+
+int
+glusterd_mgmt_hndsk_version_ack_cbk (struct rpc_req *req, struct iovec *iov,
+                                     int count, void *myframe)
+{
+        int                  ret      = -1;
+        int                  op_errno = EINVAL;
+        gf_mgmt_hndsk_rsp    rsp      = {0,};
+        xlator_t            *this     = NULL;
+        call_frame_t        *frame    = NULL;
+        glusterd_peerinfo_t *peerinfo = NULL;
+        glusterd_peerctx_t  *peerctx  = NULL;
+        char                msg[1024] = {0,};
+
+        this = THIS;
+        frame = myframe;
+        peerctx = frame->local;
+        peerinfo = peerctx->peerinfo;
+
+        if (-1 == req->rpc_status) {
+                snprintf (msg, sizeof (msg),
+                          "Error through RPC layer, retry again later");
+                gf_log ("", GF_LOG_ERROR, "%s", msg);
+                peerctx->errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_mgmt_hndsk_rsp);
+        if (ret < 0) {
+                snprintf (msg, sizeof (msg), "Failed to decode XDR");
+                gf_log ("", GF_LOG_ERROR, "%s", msg);
+                peerctx->errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        op_errno = rsp.op_errno;
+        if (-1 == rsp.op_ret) {
+                ret = -1;
+                snprintf (msg, sizeof (msg),
+                          "Failed to get handshake ack from remote server");
+                gf_log (frame->this->name, GF_LOG_ERROR, "%s", msg);
+                peerctx->errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        /* TODO: this is hardcoded as of now, but I don't forsee any problems
+         * with this as long as we are properly handshaking operating versions
+         */
+        peerinfo->mgmt = &gd_mgmt_prog;
+        peerinfo->peer = &gd_peer_prog;
+
+        ret = default_notify (this, GF_EVENT_CHILD_UP, NULL);
+
+        if (GD_MODE_ON == peerctx->args.mode) {
+                ret = glusterd_event_connected_inject (peerctx);
+                peerctx->args.req = NULL;
+        } else if (GD_MODE_SWITCH_ON == peerctx->args.mode) {
+                peerctx->args.mode = GD_MODE_ON;
+        } else {
+                gf_log (this->name, GF_LOG_WARNING, "unknown mode %d",
+                        peerctx->args.mode);
+        }
+
+        glusterd_friend_sm ();
+
+        ret = 0;
+out:
+
+        frame->local = NULL;
+        STACK_DESTROY (frame->root);
+
+        if (ret != 0)
+                rpc_transport_disconnect (peerinfo->rpc->conn.trans);
+
+        if (rsp.hndsk.hndsk_val)
+                free (rsp.hndsk.hndsk_val);
+
+        return 0;
+}
+
+int
+glusterd_mgmt_hndsk_version_cbk (struct rpc_req *req, struct iovec *iov,
+                                 int count, void *myframe)
+{
+        int                  ret       = -1;
+        int                  op_errno  = EINVAL;
+        gf_mgmt_hndsk_rsp    rsp       = {0,};
+        gf_mgmt_hndsk_req    arg       = {{0,}};
+        xlator_t            *this      = NULL;
+        call_frame_t        *frame     = NULL;
+        glusterd_peerinfo_t *peerinfo  = NULL;
+        glusterd_peerctx_t  *peerctx   = NULL;
+        dict_t              *dict      = NULL;
+        dict_t              *rsp_dict  = NULL;
+        glusterd_conf_t     *conf      = NULL;
+        char                 msg[1024] = {0,};
+
+        this = THIS;
+        conf = this->private;
+        frame = myframe;
+        peerctx = frame->local;
+        peerinfo = peerctx->peerinfo;
+
+        if (-1 == req->rpc_status) {
+                ret = -1;
+                snprintf (msg, sizeof (msg),
+                          "Error through RPC layer, retry again later");
+                gf_log (this->name, GF_LOG_ERROR, "%s", msg);
+                peerctx->errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_mgmt_hndsk_rsp);
+        if (ret < 0) {
+                snprintf (msg, sizeof (msg), "Failed to decode management "
+                          "handshake response");
+                gf_log (this->name, GF_LOG_ERROR, "%s", msg);
+                peerctx->errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        GF_PROTOCOL_DICT_UNSERIALIZE (this, dict, rsp.hndsk.hndsk_val,
+                                      rsp.hndsk.hndsk_len, ret, op_errno,
+                                      out);
+
+        op_errno = rsp.op_errno;
+        if (-1 == rsp.op_ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to get the 'versions' from peer (%s)",
+                        req->conn->trans->peerinfo.identifier);
+                goto out;
+        }
+
+        /* Check if peer can be part of cluster */
+        ret = gd_validate_peer_op_version (this, peerinfo, dict,
+                                           &peerctx->errstr);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to validate the operating version of peer (%s)",
+                        peerinfo->hostname);
+                goto out;
+        }
+
+        rsp_dict = dict_new ();
+        if (!rsp_dict)
+                goto out;
+
+        ret = dict_set_int32 (rsp_dict, GD_OP_VERSION_KEY, conf->op_version);
+        if (ret) {
+                gf_log(this->name, GF_LOG_ERROR,
+                       "failed to set operating version in dict");
+                goto out;
+        }
+
+        GF_PROTOCOL_DICT_SERIALIZE (this, rsp_dict, (&arg.hndsk.hndsk_val),
+                                    arg.hndsk.hndsk_len, op_errno, out);
+
+        ret = glusterd_submit_request (peerctx->peerinfo->rpc, &arg, frame,
+                                       &gd_clnt_mgmt_hndsk_prog,
+                                       GD_MGMT_HNDSK_VERSIONS_ACK, NULL, this,
+                                       glusterd_mgmt_hndsk_version_ack_cbk,
+                                       (xdrproc_t)xdr_gf_mgmt_hndsk_req);
+
+out:
+        if (ret) {
+                frame->local = NULL;
+                STACK_DESTROY (frame->root);
+                rpc_transport_disconnect (peerinfo->rpc->conn.trans);
+        }
+
+        if (rsp.hndsk.hndsk_val)
+                free (rsp.hndsk.hndsk_val);
+
+        if (arg.hndsk.hndsk_val)
+                GF_FREE (arg.hndsk.hndsk_val);
+
+        if (dict)
+                dict_unref (dict);
+
+        if (rsp_dict)
+                dict_unref (rsp_dict);
+
+        return 0;
+}
+
+int
+glusterd_mgmt_handshake (xlator_t *this, glusterd_peerctx_t *peerctx)
+{
+        call_frame_t        *frame    = NULL;
+        gf_mgmt_hndsk_req    req      = {{0,},};
+        int                  ret      = -1;
+
+        frame = create_frame (this, this->ctx->pool);
+        if (!frame)
+                goto out;
+
+        frame->local = peerctx;
+
+        ret = glusterd_submit_request (peerctx->peerinfo->rpc, &req, frame,
+                                       &gd_clnt_mgmt_hndsk_prog,
+                                       GD_MGMT_HNDSK_VERSIONS, NULL, this,
+                                       glusterd_mgmt_hndsk_version_cbk,
+                                       (xdrproc_t)xdr_gf_mgmt_hndsk_req);
+        ret = 0;
+out:
+        if (ret && frame)
+                STACK_DESTROY (frame->root);
+
+        return ret;
+}
+
 int
 glusterd_set_clnt_mgmt_program (glusterd_peerinfo_t *peerinfo,
                                 gf_prog_detail *prog)
 {
-        gf_prog_detail *trav     = NULL;
-        int             ret      = -1;
+        gf_prog_detail  *trav   = NULL;
+        int             ret     = -1;
 
         if (!peerinfo || !prog)
                 goto out;
@@ -356,39 +837,65 @@ glusterd_set_clnt_mgmt_program (glusterd_peerinfo_t *peerinfo,
         trav = prog;
 
         while (trav) {
-                /* Select 'programs' */
+                ret = -1;
                 if ((gd_mgmt_prog.prognum == trav->prognum) &&
                     (gd_mgmt_prog.progver == trav->progver)) {
                         peerinfo->mgmt = &gd_mgmt_prog;
                         ret = 0;
                 }
+
                 if ((gd_peer_prog.prognum == trav->prognum) &&
                     (gd_peer_prog.progver == trav->progver)) {
                         peerinfo->peer = &gd_peer_prog;
                         ret = 0;
                 }
+
                 if (ret) {
                         gf_log ("", GF_LOG_DEBUG,
                                 "%s (%"PRId64":%"PRId64") not supported",
                                 trav->progname, trav->prognum,
                                 trav->progver);
                 }
+
                 trav = trav->next;
         }
 
         if (peerinfo->mgmt) {
-                gf_log ("", GF_LOG_INFO,
-                        "Using Program %s, Num (%d), Version (%d)",
-                        peerinfo->mgmt->progname, peerinfo->mgmt->prognum,
-                        peerinfo->mgmt->progver);
+                 gf_log ("", GF_LOG_INFO,
+                         "Using Program %s, Num (%d), Version (%d)",
+                         peerinfo->mgmt->progname, peerinfo->mgmt->prognum,
+                         peerinfo->mgmt->progver);
         }
+
         if (peerinfo->peer) {
-                gf_log ("", GF_LOG_INFO,
-                        "Using Program %s, Num (%d), Version (%d)",
-                        peerinfo->peer->progname, peerinfo->peer->prognum,
-                        peerinfo->peer->progver);
+                 gf_log ("", GF_LOG_INFO,
+                         "Using Program %s, Num (%d), Version (%d)",
+                         peerinfo->peer->progname, peerinfo->peer->prognum,
+                         peerinfo->peer->progver);
         }
+        ret = 0;
+out:
+        return ret;
+
+}
 
+static gf_boolean_t
+_mgmt_hndsk_prog_present (gf_prog_detail *prog) {
+        gf_boolean_t    ret = _gf_false;
+        gf_prog_detail  *trav = NULL;
+
+        GF_ASSERT (prog);
+
+        trav = prog;
+
+        while (trav) {
+                if ((trav->prognum == GD_MGMT_HNDSK_PROGRAM) &&
+                    (trav->progver == GD_MGMT_HNDSK_VERSION)) {
+                        ret = _gf_true;
+                        goto out;
+                }
+                trav = trav->next;
+        }
 out:
         return ret;
 }
@@ -405,9 +912,11 @@ glusterd_peer_dump_version_cbk (struct rpc_req *req, struct iovec *iov,
         call_frame_t        *frame    = NULL;
         glusterd_peerinfo_t *peerinfo = NULL;
         glusterd_peerctx_t  *peerctx  = NULL;
+        glusterd_conf_t     *conf     = NULL;
         char                msg[1024] = {0,};
 
         this = THIS;
+        conf = this->private;
         frame = myframe;
         peerctx = frame->local;
         peerinfo = peerctx->peerinfo;
@@ -435,6 +944,22 @@ glusterd_peer_dump_version_cbk (struct rpc_req *req, struct iovec *iov,
                 goto out;
         }
 
+        if (_mgmt_hndsk_prog_present (rsp.prog)) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "Proceeding to op-version handshake with peer %s",
+                        peerinfo->hostname);
+                ret = glusterd_mgmt_handshake (this, peerctx);
+                goto out;
+        } else if (conf->op_version > 1) {
+                ret = -1;
+                snprintf (msg, sizeof (msg),
+                          "Peer %s does not support required op-version",
+                          peerinfo->hostname);
+                peerctx->errstr = gf_strdup (msg);
+                gf_log (this->name, GF_LOG_ERROR, "%s", msg);
+                goto out;
+        }
+
         /* Make sure we assign the proper program to peer */
         ret = glusterd_set_clnt_mgmt_program (peerinfo, rsp.prog);
         if (ret) {
@@ -454,10 +979,11 @@ glusterd_peer_dump_version_cbk (struct rpc_req *req, struct iovec *iov,
                         peerctx->args.mode);
         }
 
-        glusterd_friend_sm ();
-        glusterd_op_sm ();
+        glusterd_friend_sm();
+        glusterd_op_sm();
 
         ret = 0;
+
 out:
 
         /* don't use GF_FREE, buffer was allocated by libc */
@@ -482,8 +1008,8 @@ out:
 
 
 int
-glusterd_peer_handshake (xlator_t *this, struct rpc_clnt *rpc,
-                         glusterd_peerctx_t *peerctx)
+glusterd_peer_dump_version (xlator_t *this, struct rpc_clnt *rpc,
+                            glusterd_peerctx_t *peerctx)
 {
         call_frame_t        *frame    = NULL;
         gf_dump_req          req      = {0,};
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.c b/xlators/mgmt/glusterd/src/glusterd-store.c
index 6f08cc7e224..6fa490e47d5 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.c
+++ b/xlators/mgmt/glusterd/src/glusterd-store.c
@@ -1505,36 +1505,37 @@ out:
         return ret;
 }
 
-int32_t
-glusterd_store_uuid ()
+int
+glusterd_store_global_info (xlator_t *this)
 {
-        glusterd_conf_t *priv = NULL;
-        char            path[PATH_MAX] = {0,};
-        int32_t         ret = -1;
-        glusterd_store_handle_t *handle = NULL;
-        xlator_t       *this    = NULL;
+        int                     ret                     = -1;
+        glusterd_conf_t         *conf                   = NULL;
+        char                    op_version_str[15]      = {0,};
+        char                    path[PATH_MAX]          = {0,};
+        glusterd_store_handle_t *handle                 = NULL;
+        char                    *uuid_str               = NULL;
 
-        this = THIS;
-        priv = this->private;
+        conf = this->private;
 
-        snprintf (path, PATH_MAX, "%s/%s", priv->workdir,
-                  GLUSTERD_INFO_FILE);
+        uuid_str = gf_strdup (uuid_utoa (MY_UUID));
+        if (!uuid_str)
+                goto out;
 
-        if (!priv->handle) {
+        if (!conf->handle) {
+                snprintf (path, PATH_MAX, "%s/%s", conf->workdir,
+                          GLUSTERD_INFO_FILE);
                 ret = glusterd_store_handle_new (path, &handle);
-
                 if (ret) {
                         gf_log (this->name, GF_LOG_ERROR,
-                                "Unable to get store handle!");
+                                "Unable to get store handle");
                         goto out;
                 }
 
-                priv->handle = handle;
-        } else {
-                handle = priv->handle;
-        }
+                conf->handle = handle;
+        } else
+                handle = conf->handle;
 
-        /* make glusterd's uuid available for users */
+        /* These options need to be available for all users */
         ret = chmod (handle->path, 0644);
         if (ret) {
                 gf_log (this->name, GF_LOG_ERROR, "chmod error for %s: %s",
@@ -1542,27 +1543,132 @@ glusterd_store_uuid ()
                 goto out;
         }
 
-        handle->fd = open (handle->path, O_RDWR | O_CREAT | O_TRUNC, 0644);
+        handle->fd = glusterd_store_mkstemp (handle);
         if (handle->fd <= 0) {
                 ret = -1;
                 goto out;
         }
-        ret = glusterd_store_save_value (handle->fd, GLUSTERD_STORE_UUID_KEY,
-                                         uuid_utoa (MY_UUID));
 
+        ret = glusterd_store_save_value (handle->fd, GLUSTERD_STORE_UUID_KEY,
+                                         uuid_str);
         if (ret) {
                 gf_log (this->name, GF_LOG_CRITICAL,
                         "Storing uuid failed ret = %d", ret);
                 goto out;
         }
 
+        snprintf (op_version_str, 15, "%d", conf->op_version);
+        ret = glusterd_store_save_value (handle->fd, GD_OP_VERSION_KEY,
+                                         op_version_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Storing op-version failed ret = %d", ret);
+                goto out;
+        }
 
+        ret = glusterd_store_rename_tmppath (handle);
 out:
+        if (ret && (handle->fd > 0))
+                glusterd_store_unlink_tmppath (handle);
+
         if (handle->fd > 0) {
                 close (handle->fd);
                 handle->fd = 0;
         }
-        gf_log (this->name, GF_LOG_DEBUG, "Returning %d", ret);
+
+        if (uuid_str)
+                GF_FREE (uuid_str);
+
+        if (ret)
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to store glusterd global-info");
+
+        return ret;
+}
+
+int
+glusterd_retrieve_op_version (xlator_t *this, int *op_version)
+{
+        char                    *op_version_str = NULL;
+        glusterd_conf_t         *priv           = NULL;
+        int                     ret             = -1;
+        int                     tmp_version     = 0;
+        char                    *tmp            = NULL;
+        char                    path[PATH_MAX]  = {0,};
+        glusterd_store_handle_t *handle         = NULL;
+
+        priv = this->private;
+
+        if (!priv->handle) {
+                snprintf (path, PATH_MAX, "%s/%s", priv->workdir,
+                          GLUSTERD_INFO_FILE);
+                ret = glusterd_store_handle_retrieve (path, &handle);
+
+                if (ret) {
+                        gf_log ("", GF_LOG_ERROR, "Unable to get store "
+                                "handle!");
+                        goto out;
+                }
+
+                priv->handle = handle;
+        }
+
+        ret = glusterd_store_retrieve_value (priv->handle,
+                                             GD_OP_VERSION_KEY,
+                                             &op_version_str);
+        if (ret) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "No previous op_version present");
+                goto out;
+        }
+
+        tmp_version = strtol (op_version_str, &tmp, 10);
+        if ((tmp_version <= 0) || (tmp && strlen (tmp) > 1)) {
+                gf_log (this->name, GF_LOG_WARNING, "invalid version number");
+                goto out;
+        }
+
+        *op_version = tmp_version;
+
+        ret = 0;
+out:
+        if (op_version_str)
+                GF_FREE (op_version_str);
+
+        return ret;
+}
+
+static int
+glusterd_restore_op_version (xlator_t *this)
+{
+        glusterd_conf_t *conf = NULL;
+        int ret = 0;
+        int op_version = 0;
+
+        conf = this->private;
+
+        ret = glusterd_retrieve_op_version (this, &op_version);
+        if (!ret) {
+                if ((op_version < GD_OP_VERSION_MIN) ||
+                    (op_version > GD_OP_VERSION_MAX)) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "wrong op-version (%d) retreived", op_version);
+                        ret = -1;
+                        goto out;
+                }
+                conf->op_version = op_version;
+                gf_log ("glusterd", GF_LOG_INFO,
+                        "retrieved op-version: %d", conf->op_version);
+                goto out;
+        }
+
+        gf_log (this->name, GF_LOG_INFO, "op-version not found in store, "
+                "setting it to minimum op-version : %d", GD_OP_VERSION_MIN);
+
+        /* If op-version is missing, set it to  GD_OP_VERSION_MIN */
+        conf->op_version = GD_OP_VERSION_MIN;
+        ret = 0;
+out:
         return ret;
 }
 
@@ -1577,7 +1683,6 @@ glusterd_retrieve_uuid ()
 
         priv = THIS->private;
 
-
         if (!priv->handle) {
                 snprintf (path, PATH_MAX, "%s/%s", priv->workdir,
                           GLUSTERD_INFO_FILE);
@@ -2793,8 +2898,15 @@ glusterd_restore ()
 
         this = THIS;
 
-        ret = glusterd_store_retrieve_volumes (this);
+        ret = glusterd_restore_op_version (this);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to restore op_version");
+                goto out;
+        }
 
+
+        ret = glusterd_store_retrieve_volumes (this);
         if (ret)
                 goto out;
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-store.h b/xlators/mgmt/glusterd/src/glusterd-store.h
index dd34250d7b4..1e9e599a429 100644
--- a/xlators/mgmt/glusterd/src/glusterd-store.h
+++ b/xlators/mgmt/glusterd/src/glusterd-store.h
@@ -95,9 +95,6 @@ int32_t
 glusterd_store_delete_volume (glusterd_volinfo_t *volinfo);
 
 int32_t
-glusterd_store_uuid ();
-
-int32_t
 glusterd_store_handle_new (char *path, glusterd_store_handle_t **handle);
 
 int32_t
@@ -134,4 +131,11 @@ glusterd_store_is_valid_brickpath (char *volname, char *brick);
 
 int32_t
 glusterd_store_perform_node_state_store (glusterd_volinfo_t *volinfo);
+
+int
+glusterd_retrieve_op_version (xlator_t *this, int *op_version);
+
+int
+glusterd_store_global_info (xlator_t *this);
+
 #endif
diff --git a/xlators/mgmt/glusterd/src/glusterd.c b/xlators/mgmt/glusterd/src/glusterd.c
index 164009bfe4d..cb5ac321c2f 100644
--- a/xlators/mgmt/glusterd/src/glusterd.c
+++ b/xlators/mgmt/glusterd/src/glusterd.c
@@ -51,6 +51,7 @@ extern struct rpcsvc_program gd_svc_mgmt_prog;
 extern struct rpcsvc_program gd_svc_peer_prog;
 extern struct rpcsvc_program gd_svc_cli_prog;
 extern struct rpc_clnt_program gd_brick_prog;
+extern struct rpcsvc_program glusterd_mgmt_hndsk_prog;
 
 rpcsvc_cbk_program_t glusterd_cbk_prog = {
         .progname  = "Gluster Callback",
@@ -58,6 +59,15 @@ rpcsvc_cbk_program_t glusterd_cbk_prog = {
         .progver   = GLUSTER_CBK_VERSION,
 };
 
+struct rpcsvc_program *all_programs[] = {
+        &gd_svc_peer_prog,
+        &gd_svc_cli_prog,
+        &gd_svc_mgmt_prog,
+        &gluster_pmap_prog,
+        &gluster_handshake_prog,
+        &glusterd_mgmt_hndsk_prog,
+};
+int rpcsvc_programs_count = (sizeof (all_programs) / sizeof (all_programs[0]));
 
 static int
 glusterd_opinfo_init ()
@@ -69,33 +79,38 @@ glusterd_opinfo_init ()
         return ret;
 }
 
+
 int
 glusterd_uuid_init ()
 {
         int             ret = -1;
+        xlator_t        *this = NULL;
         glusterd_conf_t *priv = NULL;
 
-        priv = THIS->private;
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
 
 	ret = glusterd_retrieve_uuid ();
 	if (ret == 0) {
 		uuid_copy (glusterd_uuid, priv->uuid);
-		gf_log ("glusterd", GF_LOG_INFO,
+		gf_log (this->name, GF_LOG_INFO,
 			"retrieved UUID: %s", uuid_utoa (priv->uuid));
 		return 0;
 	}
 
         uuid_generate (glusterd_uuid);
 
-        gf_log ("glusterd", GF_LOG_INFO,
-                        "generated UUID: %s", uuid_utoa (glusterd_uuid));
+        gf_log (this->name, GF_LOG_INFO, "generated UUID: %s",
+                uuid_utoa (glusterd_uuid));
+
         uuid_copy (priv->uuid, glusterd_uuid);
 
-        ret = glusterd_store_uuid ();
+        ret = glusterd_store_global_info (this);
 
         if (ret) {
-                gf_log ("glusterd", GF_LOG_ERROR,
-                          "Unable to store generated UUID");
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to store generated UUID");
                 return ret;
         }
 
@@ -778,6 +793,7 @@ init (xlator_t *this)
         char               cmd_log_filename [PATH_MAX] = {0,};
         int                first_time        = 0;
         char              *mountbroker_root  = NULL;
+        int                i                 = 0;
 
 #ifdef DEBUG
         char              *valgrind_str      = NULL;
@@ -920,39 +936,16 @@ init (xlator_t *this)
                 goto out;
         }
 
-        ret = glusterd_program_register (this, rpc, &gd_svc_peer_prog);
-        if (ret) {
-                goto out;
-        }
-
-        ret = glusterd_program_register (this, rpc, &gd_svc_cli_prog);
-        if (ret) {
-                rpcsvc_program_unregister (rpc, &gd_svc_peer_prog);
-                goto out;
-        }
-
-        ret = glusterd_program_register (this, rpc, &gd_svc_mgmt_prog);
-        if (ret) {
-                rpcsvc_program_unregister (rpc, &gd_svc_peer_prog);
-                rpcsvc_program_unregister (rpc, &gd_svc_cli_prog);
-                goto out;
-        }
-
-        ret = glusterd_program_register (this, rpc, &gluster_pmap_prog);
-        if (ret) {
-                rpcsvc_program_unregister (rpc, &gd_svc_peer_prog);
-                rpcsvc_program_unregister (rpc, &gd_svc_cli_prog);
-                rpcsvc_program_unregister (rpc, &gd_svc_mgmt_prog);
-                goto out;
-        }
+        for (i = 0; i < rpcsvc_programs_count; i++) {
+                ret = glusterd_program_register (this, rpc, all_programs[i]);
+                if (ret) {
+                        i--;
+                        for (; i >= 0; i--)
+                                rpcsvc_program_unregister (rpc,
+                                                           all_programs[i]);
 
-        ret = glusterd_program_register (this, rpc, &gluster_handshake_prog);
-        if (ret) {
-                rpcsvc_program_unregister (rpc, &gd_svc_peer_prog);
-                rpcsvc_program_unregister (rpc, &gluster_pmap_prog);
-                rpcsvc_program_unregister (rpc, &gd_svc_cli_prog);
-                rpcsvc_program_unregister (rpc, &gd_svc_mgmt_prog);
-                goto out;
+                        goto out;
+                }
         }
 
         conf = GF_CALLOC (1, sizeof (glusterd_conf_t),
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index b3fe1400aaa..0bf5beee884 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -91,6 +91,23 @@ typedef struct {
         gf_boolean_t            running;
 } nodesrv_t;
 
+#define GD_OP_VERSION_KEY     "operating-version"
+#define GD_MIN_OP_VERSION_KEY "minimum-operating-version"
+#define GD_MAX_OP_VERSION_KEY "maxium-operating-version"
+
+/* Gluster versions - OP-VERSION mapping
+ *
+ * 3.3.0                - 1
+ * 3.3.Next/3.Next      - 2
+ *
+ * (TODO: Change above comment once gluster version is finalised)
+ */
+#define GD_OP_VERSION_MIN  1 /* MIN is the fresh start op-version, mostly
+                                should not change */
+#define GD_OP_VERSION_MAX  2 /* MAX VERSION is the maximum count in VME table,
+                                should keep changing with introduction of newer
+                                versions */
+
 typedef struct {
         struct _volfile_ctx *volfile;
 	pthread_mutex_t   mutex;
@@ -110,15 +127,21 @@ typedef struct {
         gf_timer_t *timer;
         glusterd_sm_tr_log_t op_sm_log;
         struct rpc_clnt_program *gfs_mgmt;
+
         struct list_head mount_specs;
 #ifdef DEBUG
         gf_boolean_t      valgrind;
 #endif
         pthread_t       brick_thread;
         void           *hooks_priv;
-        xlator_t       *xl;  /* Should be set to 'THIS' before creating thread */
+        xlator_t       *xl; /* Should be set to 'THIS' before creating thread */
+
+        /* need for proper handshake_t */
+        int             op_version; /* Starts with 1 for 3.3.0 */
+
 } glusterd_conf_t;
 
+
 typedef enum gf_brick_status {
         GF_BRICK_STOPPED,
         GF_BRICK_STARTED,
@@ -556,8 +579,8 @@ glusterd_volume_txn (rpcsvc_request_t *req, char *volname, int flags,
                      glusterd_op_t op);
 
 int
-glusterd_peer_handshake (xlator_t *this, struct rpc_clnt *rpc,
-                         glusterd_peerctx_t *peerctx);
+glusterd_peer_dump_version (xlator_t *this, struct rpc_clnt *rpc,
+                            glusterd_peerctx_t *peerctx);
 
 int
 glusterd_validate_reconfopts (glusterd_volinfo_t *volinfo, dict_t *val_dict, char **op_errstr);