86 files changed, 103530 insertions, 22831 deletions

diff --git a/xlators/mgmt/glusterd/src/Makefile.am b/xlators/mgmt/glusterd/src/Makefile.am
index 714b78e6244..685beb42d27 100644
--- a/xlators/mgmt/glusterd/src/Makefile.am
+++ b/xlators/mgmt/glusterd/src/Makefile.am
@@ -1,23 +1,79 @@
+if WITH_SERVER
 xlator_LTLIBRARIES = glusterd.la
+endif
+
 xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/mgmt
-glusterd_la_LDFLAGS = -module -avoidversion $(LIBXML2_LIBS)
-glusterd_la_SOURCES = glusterd.c glusterd-handler.c glusterd-sm.c glusterd-op-sm.c \
-	glusterd-utils.c glusterd-rpc-ops.c glusterd-store.c glusterd-handshake.c \
-	glusterd-pmap.c glusterd-volgen.c glusterd-rebalance.c
+glusterd_la_CPPFLAGS = $(AM_CPPFLAGS) \
+	-DFILTERDIR=\"$(libdir)/glusterfs/$(PACKAGE_VERSION)/filter\" \
+	-DXLATORDIR=\"$(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator\" \
+	-I$(top_srcdir)/libglusterd/src/
+
+glusterd_la_LDFLAGS = -module $(GF_XLATOR_DEFAULT_LDFLAGS)
+glusterd_la_SOURCES = glusterd.c glusterd-handler.c glusterd-sm.c \
+	glusterd-op-sm.c glusterd-utils.c glusterd-rpc-ops.c \
+	glusterd-store.c glusterd-handshake.c glusterd-pmap.c \
+	glusterd-volgen.c glusterd-rebalance.c \
+	glusterd-quota.c glusterd-bitrot.c glusterd-geo-rep.c \
+	glusterd-replace-brick.c glusterd-log-ops.c \
+	glusterd-volume-ops.c glusterd-brick-ops.c glusterd-mountbroker.c \
+	glusterd-syncop.c glusterd-hooks.c glusterd-volume-set.c \
+	glusterd-locks.c glusterd-snapshot.c glusterd-mgmt-handler.c \
+	glusterd-mgmt.c glusterd-peer-utils.c glusterd-statedump.c \
+	glusterd-snapshot-utils.c glusterd-conn-mgmt.c \
+	glusterd-proc-mgmt.c glusterd-svc-mgmt.c \
+	glusterd-nfs-svc.c glusterd-quotad-svc.c glusterd-svc-helper.c \
+	glusterd-conn-helper.c glusterd-snapd-svc.c glusterd-snapd-svc-helper.c \
+	glusterd-bitd-svc.c glusterd-scrub-svc.c glusterd-server-quorum.c \
+	glusterd-reset-brick.c glusterd-shd-svc.c glusterd-shd-svc-helper.c \
+        glusterd-gfproxyd-svc.c glusterd-gfproxyd-svc-helper.c glusterd-ganesha.c \
+	$(CONTRIBDIR)/mount/mntent.c
+
+glusterd_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la \
+	$(top_builddir)/libglusterd/src/libglusterd.la \
+	$(top_builddir)/rpc/xdr/src/libgfxdr.la \
+	$(top_builddir)/rpc/rpc-lib/src/libgfrpc.la \
+	$(XML_LIBS) -lcrypto $(URCU_LIBS) $(URCU_CDS_LIBS) $(LIB_DL) $(GF_XLATOR_MGNT_LIBADD)
 
-glusterd_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la\
-		     $(top_builddir)/rpc/xdr/src/libgfxdr.la\
-		     $(top_builddir)/rpc/rpc-lib/src/libgfrpc.la
+noinst_HEADERS = glusterd.h glusterd-utils.h glusterd-op-sm.h \
+	glusterd-sm.h glusterd-store.h glusterd-mem-types.h \
+	glusterd-pmap.h glusterd-volgen.h glusterd-mountbroker.h \
+	glusterd-syncop.h glusterd-hooks.h glusterd-locks.h glusterd-quota.h \
+	glusterd-mgmt.h glusterd-messages.h glusterd-peer-utils.h \
+	glusterd-statedump.h glusterd-snapshot-utils.h glusterd-geo-rep.h \
+	glusterd-conn-mgmt.h glusterd-conn-helper.h glusterd-proc-mgmt.h \
+	glusterd-svc-mgmt.h glusterd-nfs-svc.h \
+	glusterd-quotad-svc.h glusterd-svc-helper.h glusterd-snapd-svc.h \
+	glusterd-snapd-svc-helper.h glusterd-rcu.h glusterd-bitd-svc.h \
+	glusterd-scrub-svc.h glusterd-server-quorum.h glusterd-errno.h \
+        glusterd-shd-svc.h glusterd-shd-svc-helper.h \
+        glusterd-gfproxyd-svc.h glusterd-gfproxyd-svc-helper.h \
+	$(CONTRIBDIR)/userspace-rcu/rculist-extra.h \
+	$(CONTRIBDIR)/mount/mntent_compat.h
 
-noinst_HEADERS = glusterd.h glusterd-utils.h glusterd-op-sm.h glusterd-sm.h \
-	glusterd-store.h glusterd-mem-types.h glusterd-pmap.h glusterd-volgen.h
+AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \
+	-I$(top_srcdir)/rpc/xdr/src -I$(top_builddir)/rpc/xdr/src \
+	-I$(CONTRIBDIR)/rbtree -I$(top_srcdir)/rpc/rpc-lib/src \
+	-I$(CONTRIBDIR)/mount -I$(CONTRIBDIR)/userspace-rcu \
+	-DSBIN_DIR=\"$(sbindir)\" -DDATADIR=\"$(localstatedir)\" \
+	-DGSYNCD_PREFIX=\"$(GLUSTERFS_LIBEXECDIR)\" \
+	-DCONFDIR=\"$(localstatedir)/run/gluster/shared_storage/nfs-ganesha\" \
+	-DGANESHA_PREFIX=\"$(libexecdir)/ganesha\" \
+	-DSYNCDAEMON_COMPILE=$(SYNCDAEMON_COMPILE) \
+	-I$(top_srcdir)/libglusterd/src/
 
-AM_CFLAGS = -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -Wall -D$(GF_HOST_OS)\
-	-I$(top_srcdir)/libglusterfs/src -shared -nostartfiles $(GF_CFLAGS)\
-	-I$(rpclibdir) -L$(xlatordir)/ -I$(CONTRIBDIR)/rbtree -I$(top_srcdir)/rpc/xdr/src\
-	-I$(top_srcdir)/rpc/rpc-lib/src -I$(CONTRIBDIR)/uuid -I$(top_srcdir)/contrib/md5 -DGFS_PREFIX=\"$(prefix)\" \
-        -DDATADIR=\"$(localstatedir)\" -DGSYNCD_PREFIX=\"$(libexecdir)/glusterfs\"\
-        -DSYNCDAEMON_COMPILE=$(SYNCDAEMON_COMPILE) $(LIBXML2_CFLAGS)
 
+AM_CFLAGS = -Wall $(GF_CFLAGS) $(URCU_CFLAGS) $(URCU_CDS_CFLAGS) $(XML_CFLAGS)
+
+AM_LDFLAGS = -L$(xlatordir) $(URCU_LIBS) $(URCU_CDS_LIBS)
 
 CLEANFILES =
+
+install-data-hook:
+if WITH_SERVER
+if GF_INSTALL_GLUSTERD_WORKDIR
+	$(mkdir_p) $(DESTDIR)$(GLUSTERD_WORKDIR)
+	(stat $(DESTDIR)$(sysconfdir)/glusterd && \
+	    mv $(DESTDIR)$(sysconfdir)/glusterd $(DESTDIR)$(GLUSTERD_WORKDIR)) || true;
+	(ln -sf $(DESTDIR)$(GLUSTERD_WORKDIR) $(sysconfdir)/glusterd) || true;
+endif
+endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-bitd-svc.c b/xlators/mgmt/glusterd/src/glusterd-bitd-svc.c
new file mode 100644
index 00000000000..6adb799b18f
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-bitd-svc.c
@@ -0,0 +1,206 @@
+/*
+  Copyright (c) 2006-2012 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+#include <glusterfs/globals.h>
+#include <glusterfs/run.h>
+#include "glusterd.h"
+#include "glusterd-utils.h"
+#include "glusterd-volgen.h"
+#include "glusterd-bitd-svc.h"
+#include "glusterd-svc-helper.h"
+
+void
+glusterd_bitdsvc_build(glusterd_svc_t *svc)
+{
+    svc->manager = glusterd_bitdsvc_manager;
+    svc->start = glusterd_bitdsvc_start;
+    svc->stop = glusterd_bitdsvc_stop;
+}
+
+int
+glusterd_bitdsvc_init(glusterd_svc_t *svc)
+{
+    return glusterd_svc_init(svc, bitd_svc_name);
+}
+
+static int
+glusterd_bitdsvc_create_volfile()
+{
+    char filepath[PATH_MAX] = {
+        0,
+    };
+    int ret = -1;
+    glusterd_conf_t *conf = NULL;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    glusterd_svc_build_volfile_path(bitd_svc_name, conf->workdir, filepath,
+                                    sizeof(filepath));
+
+    ret = glusterd_create_global_volfile(build_bitd_graph, filepath, NULL);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOLFILE_CREATE_FAIL,
+               "Failed to create volfile");
+        goto out;
+    }
+
+out:
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+
+    return ret;
+}
+
+int
+glusterd_bitdsvc_manager(glusterd_svc_t *svc, void *data, int flags)
+{
+    int ret = 0;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    if (!svc->inited) {
+        ret = glusterd_bitdsvc_init(svc);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BITD_INIT_FAIL,
+                   "Failed to init "
+                   "bitd service");
+            goto out;
+        } else {
+            svc->inited = _gf_true;
+            gf_msg_debug(this->name, 0,
+                         "BitD service "
+                         "initialized");
+        }
+    }
+
+    if (glusterd_should_i_stop_bitd()) {
+        ret = svc->stop(svc, SIGTERM);
+    } else {
+        ret = glusterd_bitdsvc_create_volfile();
+        if (ret)
+            goto out;
+
+        ret = svc->stop(svc, SIGKILL);
+        if (ret)
+            goto out;
+
+        ret = svc->start(svc, flags);
+        if (ret)
+            goto out;
+
+        ret = glusterd_conn_connect(&(svc->conn));
+        if (ret)
+            goto out;
+    }
+
+out:
+    if (ret)
+        gf_event(EVENT_SVC_MANAGER_FAILED, "svc_name=%s", svc->name);
+
+    gf_msg_debug(THIS->name, 0, "Returning %d", ret);
+
+    return ret;
+}
+
+int
+glusterd_bitdsvc_start(glusterd_svc_t *svc, int flags)
+{
+    int ret = -1;
+    dict_t *cmdict = NULL;
+
+    cmdict = dict_new();
+    if (!cmdict)
+        goto error_return;
+
+    ret = dict_set_str(cmdict, "cmdarg0", "--global-timer-wheel");
+    if (ret)
+        goto dealloc_dict;
+
+    ret = glusterd_svc_start(svc, flags, cmdict);
+
+dealloc_dict:
+    dict_unref(cmdict);
+error_return:
+    return ret;
+}
+
+int
+glusterd_bitdsvc_stop(glusterd_svc_t *svc, int sig)
+{
+    return glusterd_svc_stop(svc, sig);
+}
+
+int
+glusterd_bitdsvc_reconfigure()
+{
+    int ret = -1;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+    gf_boolean_t identical = _gf_false;
+
+    this = THIS;
+    GF_VALIDATE_OR_GOTO("glusterd", this, out);
+
+    priv = this->private;
+    GF_VALIDATE_OR_GOTO(this->name, priv, out);
+
+    if (glusterd_should_i_stop_bitd())
+        goto manager;
+    /*
+     * Check both OLD and NEW volfiles, if they are SAME by size
+     * and cksum i.e. "character-by-character". If YES, then
+     * NOTHING has been changed, just return.
+     */
+    ret = glusterd_svc_check_volfile_identical(priv->bitd_svc.name,
+                                               build_bitd_graph, &identical);
+    if (ret)
+        goto out;
+    if (identical) {
+        ret = 0;
+        goto out;
+    }
+
+    /*
+     * They are not identical. Find out if the topology is changed
+     * OR just the volume options. If just the options which got
+     * changed, then inform the xlator to reconfigure the options.
+     */
+    identical = _gf_false; /* RESET the FLAG */
+    ret = glusterd_svc_check_topology_identical(priv->bitd_svc.name,
+                                                build_bitd_graph, &identical);
+    if (ret)
+        goto out; /*not able to compare due to some corruption */
+
+    /* Topology is not changed, but just the options. But write the
+     * options to bitd volfile, so that bitd will be reconfigured.
+     */
+    if (identical) {
+        ret = glusterd_bitdsvc_create_volfile();
+        if (ret == 0) { /* Only if above PASSES */
+            ret = glusterd_fetchspec_notify(THIS);
+        }
+        goto out;
+    }
+
+manager:
+    /*
+     * bitd volfile's topology has been changed. bitd server needs
+     * to be RESTARTED to ACT on the changed volfile.
+     */
+    ret = priv->bitd_svc.manager(&(priv->bitd_svc), NULL, PROC_START_NO_WAIT);
+
+out:
+    gf_msg_debug(this ? this->name : "glusterd", 0, "Returning %d", ret);
+    return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-bitd-svc.h b/xlators/mgmt/glusterd/src/glusterd-bitd-svc.h
new file mode 100644
index 00000000000..1bff084a9a8
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-bitd-svc.h
@@ -0,0 +1,40 @@
+/*
+  Copyright (c) 2006-2012 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
+*/
+
+#ifndef _GLUSTERD_BITD_SVC_H_
+#define _GLUSTERD_BITD_SVC_H_
+
+#include "glusterd-svc-mgmt.h"
+
+#define bitd_svc_name "bitd"
+
+void
+glusterd_bitdsvc_build(glusterd_svc_t *svc);
+
+int
+glusterd_bitdsvc_init(glusterd_svc_t *svc);
+
+int
+glusterd_bitdsvc_manager(glusterd_svc_t *svc, void *data, int flags);
+
+int
+glusterd_bitdsvc_start(glusterd_svc_t *svc, int flags);
+
+int
+glusterd_bitdsvc_stop(glusterd_svc_t *svc, int sig);
+
+int
+glusterd_bitdsvc_reconfigure();
+
+void
+glusterd_bitdsvc_build_volfile_path(char *server, char *workdir, char *volfile,
+                                    size_t len);
+
+#endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-bitrot.c b/xlators/mgmt/glusterd/src/glusterd-bitrot.c
new file mode 100644
index 00000000000..37429fe9214
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-bitrot.c
@@ -0,0 +1,822 @@
+/*
+   Copyright (c) 2011-2012 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+ */
+
+#include <glusterfs/common-utils.h>
+#include "cli1-xdr.h"
+#include "xdr-generic.h"
+#include "glusterd.h"
+#include "glusterd-op-sm.h"
+#include "glusterd-store.h"
+#include "glusterd-utils.h"
+#include "glusterd-volgen.h"
+#include <glusterfs/run.h>
+#include <glusterfs/syscall.h>
+#include <glusterfs/byte-order.h>
+#include <glusterfs/compat-errno.h>
+#include "glusterd-scrub-svc.h"
+#include "glusterd-messages.h"
+
+#include <sys/wait.h>
+#include <dlfcn.h>
+
+const char *gd_bitrot_op_list[GF_BITROT_OPTION_TYPE_MAX] = {
+    [GF_BITROT_OPTION_TYPE_NONE] = "none",
+    [GF_BITROT_OPTION_TYPE_ENABLE] = "enable",
+    [GF_BITROT_OPTION_TYPE_DISABLE] = "disable",
+    [GF_BITROT_OPTION_TYPE_SCRUB_THROTTLE] = "scrub-throttle",
+    [GF_BITROT_OPTION_TYPE_SCRUB_FREQ] = "scrub-frequency",
+    [GF_BITROT_OPTION_TYPE_SCRUB] = "scrub",
+    [GF_BITROT_OPTION_TYPE_EXPIRY_TIME] = "expiry-time",
+    [GF_BITROT_OPTION_TYPE_SIGNER_THREADS] = "signer-threads",
+};
+
+int
+__glusterd_handle_bitrot(rpcsvc_request_t *req)
+{
+    int32_t ret = -1;
+    gf_cli_req cli_req = {{
+        0,
+    }};
+    dict_t *dict = NULL;
+    glusterd_op_t cli_op = GD_OP_BITROT;
+    char *volname = NULL;
+    char *scrub = NULL;
+    int32_t type = 0;
+    char msg[256] = {
+        0,
+    };
+    xlator_t *this = NULL;
+    glusterd_conf_t *conf = NULL;
+
+    GF_ASSERT(req);
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        req->rpc_err = GARBAGE_ARGS;
+        goto out;
+    }
+
+    if (cli_req.dict.dict_len) {
+        /* Unserialize the dictionary */
+        dict = dict_new();
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to "
+                   "unserialize req-buffer to dictionary");
+            snprintf(msg, sizeof(msg),
+                     "Unable to decode the "
+                     "command");
+            goto out;
+        } else {
+            dict->extra_stdfree = cli_req.dict.dict_val;
+        }
+    }
+
+    ret = dict_get_str(dict, "volname", &volname);
+    if (ret) {
+        snprintf(msg, sizeof(msg), "Unable to get volume name");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name, "
+               "while handling bitrot command");
+        goto out;
+    }
+
+    ret = dict_get_int32(dict, "type", &type);
+    if (ret) {
+        snprintf(msg, sizeof(msg), "Unable to get type of command");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get type of cmd, "
+               "while handling bitrot command");
+        goto out;
+    }
+
+    if (conf->op_version < GD_OP_VERSION_3_7_0) {
+        snprintf(msg, sizeof(msg),
+                 "Cannot execute command. The "
+                 "cluster is operating at version %d. Bitrot command "
+                 "%s is unavailable in this version",
+                 conf->op_version, gd_bitrot_op_list[type]);
+        ret = -1;
+        goto out;
+    }
+
+    if (type == GF_BITROT_CMD_SCRUB_STATUS) {
+        /* Backward compatibility handling for scrub status command*/
+        if (conf->op_version < GD_OP_VERSION_3_7_7) {
+            snprintf(msg, sizeof(msg),
+                     "Cannot execute command. "
+                     "The cluster is operating at version %d. "
+                     "Bitrot scrub status command unavailable in "
+                     "this version",
+                     conf->op_version);
+            ret = -1;
+            goto out;
+        }
+
+        ret = dict_get_str(dict, "scrub-value", &scrub);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
+                   "Failed to get scrub value.");
+            ret = -1;
+            goto out;
+        }
+
+        if (!strncmp(scrub, "status", SLEN("status"))) {
+            ret = glusterd_op_begin_synctask(req, GD_OP_SCRUB_STATUS, dict);
+            goto out;
+        }
+    }
+
+    if (type == GF_BITROT_CMD_SCRUB_ONDEMAND) {
+        /* Backward compatibility handling for scrub status command*/
+        if (conf->op_version < GD_OP_VERSION_3_9_0) {
+            snprintf(msg, sizeof(msg),
+                     "Cannot execute command. "
+                     "The cluster is operating at version %d. "
+                     "Bitrot scrub ondemand command unavailable in "
+                     "this version",
+                     conf->op_version);
+            ret = -1;
+            goto out;
+        }
+
+        ret = dict_get_str(dict, "scrub-value", &scrub);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
+                   "Failed to get scrub value.");
+            ret = -1;
+            goto out;
+        }
+
+        if (!strncmp(scrub, "ondemand", SLEN("ondemand"))) {
+            ret = glusterd_op_begin_synctask(req, GD_OP_SCRUB_ONDEMAND, dict);
+            goto out;
+        }
+    }
+
+    ret = glusterd_op_begin_synctask(req, GD_OP_BITROT, dict);
+
+out:
+    if (ret) {
+        if (msg[0] == '\0')
+            snprintf(msg, sizeof(msg), "Bitrot operation failed");
+        ret = glusterd_op_send_cli_response(cli_op, ret, 0, req, dict, msg);
+    }
+
+    return ret;
+}
+
+int
+glusterd_handle_bitrot(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_bitrot);
+}
+
+static int
+glusterd_bitrot_scrub_throttle(glusterd_volinfo_t *volinfo, dict_t *dict,
+                               char *key, char **op_errstr)
+{
+    int32_t ret = -1;
+    char *scrub_throttle = NULL;
+    char *option = NULL;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_str(dict, "scrub-throttle-value", &scrub_throttle);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to fetch scrub-"
+               "throttle value");
+        goto out;
+    }
+
+    option = gf_strdup(scrub_throttle);
+    ret = dict_set_dynstr(volinfo->dict, key, option);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option %s", key);
+        goto out;
+    }
+
+    ret = glusterd_scrubsvc_reconfigure();
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SCRUBSVC_RECONF_FAIL,
+               "Failed to reconfigure scrub "
+               "services");
+        goto out;
+    }
+
+out:
+    return ret;
+}
+
+static int
+glusterd_bitrot_scrub_freq(glusterd_volinfo_t *volinfo, dict_t *dict, char *key,
+                           char **op_errstr)
+{
+    int32_t ret = -1;
+    char *scrub_freq = NULL;
+    xlator_t *this = NULL;
+    char *option = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_str(dict, "scrub-frequency-value", &scrub_freq);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to fetch scrub-"
+               "freq value");
+        goto out;
+    }
+
+    option = gf_strdup(scrub_freq);
+    ret = dict_set_dynstr(volinfo->dict, key, option);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option %s", key);
+        goto out;
+    }
+
+    ret = glusterd_scrubsvc_reconfigure();
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SCRUBSVC_RECONF_FAIL,
+               "Failed to reconfigure scrub "
+               "services");
+        goto out;
+    }
+
+out:
+    return ret;
+}
+
+static int
+glusterd_bitrot_scrub(glusterd_volinfo_t *volinfo, dict_t *dict, char *key,
+                      char **op_errstr)
+{
+    int32_t ret = -1;
+    char *scrub_value = NULL;
+    xlator_t *this = NULL;
+    char *option = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_str(dict, "scrub-value", &scrub_value);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED,
+               "Unable to fetch scrub"
+               "value");
+        goto out;
+    }
+
+    if (!strcmp(scrub_value, "resume")) {
+        option = gf_strdup("Active");
+    } else {
+        option = gf_strdup(scrub_value);
+    }
+
+    ret = dict_set_dynstr(volinfo->dict, key, option);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option %s", key);
+        goto out;
+    }
+
+    ret = glusterd_scrubsvc_reconfigure();
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SCRUBSVC_RECONF_FAIL,
+               "Failed to reconfigure scrub "
+               "services");
+        goto out;
+    }
+
+out:
+    return ret;
+}
+
+static int
+glusterd_bitrot_expiry_time(glusterd_volinfo_t *volinfo, dict_t *dict,
+                            char *key, char **op_errstr)
+{
+    int32_t ret = -1;
+    uint32_t expiry_time = 0;
+    xlator_t *this = NULL;
+    char dkey[32] = {
+        0,
+    };
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_uint32(dict, "expiry-time", &expiry_time);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get bitrot expiry"
+               " timer value.");
+        goto out;
+    }
+
+    snprintf(dkey, sizeof(dkey), "%d", expiry_time);
+
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, key, dkey);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option %s", key);
+        goto out;
+    }
+
+    ret = glusterd_bitdsvc_reconfigure();
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BITDSVC_RECONF_FAIL,
+               "Failed to reconfigure bitrot"
+               "services");
+        goto out;
+    }
+out:
+    return ret;
+}
+
+static gf_boolean_t
+is_bitd_configure_noop(xlator_t *this, glusterd_volinfo_t *volinfo)
+{
+    gf_boolean_t noop = _gf_true;
+    glusterd_brickinfo_t *brickinfo = NULL;
+
+    if (!glusterd_is_bitrot_enabled(volinfo))
+        goto out;
+    else if (volinfo->status != GLUSTERD_STATUS_STARTED)
+        goto out;
+    else {
+        cds_list_for_each_entry(brickinfo, &volinfo->bricks, brick_list)
+        {
+            if (!glusterd_is_local_brick(this, volinfo, brickinfo))
+                continue;
+            noop = _gf_false;
+            return noop;
+        }
+    }
+out:
+    return noop;
+}
+
+static int
+glusterd_bitrot_signer_threads(glusterd_volinfo_t *volinfo, dict_t *dict,
+                               char *key, char **op_errstr)
+{
+    int32_t ret = -1;
+    uint32_t signer_th_count = 0;
+    uint32_t existing_th_count = 0;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+    char dkey[32] = {
+        0,
+    };
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    priv = this->private;
+    GF_VALIDATE_OR_GOTO(this->name, priv, out);
+
+    ret = dict_get_uint32(dict, "signer-threads", &signer_th_count);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get bitrot signer thread count.");
+        goto out;
+    }
+
+    ret = dict_get_uint32(volinfo->dict, key, &existing_th_count);
+    if (ret == 0 && signer_th_count == existing_th_count) {
+        goto out;
+    }
+
+    snprintf(dkey, sizeof(dkey), "%d", signer_th_count);
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, key, dkey);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option %s", key);
+        goto out;
+    }
+
+    if (!is_bitd_configure_noop(this, volinfo)) {
+        ret = priv->bitd_svc.manager(&(priv->bitd_svc), NULL,
+                                     PROC_START_NO_WAIT);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BITDSVC_RECONF_FAIL,
+                   "Failed to reconfigure bitrot services");
+            goto out;
+        }
+    }
+out:
+    return ret;
+}
+
+static int
+glusterd_bitrot_enable(glusterd_volinfo_t *volinfo, char **op_errstr)
+{
+    int32_t ret = -1;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    GF_VALIDATE_OR_GOTO(this->name, volinfo, out);
+    GF_VALIDATE_OR_GOTO(this->name, op_errstr, out);
+
+    if (glusterd_is_volume_started(volinfo) == 0) {
+        *op_errstr = gf_strdup(
+            "Volume is stopped, start volume "
+            "to enable bitrot.");
+        ret = -1;
+        goto out;
+    }
+
+    ret = glusterd_is_bitrot_enabled(volinfo);
+    if (ret) {
+        *op_errstr = gf_strdup("Bitrot is already enabled");
+        ret = -1;
+        goto out;
+    }
+
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, VKEY_FEATURES_BITROT, "on");
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "dict set failed");
+        goto out;
+    }
+
+    /*Once bitrot is enable scrubber should be in Active state*/
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, "features.scrub", "Active");
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set option "
+               "features.scrub value");
+        goto out;
+    }
+
+    ret = 0;
+out:
+    if (ret && op_errstr && !*op_errstr)
+        gf_asprintf(op_errstr,
+                    "Enabling bitrot on volume %s has been "
+                    "unsuccessful",
+                    volinfo->volname);
+    return ret;
+}
+
+static int
+glusterd_bitrot_disable(glusterd_volinfo_t *volinfo, char **op_errstr)
+{
+    int32_t ret = -1;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_VALIDATE_OR_GOTO("glusterd", this, out);
+
+    GF_VALIDATE_OR_GOTO(this->name, volinfo, out);
+    GF_VALIDATE_OR_GOTO(this->name, op_errstr, out);
+
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, VKEY_FEATURES_BITROT,
+                                     "off");
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "dict set failed");
+        goto out;
+    }
+
+    /*Once bitrot disabled scrubber should be Inactive state*/
+    ret = dict_set_dynstr_with_alloc(volinfo->dict, "features.scrub",
+                                     "Inactive");
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set "
+               "features.scrub value");
+        goto out;
+    }
+
+    ret = 0;
+out:
+    if (ret && op_errstr && !*op_errstr)
+        gf_asprintf(op_errstr,
+                    "Disabling bitrot on volume %s has "
+                    "been unsuccessful",
+                    volinfo->volname);
+    return ret;
+}
+
+gf_boolean_t
+glusterd_should_i_stop_bitd()
+{
+    glusterd_conf_t *conf = THIS->private;
+    glusterd_volinfo_t *volinfo = NULL;
+    gf_boolean_t stopped = _gf_true;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    cds_list_for_each_entry(volinfo, &conf->volumes, vol_list)
+    {
+        if (!glusterd_is_bitrot_enabled(volinfo))
+            continue;
+        else if (volinfo->status != GLUSTERD_STATUS_STARTED)
+            continue;
+        else {
+            cds_list_for_each_entry(brickinfo, &volinfo->bricks, brick_list)
+            {
+                if (!glusterd_is_local_brick(this, volinfo, brickinfo))
+                    continue;
+                stopped = _gf_false;
+                return stopped;
+            }
+
+            /* Before stopping bitrot/scrubber daemon check
+             * other volume also whether respective volume
+             * host a brick from this node or not.*/
+            continue;
+        }
+    }
+
+    return stopped;
+}
+
+static int
+glusterd_manage_bitrot(int opcode)
+{
+    int ret = -1;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    switch (opcode) {
+        case GF_BITROT_OPTION_TYPE_ENABLE:
+        case GF_BITROT_OPTION_TYPE_DISABLE:
+            ret = priv->bitd_svc.manager(&(priv->bitd_svc), NULL,
+                                         PROC_START_NO_WAIT);
+            if (ret)
+                break;
+            ret = priv->scrub_svc.manager(&(priv->scrub_svc), NULL,
+                                          PROC_START_NO_WAIT);
+            break;
+        default:
+            ret = 0;
+            break;
+    }
+
+    return ret;
+}
+
+int
+glusterd_op_bitrot(dict_t *dict, char **op_errstr, dict_t *rsp_dict)
+{
+    glusterd_volinfo_t *volinfo = NULL;
+    int32_t ret = -1;
+    char *volname = NULL;
+    int type = -1;
+    glusterd_conf_t *priv = NULL;
+    xlator_t *this = NULL;
+
+    GF_ASSERT(dict);
+    GF_ASSERT(op_errstr);
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = dict_get_str(dict, "volname", &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_asprintf(op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
+        goto out;
+    }
+
+    ret = dict_get_int32(dict, "type", &type);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get type from "
+               "dict");
+        goto out;
+    }
+
+    switch (type) {
+        case GF_BITROT_OPTION_TYPE_ENABLE:
+            ret = glusterd_bitrot_enable(volinfo, op_errstr);
+            if (ret < 0)
+                goto out;
+            break;
+
+        case GF_BITROT_OPTION_TYPE_DISABLE:
+            ret = glusterd_bitrot_disable(volinfo, op_errstr);
+            if (ret < 0)
+                goto out;
+
+            break;
+
+        case GF_BITROT_OPTION_TYPE_SCRUB_THROTTLE:
+            ret = glusterd_bitrot_scrub_throttle(
+                volinfo, dict, "features.scrub-throttle", op_errstr);
+            if (ret)
+                goto out;
+            break;
+
+        case GF_BITROT_OPTION_TYPE_SCRUB_FREQ:
+            ret = glusterd_bitrot_scrub_freq(volinfo, dict,
+                                             "features.scrub-freq", op_errstr);
+            if (ret)
+                goto out;
+            break;
+
+        case GF_BITROT_OPTION_TYPE_SCRUB:
+            ret = glusterd_bitrot_scrub(volinfo, dict, "features.scrub",
+                                        op_errstr);
+            if (ret)
+                goto out;
+            break;
+
+        case GF_BITROT_OPTION_TYPE_EXPIRY_TIME:
+            ret = glusterd_bitrot_expiry_time(
+                volinfo, dict, "features.expiry-time", op_errstr);
+            if (ret)
+                goto out;
+            break;
+
+        case GF_BITROT_OPTION_TYPE_SIGNER_THREADS:
+            ret = glusterd_bitrot_signer_threads(
+                volinfo, dict, "features.signer-threads", op_errstr);
+            if (ret)
+                goto out;
+            break;
+
+        case GF_BITROT_CMD_SCRUB_STATUS:
+        case GF_BITROT_CMD_SCRUB_ONDEMAND:
+            break;
+
+        default:
+            gf_asprintf(op_errstr,
+                        "Bitrot command failed. Invalid "
+                        "opcode");
+            ret = -1;
+            goto out;
+    }
+
+    ret = glusterd_manage_bitrot(type);
+    if (ret)
+        goto out;
+
+    ret = glusterd_create_volfiles_and_notify_services(volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOLFILE_CREATE_FAIL,
+               "Unable to re-create "
+               "volfiles");
+        ret = -1;
+        goto out;
+    }
+
+    ret = glusterd_store_volinfo(volinfo, GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+    if (ret) {
+        gf_msg_debug(this->name, 0,
+                     "Failed to store volinfo for "
+                     "bitrot");
+        goto out;
+    }
+
+out:
+    return ret;
+}
+
+int
+glusterd_op_stage_bitrot(dict_t *dict, char **op_errstr, dict_t *rsp_dict)
+{
+    int ret = 0;
+    char *volname = NULL;
+    char *scrub_cmd = NULL;
+    char *scrub_cmd_from_dict = NULL;
+    char msg[2048] = {
+        0,
+    };
+    int type = 0;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    GF_ASSERT(dict);
+    GF_ASSERT(op_errstr);
+
+    ret = dict_get_str(dict, "volname", &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_asprintf(op_errstr, FMTSTR_CHECK_VOL_EXISTS, volname);
+        goto out;
+    }
+
+    if (!glusterd_is_volume_started(volinfo)) {
+        *op_errstr = gf_strdup(
+            "Volume is stopped, start volume "
+            "before executing bit rot command.");
+        ret = -1;
+        goto out;
+    }
+
+    ret = dict_get_int32(dict, "type", &type);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get type for "
+               "operation");
+
+        *op_errstr = gf_strdup(
+            "Staging stage failed for bitrot "
+            "operation.");
+        goto out;
+    }
+
+    if ((GF_BITROT_OPTION_TYPE_ENABLE != type) &&
+        (glusterd_is_bitrot_enabled(volinfo) == 0)) {
+        ret = -1;
+        gf_asprintf(op_errstr, "Bitrot is not enabled on volume %s", volname);
+        goto out;
+    }
+
+    if ((GF_BITROT_OPTION_TYPE_SCRUB == type)) {
+        ret = dict_get_str(volinfo->dict, "features.scrub",
+                           &scrub_cmd_from_dict);
+        if (!ret) {
+            ret = dict_get_str(dict, "scrub-value", &scrub_cmd);
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                       "Unable to "
+                       "get scrub-value");
+                *op_errstr = gf_strdup(
+                    "Staging failed for "
+                    "bitrot operation. "
+                    "Please check log file"
+                    " for more details.");
+                goto out;
+            }
+            /* If scrubber is resume then value of scrubber will be
+             * "Active" in the dictionary. */
+            if (!strcmp(scrub_cmd_from_dict, scrub_cmd) ||
+                (!strncmp("Active", scrub_cmd_from_dict, SLEN("Active")) &&
+                 !strncmp("resume", scrub_cmd, SLEN("resume")))) {
+                snprintf(msg, sizeof(msg),
+                         "Scrub is already"
+                         " %sd for volume %s",
+                         scrub_cmd, volinfo->volname);
+                *op_errstr = gf_strdup(msg);
+                ret = -1;
+                goto out;
+            }
+        }
+        ret = 0;
+    }
+
+out:
+    if (ret && op_errstr && *op_errstr)
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_OP_STAGE_BITROT_FAIL, "%s",
+               *op_errstr);
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+
+    return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-brick-ops.c b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
new file mode 100644
index 00000000000..e56cd0e6c74
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-brick-ops.c
@@ -0,0 +1,2796 @@
+/*
+   Copyright (c) 2011-2012 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#include <glusterfs/common-utils.h>
+#include "cli1-xdr.h"
+#include "xdr-generic.h"
+#include "glusterd.h"
+#include "glusterd-op-sm.h"
+#include "glusterd-geo-rep.h"
+#include "glusterd-store.h"
+#include "glusterd-mgmt.h"
+#include "glusterd-utils.h"
+#include "glusterd-volgen.h"
+#include "glusterd-svc-helper.h"
+#include "glusterd-messages.h"
+#include "glusterd-server-quorum.h"
+#include <glusterfs/run.h>
+#include <glusterfs/syscall.h>
+#include <sys/signal.h>
+
+/* misc */
+
+/* In this function, we decide, based on the 'count' of the brick,
+   where to add it in the current volume. 'count' tells us already
+   how many of the given bricks are added. other argument are self-
+   descriptive. */
+int
+add_brick_at_right_order(glusterd_brickinfo_t *brickinfo,
+                         glusterd_volinfo_t *volinfo, int count,
+                         int32_t stripe_cnt, int32_t replica_cnt)
+{
+    int idx = 0;
+    int i = 0;
+    int sub_cnt = 0;
+    glusterd_brickinfo_t *brick = NULL;
+
+    /* The complexity of the function is in deciding at which index
+       to add new brick. Even though it can be defined with a complex
+       single formula for all volume, it is separated out to make it
+       more readable */
+    if (stripe_cnt) {
+        /* common formula when 'stripe_count' is set */
+        /* idx = ((count / ((stripe_cnt * volinfo->replica_count) -
+           volinfo->dist_leaf_count)) * volinfo->dist_leaf_count) +
+           (count + volinfo->dist_leaf_count);
+        */
+
+        sub_cnt = volinfo->dist_leaf_count;
+
+        idx = ((count / ((stripe_cnt * volinfo->replica_count) - sub_cnt)) *
+               sub_cnt) +
+              (count + sub_cnt);
+
+        goto insert_brick;
+    }
+
+    /* replica count is set */
+    /* common formula when 'replica_count' is set */
+    /* idx = ((count / (replica_cnt - existing_replica_count)) *
+       existing_replica_count) +
+       (count + existing_replica_count);
+    */
+
+    sub_cnt = volinfo->replica_count;
+    idx = (count / (replica_cnt - sub_cnt) * sub_cnt) + (count + sub_cnt);
+
+insert_brick:
+    i = 0;
+    cds_list_for_each_entry(brick, &volinfo->bricks, brick_list)
+    {
+        i++;
+        if (i < idx)
+            continue;
+        gf_msg_debug(THIS->name, 0, "brick:%s index=%d, count=%d", brick->path,
+                     idx, count);
+
+        cds_list_add(&brickinfo->brick_list, &brick->brick_list);
+        break;
+    }
+
+    return 0;
+}
+
+static int
+gd_addbr_validate_replica_count(glusterd_volinfo_t *volinfo, int replica_count,
+                                int arbiter_count, int total_bricks, int *type,
+                                char *err_str, int err_len)
+{
+    int ret = -1;
+
+    /* replica count is set */
+    switch (volinfo->type) {
+        case GF_CLUSTER_TYPE_NONE:
+            if ((volinfo->brick_count * replica_count) == total_bricks) {
+                /* Change the volume type */
+                *type = GF_CLUSTER_TYPE_REPLICATE;
+                gf_msg(THIS->name, GF_LOG_INFO, 0,
+                       GD_MSG_VOL_TYPE_CHANGING_INFO,
+                       "Changing the type of volume %s from "
+                       "'distribute' to 'replica'",
+                       volinfo->volname);
+                ret = 0;
+                goto out;
+
+            } else {
+                snprintf(err_str, err_len,
+                         "Incorrect number of "
+                         "bricks (%d) supplied for replica count (%d).",
+                         (total_bricks - volinfo->brick_count), replica_count);
+                gf_msg(THIS->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
+                       "%s", err_str);
+                goto out;
+            }
+            break;
+        case GF_CLUSTER_TYPE_REPLICATE:
+            if (replica_count < volinfo->replica_count) {
+                snprintf(err_str, err_len,
+                         "Incorrect replica count (%d) supplied. "
+                         "Volume already has (%d)",
+                         replica_count, volinfo->replica_count);
+                gf_msg(THIS->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
+                       "%s", err_str);
+                goto out;
+            }
+            if (replica_count == volinfo->replica_count) {
+                if (arbiter_count && !volinfo->arbiter_count) {
+                    snprintf(err_str, err_len,
+                             "Cannot convert replica 3 volume "
+                             "to arbiter volume.");
+                    gf_msg(THIS->name, GF_LOG_ERROR, EINVAL,
+                           GD_MSG_INVALID_ENTRY, "%s", err_str);
+                    goto out;
+                }
+                if (!(total_bricks % volinfo->dist_leaf_count)) {
+                    ret = 1;
+                    goto out;
+                }
+            }
+            if (replica_count > volinfo->replica_count) {
+                /* We have to make sure before and after 'add-brick',
+                   the number or subvolumes for distribute will remain
+                   same, when replica count is given */
+                if ((total_bricks * volinfo->dist_leaf_count) ==
+                    (volinfo->brick_count *
+                     (replica_count * volinfo->stripe_count))) {
+                    /* Change the dist_leaf_count */
+                    gf_msg(THIS->name, GF_LOG_INFO, 0,
+                           GD_MSG_REPLICA_COUNT_CHANGE_INFO,
+                           "Changing the replica count of "
+                           "volume %s from %d to %d",
+                           volinfo->volname, volinfo->replica_count,
+                           replica_count);
+                    ret = 0;
+                    goto out;
+                }
+            }
+            break;
+        case GF_CLUSTER_TYPE_DISPERSE:
+            snprintf(err_str, err_len,
+                     "Volume %s cannot be converted "
+                     "from dispersed to replicated-"
+                     "dispersed",
+                     volinfo->volname);
+            gf_msg(THIS->name, GF_LOG_ERROR, EPERM, GD_MSG_OP_NOT_PERMITTED,
+                   "%s", err_str);
+            goto out;
+    }
+out:
+    return ret;
+}
+
+static int
+gd_rmbr_validate_replica_count(glusterd_volinfo_t *volinfo,
+                               int32_t replica_count, int32_t brick_count,
+                               char *err_str, size_t err_len)
+{
+    int ret = -1;
+    int replica_nodes = 0;
+    xlator_t *this = NULL;
+    this = THIS;
+    GF_ASSERT(this);
+
+    switch (volinfo->type) {
+        case GF_CLUSTER_TYPE_NONE:
+        case GF_CLUSTER_TYPE_DISPERSE:
+            snprintf(err_str, err_len,
+                     "replica count (%d) option given for non replicate "
+                     "volume %s",
+                     replica_count, volinfo->volname);
+            gf_smsg(this->name, GF_LOG_WARNING, EINVAL, GD_MSG_INVALID_ARGUMENT,
+                    err_str, NULL);
+            goto out;
+
+        case GF_CLUSTER_TYPE_REPLICATE:
+            /* in remove brick, you can only reduce the replica count */
+            if (replica_count > volinfo->replica_count) {
+                snprintf(err_str, err_len,
+                         "given replica count (%d) option is more "
+                         "than volume %s's replica count (%d)",
+                         replica_count, volinfo->volname,
+                         volinfo->replica_count);
+                gf_smsg(this->name, GF_LOG_WARNING, EINVAL,
+                        GD_MSG_INVALID_ARGUMENT, err_str, NULL);
+                goto out;
+            }
+            if (replica_count == volinfo->replica_count) {
+                /* This means the 'replica N' option on CLI was
+                   redundant. Check if the total number of bricks given
+                   for removal is same as 'dist_leaf_count' */
+                if (brick_count % volinfo->dist_leaf_count) {
+                    snprintf(err_str, err_len,
+                             "number of bricks provided (%d) is "
+                             "not valid. need at least %d "
+                             "(or %dxN)",
+                             brick_count, volinfo->dist_leaf_count,
+                             volinfo->dist_leaf_count);
+                    gf_smsg(this->name, GF_LOG_WARNING, EINVAL,
+                            GD_MSG_INVALID_ARGUMENT, err_str, NULL);
+                    goto out;
+                }
+                ret = 1;
+                goto out;
+            }
+
+            replica_nodes = ((volinfo->brick_count / volinfo->replica_count) *
+                             (volinfo->replica_count - replica_count));
+
+            if (brick_count % replica_nodes) {
+                snprintf(err_str, err_len,
+                         "need %d(xN) bricks for reducing replica "
+                         "count of the volume from %d to %d",
+                         replica_nodes, volinfo->replica_count, replica_count);
+                gf_smsg(this->name, GF_LOG_WARNING, EINVAL,
+                        GD_MSG_INVALID_ARGUMENT, err_str, NULL);
+                goto out;
+            }
+            break;
+    }
+
+    ret = 0;
+out:
+    return ret;
+}
+
+/* Handler functions */
+int
+__glusterd_handle_add_brick(rpcsvc_request_t *req)
+{
+    int32_t ret = -1;
+    gf_cli_req cli_req = {{
+        0,
+    }};
+    dict_t *dict = NULL;
+    char *bricks = NULL;
+    char *volname = NULL;
+    int brick_count = 0;
+    void *cli_rsp = NULL;
+    char err_str[2048] = "";
+    gf_cli_rsp rsp = {
+        0,
+    };
+    glusterd_volinfo_t *volinfo = NULL;
+    xlator_t *this = NULL;
+    int total_bricks = 0;
+    int32_t replica_count = 0;
+    int32_t arbiter_count = 0;
+    int32_t stripe_count = 0;
+    int type = 0;
+    glusterd_conf_t *conf = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    GF_ASSERT(req);
+
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        // failed to decode msg;
+        req->rpc_err = GARBAGE_ARGS;
+        snprintf(err_str, sizeof(err_str), "Garbage args received");
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_GARBAGE_ARGS, NULL);
+        goto out;
+    }
+
+    gf_msg(this->name, GF_LOG_INFO, 0, GD_MSG_ADD_BRICK_REQ_RECVD,
+           "Received add brick req");
+
+    if (cli_req.dict.dict_len) {
+        /* Unserialize the dictionary */
+        dict = dict_new();
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, errno,
+                   GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to "
+                   "unserialize req-buffer to dictionary");
+            snprintf(err_str, sizeof(err_str),
+                     "Unable to decode "
+                     "the command");
+            goto out;
+        }
+    }
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get volume "
+                 "name");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get volinfo "
+                 "for volume name %s",
+                 volname);
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOLINFO_GET_FAIL, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "count", SLEN("count"), &brick_count);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get volume "
+                 "brick count");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "replica-count", SLEN("replica-count"),
+                          &replica_count);
+    if (!ret) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+               "replica-count is %d", replica_count);
+    }
+
+    ret = dict_get_int32n(dict, "arbiter-count", SLEN("arbiter-count"),
+                          &arbiter_count);
+    if (!ret) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+               "arbiter-count is %d", arbiter_count);
+    }
+
+    ret = dict_get_int32n(dict, "stripe-count", SLEN("stripe-count"),
+                          &stripe_count);
+    if (!ret) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+               "stripe-count is %d", stripe_count);
+    }
+
+    if (!dict_getn(dict, "force", SLEN("force"))) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Failed to get flag");
+        goto out;
+    }
+
+    total_bricks = volinfo->brick_count + brick_count;
+
+    if (!stripe_count && !replica_count) {
+        if (volinfo->type == GF_CLUSTER_TYPE_NONE)
+            goto brick_val;
+
+        if ((volinfo->brick_count < volinfo->dist_leaf_count) &&
+            (total_bricks <= volinfo->dist_leaf_count))
+            goto brick_val;
+
+        if ((brick_count % volinfo->dist_leaf_count) != 0) {
+            snprintf(err_str, sizeof(err_str),
+                     "Incorrect number "
+                     "of bricks supplied %d with count %d",
+                     brick_count, volinfo->dist_leaf_count);
+            gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_REPLICA,
+                   "%s", err_str);
+            ret = -1;
+            goto out;
+        }
+        goto brick_val;
+        /* done with validation.. below section is if stripe|replica
+           count is given */
+    }
+
+    ret = gd_addbr_validate_replica_count(volinfo, replica_count, arbiter_count,
+                                          total_bricks, &type, err_str,
+                                          sizeof(err_str));
+    if (ret == -1) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_COUNT_VALIDATE_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    /* if replica count is same as earlier, set it back to 0 */
+    if (ret == 1)
+        replica_count = 0;
+
+    ret = dict_set_int32n(dict, "replica-count", SLEN("replica-count"),
+                          replica_count);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "failed to set the replica-count in dict");
+        goto out;
+    }
+
+brick_val:
+    ret = dict_get_strn(dict, "bricks", SLEN("bricks"), &bricks);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get volume "
+                 "bricks");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    if (type != volinfo->type) {
+        ret = dict_set_int32n(dict, "type", SLEN("type"), type);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+                   "failed to set the new type in dict");
+            goto out;
+        }
+    }
+
+    if (conf->op_version <= GD_OP_VERSION_3_7_5) {
+        gf_msg_debug(this->name, 0,
+                     "The cluster is operating at "
+                     "version less than or equal to %d. Falling back "
+                     "to syncop framework.",
+                     GD_OP_VERSION_3_7_5);
+        ret = glusterd_op_begin_synctask(req, GD_OP_ADD_BRICK, dict);
+    } else {
+        ret = glusterd_mgmt_v3_initiate_all_phases(req, GD_OP_ADD_BRICK, dict);
+    }
+
+out:
+    if (ret) {
+        rsp.op_ret = -1;
+        rsp.op_errno = 0;
+        if (err_str[0] == '\0')
+            snprintf(err_str, sizeof(err_str), "Operation failed");
+        rsp.op_errstr = err_str;
+        cli_rsp = &rsp;
+        glusterd_to_cli(req, cli_rsp, NULL, 0, NULL, (xdrproc_t)xdr_gf_cli_rsp,
+                        dict);
+        ret = 0;  // sent error to cli, prevent second reply
+    }
+
+    free(cli_req.dict.dict_val);  // its malloced by xdr
+
+    return ret;
+}
+
+int
+glusterd_handle_add_brick(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_add_brick);
+}
+
+static int
+subvol_matcher_init(int **subvols, int count)
+{
+    int ret = -1;
+
+    *subvols = GF_CALLOC(count, sizeof(int), gf_gld_mt_int);
+    if (*subvols)
+        ret = 0;
+
+    return ret;
+}
+
+static void
+subvol_matcher_update(int *subvols, glusterd_volinfo_t *volinfo,
+                      glusterd_brickinfo_t *brickinfo)
+{
+    glusterd_brickinfo_t *tmp = NULL;
+    int32_t sub_volume = 0;
+    int pos = 0;
+    if (subvols) {
+        cds_list_for_each_entry(tmp, &volinfo->bricks, brick_list)
+        {
+            if (strcmp(tmp->hostname, brickinfo->hostname) ||
+                strcmp(tmp->path, brickinfo->path)) {
+                pos++;
+                continue;
+            }
+            gf_msg_debug(THIS->name, 0, LOGSTR_FOUND_BRICK, brickinfo->hostname,
+                         brickinfo->path, volinfo->volname);
+            sub_volume = (pos / volinfo->dist_leaf_count);
+            subvols[sub_volume]++;
+            break;
+        }
+    }
+}
+
+static int
+subvol_matcher_verify(int *subvols, glusterd_volinfo_t *volinfo, char *err_str,
+                      size_t err_len, char *vol_type, int replica_count)
+{
+    int i = 0;
+    int ret = 0;
+    int count = volinfo->replica_count - replica_count;
+    xlator_t *this = THIS;
+    GF_ASSERT(this);
+
+    if (replica_count && subvols) {
+        for (i = 0; i < volinfo->subvol_count; i++) {
+            if (subvols[i] != count) {
+                ret = -1;
+                snprintf(err_str, err_len,
+                         "Remove exactly %d"
+                         " brick(s) from each subvolume.",
+                         count);
+                gf_smsg(this->name, GF_LOG_ERROR, errno,
+                        GD_MSG_BRICK_SUBVOL_VERIFY_FAIL, err_str, NULL);
+                break;
+            }
+        }
+        return ret;
+    }
+
+    do {
+        if (subvols && (subvols[i] % volinfo->dist_leaf_count == 0)) {
+            continue;
+        } else {
+            ret = -1;
+            snprintf(err_str, err_len, "Bricks not from same subvol for %s",
+                     vol_type);
+            gf_smsg(this->name, GF_LOG_ERROR, errno,
+                    GD_MSG_BRICK_SUBVOL_VERIFY_FAIL, err_str, NULL);
+            break;
+        }
+    } while (++i < volinfo->subvol_count);
+
+    return ret;
+}
+
+static void
+subvol_matcher_destroy(int *subvols)
+{
+    GF_FREE(subvols);
+}
+
+static int
+glusterd_remove_brick_validate_arbiters(glusterd_volinfo_t *volinfo,
+                                        int32_t count, int32_t replica_count,
+                                        glusterd_brickinfo_t **brickinfo_list,
+                                        char *err_str, size_t err_len)
+{
+    int i = 0;
+    int ret = 0;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    glusterd_brickinfo_t *last = NULL;
+    char *arbiter_array = NULL;
+    xlator_t *this = NULL;
+    this = THIS;
+    GF_ASSERT(this);
+
+    if (volinfo->type != GF_CLUSTER_TYPE_REPLICATE)
+        goto out;
+
+    if (!replica_count || !volinfo->arbiter_count)
+        goto out;
+
+    if (replica_count == 2) {
+        /* If it is an arbiter to replica 2 conversion, only permit
+         *  removal of the arbiter brick.*/
+        for (i = 0; i < count; i++) {
+            brickinfo = brickinfo_list[i];
+            last = get_last_brick_of_brick_group(volinfo, brickinfo);
+            if (last != brickinfo) {
+                snprintf(err_str, err_len,
+                         "Remove arbiter "
+                         "brick(s) only when converting from "
+                         "arbiter to replica 2 subvolume.");
+                gf_smsg(this->name, GF_LOG_ERROR, errno,
+                        GD_MSG_REMOVE_ARBITER_BRICK, err_str, NULL);
+                ret = -1;
+                goto out;
+            }
+        }
+    } else if (replica_count == 1) {
+        /* If it is an arbiter to plain distribute conversion, in every
+         * replica subvol, the arbiter has to be one of the bricks that
+         * are removed. */
+        arbiter_array = GF_CALLOC(volinfo->subvol_count, sizeof(*arbiter_array),
+                                  gf_common_mt_char);
+        if (!arbiter_array)
+            return -1;
+        for (i = 0; i < count; i++) {
+            brickinfo = brickinfo_list[i];
+            last = get_last_brick_of_brick_group(volinfo, brickinfo);
+            if (last == brickinfo)
+                arbiter_array[brickinfo->group] = 1;
+        }
+        for (i = 0; i < volinfo->subvol_count; i++)
+            if (!arbiter_array[i]) {
+                snprintf(err_str, err_len,
+                         "Removed bricks "
+                         "must contain arbiter when converting"
+                         " to plain distribute.");
+                gf_smsg(this->name, GF_LOG_ERROR, errno,
+                        GD_MSG_REMOVE_ARBITER_BRICK, err_str, NULL);
+                ret = -1;
+                break;
+            }
+        GF_FREE(arbiter_array);
+    }
+
+out:
+    return ret;
+}
+
+int
+__glusterd_handle_remove_brick(rpcsvc_request_t *req)
+{
+    int32_t ret = -1;
+    gf_cli_req cli_req = {{
+        0,
+    }};
+    dict_t *dict = NULL;
+    int32_t count = 0;
+    char *brick = NULL;
+    char key[64] = "";
+    int keylen;
+    int i = 1;
+    glusterd_conf_t *conf = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    glusterd_brickinfo_t **brickinfo_list = NULL;
+    int *subvols = NULL;
+    char err_str[2048] = "";
+    gf_cli_rsp rsp = {
+        0,
+    };
+    void *cli_rsp = NULL;
+    char vol_type[256] = "";
+    int32_t replica_count = 0;
+    char *volname = 0;
+    xlator_t *this = NULL;
+    int cmd = -1;
+
+    GF_ASSERT(req);
+    this = THIS;
+    GF_ASSERT(this);
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        // failed to decode msg;
+        req->rpc_err = GARBAGE_ARGS;
+        snprintf(err_str, sizeof(err_str), "Received garbage args");
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_GARBAGE_ARGS, NULL);
+        goto out;
+    }
+
+    gf_msg(this->name, GF_LOG_INFO, 0, GD_MSG_REM_BRICK_REQ_RECVD,
+           "Received rem brick req");
+
+    if (cli_req.dict.dict_len) {
+        /* Unserialize the dictionary */
+        dict = dict_new();
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, errno,
+                   GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to "
+                   "unserialize req-buffer to dictionary");
+            snprintf(err_str, sizeof(err_str),
+                     "Unable to decode "
+                     "the command");
+            goto out;
+        }
+    }
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get volume "
+                 "name");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "count", SLEN("count"), &count);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get brick "
+                 "count");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str), "Volume %s does not exist", volname);
+        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_FOUND, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "command", SLEN("command"), &cmd);
+    if (ret) {
+        snprintf(err_str, sizeof(err_str),
+                 "Unable to get cmd "
+                 "ccommand");
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "replica-count", SLEN("replica-count"),
+                          &replica_count);
+    if (!ret) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_FAILED,
+               "request to change replica-count to %d", replica_count);
+        ret = gd_rmbr_validate_replica_count(volinfo, replica_count, count,
+                                             err_str, sizeof(err_str));
+        if (ret < 0) {
+            /* logging and error msg are done in above function
+               itself */
+            goto out;
+        }
+        dict_deln(dict, "replica-count", SLEN("replica-count"));
+        if (ret) {
+            replica_count = 0;
+        } else {
+            ret = dict_set_int32n(dict, "replica-count", SLEN("replica-count"),
+                                  replica_count);
+            if (ret) {
+                gf_msg(this->name, GF_LOG_WARNING, errno,
+                       GD_MSG_DICT_SET_FAILED,
+                       "failed to set the replica_count "
+                       "in dict");
+                goto out;
+            }
+        }
+    }
+
+    /* 'vol_type' is used for giving the meaning full error msg for user */
+    if (volinfo->type == GF_CLUSTER_TYPE_REPLICATE) {
+        strcpy(vol_type, "replica");
+    } else if (volinfo->type == GF_CLUSTER_TYPE_DISPERSE) {
+        strcpy(vol_type, "disperse");
+    } else {
+        strcpy(vol_type, "distribute");
+    }
+
+    if (!replica_count && (volinfo->type == GF_CLUSTER_TYPE_REPLICATE) &&
+        (volinfo->brick_count == volinfo->dist_leaf_count)) {
+        snprintf(err_str, sizeof(err_str),
+                 "Removing bricks from replicate configuration "
+                 "is not allowed without reducing replica count "
+                 "explicitly.");
+        gf_msg(this->name, GF_LOG_ERROR, EPERM, GD_MSG_OP_NOT_PERMITTED_AC_REQD,
+               "%s", err_str);
+        ret = -1;
+        goto out;
+    }
+
+    /* Do not allow remove-brick if the bricks given is less than
+       the replica count or stripe count */
+    if (!replica_count && (volinfo->type != GF_CLUSTER_TYPE_NONE)) {
+        if (volinfo->dist_leaf_count && (count % volinfo->dist_leaf_count)) {
+            snprintf(err_str, sizeof(err_str),
+                     "Remove brick "
+                     "incorrect brick count of %d for %s %d",
+                     count, vol_type, volinfo->dist_leaf_count);
+            gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY, "%s",
+                   err_str);
+            ret = -1;
+            goto out;
+        }
+    }
+
+    if ((volinfo->type != GF_CLUSTER_TYPE_NONE) &&
+        (volinfo->subvol_count > 1)) {
+        ret = subvol_matcher_init(&subvols, volinfo->subvol_count);
+        if (ret)
+            goto out;
+    }
+
+    brickinfo_list = GF_CALLOC(count, sizeof(*brickinfo_list),
+                               gf_common_mt_pointer);
+    if (!brickinfo_list) {
+        ret = -1;
+        goto out;
+    }
+
+    while (i <= count) {
+        keylen = snprintf(key, sizeof(key), "brick%d", i);
+        ret = dict_get_strn(dict, key, keylen, &brick);
+        if (ret) {
+            snprintf(err_str, sizeof(err_str), "Unable to get %s", key);
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                   "%s", err_str);
+            goto out;
+        }
+        gf_msg_debug(this->name, 0,
+                     "Remove brick count %d brick:"
+                     " %s",
+                     i, brick);
+
+        ret = glusterd_volume_brickinfo_get_by_brick(brick, volinfo, &brickinfo,
+                                                     _gf_false);
+
+        if (ret) {
+            snprintf(err_str, sizeof(err_str),
+                     "Incorrect brick "
+                     "%s for volume %s",
+                     brick, volname);
+            gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_BRICK_NOT_FOUND,
+                   "%s", err_str);
+            goto out;
+        }
+        brickinfo_list[i - 1] = brickinfo;
+
+        i++;
+        if ((volinfo->type == GF_CLUSTER_TYPE_NONE) ||
+            (volinfo->brick_count <= volinfo->dist_leaf_count))
+            continue;
+
+        subvol_matcher_update(subvols, volinfo, brickinfo);
+    }
+
+    if ((volinfo->type != GF_CLUSTER_TYPE_NONE) &&
+        (volinfo->subvol_count > 1)) {
+        ret = subvol_matcher_verify(subvols, volinfo, err_str, sizeof(err_str),
+                                    vol_type, replica_count);
+        if (ret)
+            goto out;
+    }
+
+    ret = glusterd_remove_brick_validate_arbiters(volinfo, count, replica_count,
+                                                  brickinfo_list, err_str,
+                                                  sizeof(err_str));
+    if (ret)
+        goto out;
+
+    if (conf->op_version < GD_OP_VERSION_8_0) {
+        gf_msg_debug(this->name, 0,
+                     "The cluster is operating at "
+                     "version less than %d. remove-brick operation"
+                     "falling back to syncop framework.",
+                     GD_OP_VERSION_8_0);
+        ret = glusterd_op_begin_synctask(req, GD_OP_REMOVE_BRICK, dict);
+    } else {
+        ret = glusterd_mgmt_v3_initiate_all_phases(req, GD_OP_REMOVE_BRICK,
+                                                   dict);
+    }
+
+out:
+    if (ret) {
+        rsp.op_ret = -1;
+        rsp.op_errno = 0;
+        if (err_str[0] == '\0')
+            snprintf(err_str, sizeof(err_str), "Operation failed");
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_GLUSTERD_OP_FAILED, "%s",
+               err_str);
+        rsp.op_errstr = err_str;
+        cli_rsp = &rsp;
+        glusterd_to_cli(req, cli_rsp, NULL, 0, NULL, (xdrproc_t)xdr_gf_cli_rsp,
+                        dict);
+
+        ret = 0;  // sent error to cli, prevent second reply
+    }
+
+    if (brickinfo_list)
+        GF_FREE(brickinfo_list);
+    subvol_matcher_destroy(subvols);
+    free(cli_req.dict.dict_val);  // its malloced by xdr
+
+    return ret;
+}
+
+int
+glusterd_handle_remove_brick(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_remove_brick);
+}
+
+static int
+_glusterd_restart_gsync_session(dict_t *this, char *key, data_t *value,
+                                void *data)
+{
+    char *slave = NULL;
+    char *slave_buf = NULL;
+    char *path_list = NULL;
+    char *slave_vol = NULL;
+    char *slave_host = NULL;
+    char *slave_url = NULL;
+    char *conf_path = NULL;
+    char **errmsg = NULL;
+    int ret = -1;
+    glusterd_gsync_status_temp_t *param = NULL;
+    gf_boolean_t is_running = _gf_false;
+
+    param = (glusterd_gsync_status_temp_t *)data;
+
+    GF_ASSERT(param);
+    GF_ASSERT(param->volinfo);
+
+    slave = strchr(value->data, ':');
+    if (slave) {
+        slave++;
+        slave_buf = gf_strdup(slave);
+        if (!slave_buf) {
+            gf_msg("glusterd", GF_LOG_ERROR, ENOMEM, GD_MSG_NO_MEMORY,
+                   "Failed to gf_strdup");
+            ret = -1;
+            goto out;
+        }
+    } else
+        return 0;
+
+    ret = dict_set_dynstrn(param->rsp_dict, "slave", SLEN("slave"), slave_buf);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Unable to store slave");
+        if (slave_buf)
+            GF_FREE(slave_buf);
+        goto out;
+    }
+
+    ret = glusterd_get_slave_details_confpath(param->volinfo, param->rsp_dict,
+                                              &slave_url, &slave_host,
+                                              &slave_vol, &conf_path, errmsg);
+    if (ret) {
+        if (errmsg && *errmsg)
+            gf_msg("glusterd", GF_LOG_ERROR, 0,
+                   GD_MSG_SLAVE_CONFPATH_DETAILS_FETCH_FAIL, "%s", *errmsg);
+        else
+            gf_msg("glusterd", GF_LOG_ERROR, 0,
+                   GD_MSG_SLAVE_CONFPATH_DETAILS_FETCH_FAIL,
+                   "Unable to fetch slave or confpath details.");
+        goto out;
+    }
+
+    /* In cases that gsyncd is not running, we will not invoke it
+     * because of add-brick. */
+    ret = glusterd_check_gsync_running_local(param->volinfo->volname, slave,
+                                             conf_path, &is_running);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, 0, GD_MSG_GSYNC_VALIDATION_FAIL,
+               "gsync running validation failed.");
+        goto out;
+    }
+    if (_gf_false == is_running) {
+        gf_msg_debug("glusterd", 0,
+                     "gsync session for %s and %s is"
+                     " not running on this node. Hence not restarting.",
+                     param->volinfo->volname, slave);
+        ret = 0;
+        goto out;
+    }
+
+    ret = glusterd_get_local_brickpaths(param->volinfo, &path_list);
+    if (!path_list) {
+        gf_msg_debug("glusterd", 0,
+                     "This node not being part of"
+                     " volume should not be running gsyncd. Hence"
+                     " no gsyncd process to restart.");
+        ret = 0;
+        goto out;
+    }
+
+    ret = glusterd_check_restart_gsync_session(
+        param->volinfo, slave, param->rsp_dict, path_list, conf_path, 0);
+    if (ret)
+        gf_msg("glusterd", GF_LOG_ERROR, 0, GD_MSG_GSYNC_RESTART_FAIL,
+               "Unable to restart gsync session.");
+
+out:
+    gf_msg_debug("glusterd", 0, "Returning %d.", ret);
+    return ret;
+}
+
+/* op-sm */
+
+int
+glusterd_op_perform_add_bricks(glusterd_volinfo_t *volinfo, int32_t count,
+                               char *bricks, dict_t *dict)
+{
+    char *brick = NULL;
+    int32_t i = 1;
+    char *brick_list = NULL;
+    char *free_ptr1 = NULL;
+    char *free_ptr2 = NULL;
+    char *saveptr = NULL;
+    int32_t ret = -1;
+    int32_t stripe_count = 0;
+    int32_t replica_count = 0;
+    int32_t arbiter_count = 0;
+    int32_t type = 0;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    glusterd_gsync_status_temp_t param = {
+        0,
+    };
+    gf_boolean_t restart_needed = 0;
+    int brickid = 0;
+    char key[64] = "";
+    char *brick_mount_dir = NULL;
+    xlator_t *this = NULL;
+    glusterd_conf_t *conf = NULL;
+    gf_boolean_t is_valid_add_brick = _gf_false;
+    gf_boolean_t restart_shd = _gf_false;
+    struct statvfs brickstat = {
+        0,
+    };
+
+    this = THIS;
+    GF_ASSERT(this);
+    GF_ASSERT(volinfo);
+
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    if (bricks) {
+        brick_list = gf_strdup(bricks);
+        free_ptr1 = brick_list;
+    }
+
+    if (count)
+        brick = strtok_r(brick_list + 1, " \n", &saveptr);
+
+    if (dict) {
+        ret = dict_get_int32n(dict, "stripe-count", SLEN("stripe-count"),
+                              &stripe_count);
+        if (!ret)
+            gf_msg(THIS->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+                   "stripe-count is set %d", stripe_count);
+
+        ret = dict_get_int32n(dict, "replica-count", SLEN("replica-count"),
+                              &replica_count);
+        if (!ret)
+            gf_msg(THIS->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+                   "replica-count is set %d", replica_count);
+        ret = dict_get_int32n(dict, "arbiter-count", SLEN("arbiter-count"),
+                              &arbiter_count);
+        if (!ret)
+            gf_msg(THIS->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+                   "arbiter-count is set %d", arbiter_count);
+        ret = dict_get_int32n(dict, "type", SLEN("type"), &type);
+        if (!ret)
+            gf_msg(THIS->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_SUCCESS,
+                   "type is set %d, need to change it", type);
+    }
+
+    brickid = glusterd_get_next_available_brickid(volinfo);
+    if (brickid < 0)
+        goto out;
+    while (i <= count) {
+        ret = glusterd_brickinfo_new_from_brick(brick, &brickinfo, _gf_true,
+                                                NULL);
+        if (ret)
+            goto out;
+
+        GLUSTERD_ASSIGN_BRICKID_TO_BRICKINFO(brickinfo, volinfo, brickid++);
+
+        /* A bricks mount dir is required only by snapshots which were
+         * introduced in gluster-3.6.0
+         */
+        if (conf->op_version >= GD_OP_VERSION_3_6_0) {
+            brick_mount_dir = NULL;
+
+            snprintf(key, sizeof(key), "brick%d.mount_dir", i);
+            ret = dict_get_str(dict, key, &brick_mount_dir);
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                       "%s not present", key);
+                goto out;
+            }
+            strncpy(brickinfo->mount_dir, brick_mount_dir,
+                    SLEN(brickinfo->mount_dir));
+        }
+
+        ret = glusterd_resolve_brick(brickinfo);
+        if (ret)
+            goto out;
+
+        if (!gf_uuid_compare(brickinfo->uuid, MY_UUID)) {
+            ret = sys_statvfs(brickinfo->path, &brickstat);
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_STATVFS_FAILED,
+                       "Failed to fetch disk utilization "
+                       "from the brick (%s:%s). Please check the health of "
+                       "the brick. Error code was %s",
+                       brickinfo->hostname, brickinfo->path, strerror(errno));
+
+                goto out;
+            }
+            brickinfo->statfs_fsid = brickstat.f_fsid;
+        }
+        if (stripe_count || replica_count) {
+            add_brick_at_right_order(brickinfo, volinfo, (i - 1), stripe_count,
+                                     replica_count);
+        } else {
+            cds_list_add_tail(&brickinfo->brick_list, &volinfo->bricks);
+        }
+        brick = strtok_r(NULL, " \n", &saveptr);
+        i++;
+        volinfo->brick_count++;
+    }
+
+    /* Gets changed only if the options are given in add-brick cli */
+    if (type)
+        volinfo->type = type;
+    /* performance.client-io-threads is turned on by default,
+     * however this has adverse effects on replicate volumes due to
+     * replication design issues, till that get addressed
+     * performance.client-io-threads option is turned off for all
+     * replicate volumes if not already explicitly enabled.
+     */
+    if (type && glusterd_is_volume_replicate(volinfo) &&
+        conf->op_version >= GD_OP_VERSION_3_12_2) {
+        ret = dict_set_nstrn(volinfo->dict, "performance.client-io-threads",
+                             SLEN("performance.client-io-threads"), "off",
+                             SLEN("off"));
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
+                   "Failed to set "
+                   "performance.client-io-threads to off");
+            goto out;
+        }
+    }
+
+    if (replica_count) {
+        volinfo->replica_count = replica_count;
+    }
+    if (arbiter_count) {
+        volinfo->arbiter_count = arbiter_count;
+    }
+    if (stripe_count) {
+        volinfo->stripe_count = stripe_count;
+    }
+    volinfo->dist_leaf_count = glusterd_get_dist_leaf_count(volinfo);
+
+    /* backward compatibility */
+    volinfo->sub_count = ((volinfo->dist_leaf_count == 1)
+                              ? 0
+                              : volinfo->dist_leaf_count);
+
+    volinfo->subvol_count = (volinfo->brick_count / volinfo->dist_leaf_count);
+
+    ret = 0;
+    if (GLUSTERD_STATUS_STARTED != volinfo->status)
+        goto generate_volfiles;
+
+    ret = generate_brick_volfiles(volinfo);
+    if (ret)
+        goto out;
+
+    brick_list = gf_strdup(bricks);
+    free_ptr2 = brick_list;
+    i = 1;
+
+    if (count)
+        brick = strtok_r(brick_list + 1, " \n", &saveptr);
+
+    if (glusterd_is_volume_replicate(volinfo)) {
+        if (replica_count && conf->op_version >= GD_OP_VERSION_3_7_10) {
+            is_valid_add_brick = _gf_true;
+            if (volinfo->status == GLUSTERD_STATUS_STARTED) {
+                ret = volinfo->shd.svc.stop(&(volinfo->shd.svc), SIGTERM);
+                if (ret) {
+                    gf_msg("glusterd", GF_LOG_ERROR, 0,
+                           GD_MSG_GLUSTER_SERVICES_STOP_FAIL,
+                           "Failed to stop shd for %s.", volinfo->volname);
+                }
+                restart_shd = _gf_true;
+            }
+            ret = generate_dummy_client_volfiles(volinfo);
+            if (ret) {
+                gf_msg(THIS->name, GF_LOG_ERROR, 0, GD_MSG_VOLFILE_CREATE_FAIL,
+                       "Failed to create volfile.");
+                goto out;
+            }
+        }
+    }
+
+    while (i <= count) {
+        ret = glusterd_volume_brickinfo_get_by_brick(brick, volinfo, &brickinfo,
+                                                     _gf_true);
+        if (ret)
+            goto out;
+
+        if (gf_uuid_is_null(brickinfo->uuid)) {
+            ret = glusterd_resolve_brick(brickinfo);
+            if (ret) {
+                gf_msg("glusterd", GF_LOG_ERROR, 0, GD_MSG_RESOLVE_BRICK_FAIL,
+                       FMTSTR_RESOLVE_BRICK, brickinfo->hostname,
+                       brickinfo->path);
+                goto out;
+            }
+        }
+
+        /* if the volume is a replicate volume, do: */
+        if (is_valid_add_brick) {
+            if (!gf_uuid_compare(brickinfo->uuid, MY_UUID)) {
+                ret = glusterd_handle_replicate_brick_ops(volinfo, brickinfo,
+                                                          GD_OP_ADD_BRICK);
+                if (ret < 0)
+                    goto out;
+            }
+        }
+        ret = glusterd_brick_start(volinfo, brickinfo, _gf_true, _gf_false);
+        if (ret)
+            goto out;
+        i++;
+        brick = strtok_r(NULL, " \n", &saveptr);
+
+        /* Check if the brick is added in this node, and set
+         * the restart_needed flag. */
+        if ((!gf_uuid_compare(brickinfo->uuid, MY_UUID)) && !restart_needed) {
+            restart_needed = 1;
+            gf_msg_debug("glusterd", 0,
+                         "Restart gsyncd session, if it's already "
+                         "running.");
+        }
+    }
+
+    /* If the restart_needed flag is set, restart gsyncd sessions for that
+     * particular master with all the slaves. */
+    if (restart_needed) {
+        param.rsp_dict = dict;
+        param.volinfo = volinfo;
+        dict_foreach(volinfo->gsync_slaves, _glusterd_restart_gsync_session,
+                     &param);
+    }
+
+generate_volfiles:
+    if (conf->op_version <= GD_OP_VERSION_3_7_5) {
+        ret = glusterd_create_volfiles_and_notify_services(volinfo);
+    } else {
+        /*
+         * The cluster is operating at version greater than
+         * gluster-3.7.5. So no need to sent volfile fetch
+         * request in commit phase, the same will be done
+         * in post validate phase with v3 framework.
+         */
+    }
+
+out:
+    GF_FREE(free_ptr1);
+    GF_FREE(free_ptr2);
+    if (restart_shd) {
+        if (volinfo->shd.svc.manager(&(volinfo->shd.svc), volinfo,
+                                     PROC_START_NO_WAIT)) {
+            gf_msg("glusterd", GF_LOG_CRITICAL, 0,
+                   GD_MSG_GLUSTER_SERVICE_START_FAIL,
+                   "Failed to start shd for %s.", volinfo->volname);
+        }
+    }
+
+    gf_msg_debug("glusterd", 0, "Returning %d", ret);
+    return ret;
+}
+
+int
+glusterd_op_perform_remove_brick(glusterd_volinfo_t *volinfo, char *brick,
+                                 int force, int *need_migrate)
+{
+    glusterd_brickinfo_t *brickinfo = NULL;
+    int32_t ret = -1;
+    glusterd_conf_t *priv = NULL;
+
+    GF_ASSERT(volinfo);
+    GF_ASSERT(brick);
+
+    priv = THIS->private;
+    GF_ASSERT(priv);
+
+    ret = glusterd_volume_brickinfo_get_by_brick(brick, volinfo, &brickinfo,
+                                                 _gf_false);
+    if (ret)
+        goto out;
+
+    ret = glusterd_resolve_brick(brickinfo);
+    if (ret)
+        goto out;
+
+    glusterd_volinfo_reset_defrag_stats(volinfo);
+
+    if (!gf_uuid_compare(brickinfo->uuid, MY_UUID)) {
+        /* Only if the brick is in this glusterd, do the rebalance */
+        if (need_migrate)
+            *need_migrate = 1;
+    }
+
+    if (force) {
+        ret = glusterd_brick_stop(volinfo, brickinfo, _gf_true);
+        if (ret) {
+            gf_msg(THIS->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_STOP_FAIL,
+                   "Unable to stop "
+                   "glusterfs, ret: %d",
+                   ret);
+        }
+        goto out;
+    }
+
+    brickinfo->decommissioned = 1;
+    ret = 0;
+out:
+    gf_msg_debug("glusterd", 0, "Returning %d", ret);
+    return ret;
+}
+
+int
+glusterd_op_stage_add_brick(dict_t *dict, char **op_errstr, dict_t *rsp_dict)
+{
+    int ret = 0;
+    char *volname = NULL;
+    int count = 0;
+    int replica_count = 0;
+    int arbiter_count = 0;
+    int i = 0;
+    int32_t local_brick_count = 0;
+    char *bricks = NULL;
+    char *brick_list = NULL;
+    char *saveptr = NULL;
+    char *free_ptr = NULL;
+    char *brick = NULL;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    xlator_t *this = NULL;
+    char msg[4096] = "";
+    char key[64] = "";
+    gf_boolean_t brick_alloc = _gf_false;
+    char *all_bricks = NULL;
+    char *str_ret = NULL;
+    gf_boolean_t is_force = _gf_false;
+    glusterd_conf_t *conf = NULL;
+    int32_t len = 0;
+
+    this = THIS;
+    GF_ASSERT(this);
+    conf = this->private;
+    GF_ASSERT(conf);
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOL_NOT_FOUND,
+               "Unable to find volume: %s", volname);
+        goto out;
+    }
+
+    ret = glusterd_validate_volume_id(dict, volinfo);
+    if (ret)
+        goto out;
+
+    ret = dict_get_int32n(dict, "replica-count", SLEN("replica-count"),
+                          &replica_count);
+    if (ret) {
+        gf_msg_debug(this->name, 0, "Unable to get replica count");
+    }
+
+    if (replica_count > 0) {
+        ret = op_version_check(this, GD_OP_VER_PERSISTENT_AFR_XATTRS, msg,
+                               sizeof(msg));
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_OP_VERSION_MISMATCH,
+                   "%s", msg);
+            *op_errstr = gf_strdup(msg);
+            goto out;
+        }
+    }
+
+    glusterd_add_peers_to_auth_list(volname);
+
+    if (replica_count && glusterd_is_volume_replicate(volinfo)) {
+        /* Do not allow add-brick for stopped volumes when replica-count
+         * is being increased.
+         */
+        if (GLUSTERD_STATUS_STOPPED == volinfo->status &&
+            conf->op_version >= GD_OP_VERSION_3_7_10) {
+            ret = -1;
+            snprintf(msg, sizeof(msg),
+                     " Volume must not be in"
+                     " stopped state when replica-count needs to "
+                     " be increased.");
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL, "%s",
+                   msg);
+            *op_errstr = gf_strdup(msg);
+            goto out;
+        }
+        /* op-version check for replica 2 to arbiter conversion. If we
+         * don't have this check, an older peer added as arbiter brick
+         * will not have the  arbiter xlator in its volfile. */
+        if ((replica_count == 3) && (conf->op_version < GD_OP_VERSION_3_8_0)) {
+            ret = dict_get_int32n(dict, "arbiter-count", SLEN("arbiter-count"),
+                                  &arbiter_count);
+            if (ret) {
+                gf_msg_debug(this->name, 0,
+                             "No arbiter count present in the dict");
+            } else if (arbiter_count == 1) {
+                ret = -1;
+                snprintf(msg, sizeof(msg),
+                         "Cluster op-version must "
+                         "be >= 30800 to add arbiter brick to a "
+                         "replica 2 volume.");
+                gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL, "%s",
+                       msg);
+                *op_errstr = gf_strdup(msg);
+                goto out;
+            }
+        }
+        /* Do not allow increasing replica count for arbiter volumes. */
+        if (volinfo->arbiter_count) {
+            ret = -1;
+            snprintf(msg, sizeof(msg),
+                     "Increasing replica count "
+                     "for arbiter volumes is not supported.");
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL, "%s",
+                   msg);
+            *op_errstr = gf_strdup(msg);
+            goto out;
+        }
+    }
+
+    is_force = dict_get_str_boolean(dict, "force", _gf_false);
+
+    /* Check brick order if the volume type is replicate or disperse. If
+     * force at the end of command not given then check brick order.
+     * doing this check at the originator node is sufficient.
+     */
+
+    if (!is_force && is_origin_glusterd(dict)) {
+        ret = 0;
+        if (volinfo->type == GF_CLUSTER_TYPE_REPLICATE) {
+            gf_msg_debug(this->name, 0,
+                         "Replicate cluster type "
+                         "found. Checking brick order.");
+            if (replica_count)
+                ret = glusterd_check_brick_order(dict, msg, volinfo->type,
+                                                 &volname, &bricks, &count,
+                                                 replica_count);
+            else
+                ret = glusterd_check_brick_order(dict, msg, volinfo->type,
+                                                 &volname, &bricks, &count,
+                                                 volinfo->replica_count);
+        } else if (volinfo->type == GF_CLUSTER_TYPE_DISPERSE) {
+            gf_msg_debug(this->name, 0,
+                         "Disperse cluster type"
+                         " found. Checking brick order.");
+            ret = glusterd_check_brick_order(dict, msg, volinfo->type, &volname,
+                                             &bricks, &count,
+                                             volinfo->disperse_count);
+        }
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BAD_BRKORDER,
+                   "Not adding brick because of "
+                   "bad brick order. %s",
+                   msg);
+            *op_errstr = gf_strdup(msg);
+            goto out;
+        }
+    }
+
+    if (volinfo->replica_count < replica_count && !is_force) {
+        cds_list_for_each_entry(brickinfo, &volinfo->bricks, brick_list)
+        {
+            if (gf_uuid_compare(brickinfo->uuid, MY_UUID))
+                continue;
+            if (brickinfo->status == GF_BRICK_STOPPED) {
+                ret = -1;
+                len = snprintf(msg, sizeof(msg),
+                               "Brick %s "
+                               "is down, changing replica "
+                               "count needs all the bricks "
+                               "to be up to avoid data loss",
+                               brickinfo->path);
+                if (len < 0) {
+                    strcpy(msg, "<error>");
+                }
+                gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL, "%s",
+                       msg);
+                *op_errstr = gf_strdup(msg);
+                goto out;
+            }
+        }
+    }
+
+    if (conf->op_version > GD_OP_VERSION_3_7_5 && is_origin_glusterd(dict)) {
+        ret = glusterd_validate_quorum(this, GD_OP_ADD_BRICK, dict, op_errstr);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_CRITICAL, 0, GD_MSG_SERVER_QUORUM_NOT_MET,
+                   "Server quorum not met. Rejecting operation.");
+            goto out;
+        }
+    } else {
+        /* Case 1: conf->op_version <= GD_OP_VERSION_3_7_5
+         *         in this case the add-brick is running
+         *         syncop framework that will do a quorum
+         *         check by default
+         * Case 2: We don't need to do quorum check on every
+         *         node, only originator glusterd need to
+         *         check for quorum
+         * So nothing need to be done in else
+         */
+    }
+
+    if (glusterd_is_defrag_on(volinfo)) {
+        snprintf(msg, sizeof(msg),
+                 "Volume name %s rebalance is in "
+                 "progress. Please retry after completion",
+                 volname);
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_OIP_RETRY_LATER, "%s", msg);
+        *op_errstr = gf_strdup(msg);
+        ret = -1;
+        goto out;
+    }
+
+    if (volinfo->snap_count > 0 || !cds_list_empty(&volinfo->snap_volumes)) {
+        snprintf(msg, sizeof(msg),
+                 "Volume %s  has %" PRIu64
+                 " snapshots. "
+                 "Changing the volume configuration will not effect snapshots."
+                 "But the snapshot brick mount should be intact to "
+                 "make them function.",
+                 volname, volinfo->snap_count);
+        gf_msg("glusterd", GF_LOG_WARNING, 0, GD_MSG_SNAP_WARN, "%s", msg);
+        msg[0] = '\0';
+    }
+
+    if (!count) {
+        ret = dict_get_int32n(dict, "count", SLEN("count"), &count);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                   "Unable to get count");
+            goto out;
+        }
+    }
+
+    if (!bricks) {
+        ret = dict_get_strn(dict, "bricks", SLEN("bricks"), &bricks);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                   "Unable to get bricks");
+            goto out;
+        }
+    }
+
+    if (bricks) {
+        brick_list = gf_strdup(bricks);
+        all_bricks = gf_strdup(bricks);
+        free_ptr = brick_list;
+    }
+
+    if (count)
+        brick = strtok_r(brick_list + 1, " \n", &saveptr);
+
+    while (i < count) {
+        if (!glusterd_store_is_valid_brickpath(volname, brick) ||
+            !glusterd_is_valid_volfpath(volname, brick)) {
+            snprintf(msg, sizeof(msg),
+                     "brick path %s is "
+                     "too long",
+                     brick);
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRKPATH_TOO_LONG, "%s",
+                   msg);
+            *op_errstr = gf_strdup(msg);
+
+            ret = -1;
+            goto out;
+        }
+
+        ret = glusterd_brickinfo_new_from_brick(brick, &brickinfo, _gf_true,
+                                                NULL);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_NOT_FOUND,
+                   "Add-brick: Unable"
+                   " to get brickinfo");
+            goto out;
+        }
+        brick_alloc = _gf_true;
+
+        ret = glusterd_new_brick_validate(brick, brickinfo, msg, sizeof(msg),
+                                          NULL);
+        if (ret) {
+            *op_errstr = gf_strdup(msg);
+            ret = -1;
+            goto out;
+        }
+
+        if (!gf_uuid_compare(brickinfo->uuid, MY_UUID)) {
+            ret = glusterd_validate_and_create_brickpath(
+                brickinfo, volinfo->volume_id, volinfo->volname, op_errstr,
+                is_force, _gf_false);
+            if (ret)
+                goto out;
+
+            /* A bricks mount dir is required only by snapshots which were
+             * introduced in gluster-3.6.0
+             */
+            if (conf->op_version >= GD_OP_VERSION_3_6_0) {
+                ret = glusterd_get_brick_mount_dir(
+                    brickinfo->path, brickinfo->hostname, brickinfo->mount_dir);
+                if (ret) {
+                    gf_msg(this->name, GF_LOG_ERROR, 0,
+                           GD_MSG_BRICK_MOUNTDIR_GET_FAIL,
+                           "Failed to get brick mount_dir");
+                    goto out;
+                }
+
+                snprintf(key, sizeof(key), "brick%d.mount_dir", i + 1);
+                ret = dict_set_dynstr_with_alloc(rsp_dict, key,
+                                                 brickinfo->mount_dir);
+                if (ret) {
+                    gf_msg(this->name, GF_LOG_ERROR, errno,
+                           GD_MSG_DICT_SET_FAILED, "Failed to set %s", key);
+                    goto out;
+                }
+            }
+
+            local_brick_count = i + 1;
+        }
+
+        glusterd_brickinfo_delete(brickinfo);
+        brick_alloc = _gf_false;
+        brickinfo = NULL;
+        brick = strtok_r(NULL, " \n", &saveptr);
+        i++;
+    }
+
+    ret = dict_set_int32n(rsp_dict, "brick_count", SLEN("brick_count"),
+                          local_brick_count);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set local_brick_count");
+        goto out;
+    }
+
+out:
+    GF_FREE(free_ptr);
+    if (brick_alloc && brickinfo)
+        glusterd_brickinfo_delete(brickinfo);
+    GF_FREE(str_ret);
+    GF_FREE(all_bricks);
+
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+
+    return ret;
+}
+
+int
+glusterd_remove_brick_validate_bricks(gf1_op_commands cmd, int32_t brick_count,
+                                      dict_t *dict, glusterd_volinfo_t *volinfo,
+                                      char **errstr,
+                                      gf_cli_defrag_type cmd_defrag)
+{
+    char *brick = NULL;
+    char msg[2048] = "";
+    char key[64] = "";
+    int keylen;
+    glusterd_brickinfo_t *brickinfo = NULL;
+    glusterd_peerinfo_t *peerinfo = NULL;
+    int i = 0;
+    int ret = -1;
+    char pidfile[PATH_MAX + 1] = {
+        0,
+    };
+    glusterd_conf_t *priv = THIS->private;
+    int pid = -1;
+    xlator_t *this = THIS;
+    GF_ASSERT(this);
+
+    /* Check whether all the nodes of the bricks to be removed are
+     * up, if not fail the operation */
+    for (i = 1; i <= brick_count; i++) {
+        keylen = snprintf(key, sizeof(key), "brick%d", i);
+        ret = dict_get_strn(dict, key, keylen, &brick);
+        if (ret) {
+            snprintf(msg, sizeof(msg), "Unable to get %s", key);
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                    "key=%s", key, NULL);
+            *errstr = gf_strdup(msg);
+            goto out;
+        }
+
+        ret = glusterd_volume_brickinfo_get_by_brick(brick, volinfo, &brickinfo,
+                                                     _gf_false);
+        if (ret) {
+            snprintf(msg, sizeof(msg),
+                     "Incorrect brick "
+                     "%s for volume %s",
+                     brick, volinfo->volname);
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_INCORRECT_BRICK,
+                    "Brick=%s, Volume=%s", brick, volinfo->volname, NULL);
+            *errstr = gf_strdup(msg);
+            goto out;
+        }
+        /* Do not allow commit if the bricks are not decommissioned
+         * if its a remove brick commit
+         */
+        if (!brickinfo->decommissioned && cmd == GF_OP_CMD_COMMIT) {
+            snprintf(msg, sizeof(msg),
+                     "Brick %s "
+                     "is not decommissioned. "
+                     "Use start or force option",
+                     brick);
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_BRICK_NOT_DECOM,
+                    "Use 'start' or 'force' option, Brick=%s", brick, NULL);
+            *errstr = gf_strdup(msg);
+            ret = -1;
+            goto out;
+        }
+
+        if (glusterd_is_local_brick(THIS, volinfo, brickinfo)) {
+            switch (cmd) {
+                case GF_OP_CMD_START:
+                    goto check;
+                case GF_OP_CMD_NONE:
+                default:
+                    break;
+            }
+
+            switch (cmd_defrag) {
+                case GF_DEFRAG_CMD_NONE:
+                default:
+                    continue;
+            }
+        check:
+            if (brickinfo->status != GF_BRICK_STARTED) {
+                snprintf(msg, sizeof(msg),
+                         "Found stopped "
+                         "brick %s. Use force option to "
+                         "remove the offline brick",
+                         brick);
+                gf_smsg(
+                    this->name, GF_LOG_ERROR, errno, GD_MSG_BRICK_STOPPED,
+                    "Use 'force' option to remove the offline brick, Brick=%s",
+                    brick, NULL);
+                *errstr = gf_strdup(msg);
+                ret = -1;
+                goto out;
+            }
+            GLUSTERD_GET_BRICK_PIDFILE(pidfile, volinfo, brickinfo, priv);
+            if (!gf_is_service_running(pidfile, &pid)) {
+                snprintf(msg, sizeof(msg),
+                         "Found dead "
+                         "brick %s",
+                         brick);
+                gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_BRICK_DEAD,
+                        "Brick=%s", brick, NULL);
+                *errstr = gf_strdup(msg);
+                ret = -1;
+                goto out;
+            } else {
+                ret = 0;
+            }
+            continue;
+        }
+
+        RCU_READ_LOCK;
+        peerinfo = glusterd_peerinfo_find_by_uuid(brickinfo->uuid);
+        if (!peerinfo) {
+            RCU_READ_UNLOCK;
+            snprintf(msg, sizeof(msg),
+                     "Host node of the "
+                     "brick %s is not in cluster",
+                     brick);
+            gf_smsg(this->name, GF_LOG_ERROR, errno,
+                    GD_MSG_BRICK_HOST_NOT_FOUND, "Brick=%s", brick, NULL);
+            *errstr = gf_strdup(msg);
+            ret = -1;
+            goto out;
+        }
+        if (!peerinfo->connected) {
+            RCU_READ_UNLOCK;
+            snprintf(msg, sizeof(msg),
+                     "Host node of the "
+                     "brick %s is down",
+                     brick);
+            gf_smsg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_HOST_DOWN,
+                    "Brick=%s", brick, NULL);
+            *errstr = gf_strdup(msg);
+            ret = -1;
+            goto out;
+        }
+        RCU_READ_UNLOCK;
+    }
+
+out:
+    return ret;
+}
+
+int
+glusterd_op_stage_remove_brick(dict_t *dict, char **op_errstr)
+{
+    int ret = -1;
+    char *volname = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    char *errstr = NULL;
+    int32_t brick_count = 0;
+    char msg[2048] = "";
+    int32_t flag = 0;
+    gf1_op_commands cmd = GF_OP_CMD_NONE;
+    char *task_id_str = NULL;
+    xlator_t *this = NULL;
+    gsync_status_param_t param = {
+        0,
+    };
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = op_version_check(this, GD_OP_VER_PERSISTENT_AFR_XATTRS, msg,
+                           sizeof(msg));
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_OP_VERSION_MISMATCH, "%s",
+               msg);
+        *op_errstr = gf_strdup(msg);
+        goto out;
+    }
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOL_NOT_FOUND,
+               "Volume %s does not exist", volname);
+        goto out;
+    }
+
+    ret = glusterd_validate_volume_id(dict, volinfo);
+    if (ret)
+        goto out;
+
+    ret = dict_get_int32n(dict, "command", SLEN("command"), &flag);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get brick command");
+        goto out;
+    }
+    cmd = flag;
+
+    ret = dict_get_int32n(dict, "count", SLEN("count"), &brick_count);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get brick count");
+        goto out;
+    }
+
+    ret = 0;
+    if (volinfo->brick_count == brick_count) {
+        errstr = gf_strdup(
+            "Deleting all the bricks of the "
+            "volume is not allowed");
+        gf_smsg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_DELETE, NULL);
+        ret = -1;
+        goto out;
+    }
+
+    ret = -1;
+    switch (cmd) {
+        case GF_OP_CMD_NONE:
+            errstr = gf_strdup("no remove-brick command issued");
+            gf_smsg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_NO_REMOVE_CMD,
+                    NULL);
+            goto out;
+
+        case GF_OP_CMD_STATUS:
+            ret = 0;
+            goto out;
+        case GF_OP_CMD_START: {
+            if ((volinfo->type == GF_CLUSTER_TYPE_REPLICATE) &&
+                dict_getn(dict, "replica-count", SLEN("replica-count"))) {
+                snprintf(msg, sizeof(msg),
+                         "Migration of data is not "
+                         "needed when reducing replica count. Use the"
+                         " 'force' option");
+                errstr = gf_strdup(msg);
+                gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_USE_THE_FORCE, "%s",
+                       errstr);
+                goto out;
+            }
+
+            if (GLUSTERD_STATUS_STARTED != volinfo->status) {
+                snprintf(msg, sizeof(msg),
+                         "Volume %s needs "
+                         "to be started before remove-brick "
+                         "(you can use 'force' or 'commit' "
+                         "to override this behavior)",
+                         volinfo->volname);
+                errstr = gf_strdup(msg);
+                gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOL_NOT_STARTED,
+                       "%s", errstr);
+                goto out;
+            }
+            if (!gd_is_remove_brick_committed(volinfo)) {
+                snprintf(msg, sizeof(msg),
+                         "An earlier remove-brick "
+                         "task exists for volume %s. Either commit it"
+                         " or stop it before starting a new task.",
+                         volinfo->volname);
+                errstr = gf_strdup(msg);
+                gf_msg(this->name, GF_LOG_ERROR, 0,
+                       GD_MSG_OLD_REMOVE_BRICK_EXISTS,
+                       "Earlier remove-brick"
+                       " task exists for volume %s.",
+                       volinfo->volname);
+                goto out;
+            }
+            if (glusterd_is_defrag_on(volinfo)) {
+                errstr = gf_strdup(
+                    "Rebalance is in progress. Please "
+                    "retry after completion");
+                gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_OIP_RETRY_LATER,
+                       "%s", errstr);
+                goto out;
+            }
+
+            /* Check if the connected clients are all of version
+             * glusterfs-3.6 and higher. This is needed to prevent some data
+             * loss issues that could occur when older clients are connected
+             * when rebalance is run.
+             */
+            ret = glusterd_check_client_op_version_support(
+                volname, GD_OP_VERSION_3_6_0, NULL);
+            if (ret) {
+                ret = gf_asprintf(op_errstr,
+                                  "Volume %s has one or "
+                                  "more connected clients of a version"
+                                  " lower than GlusterFS-v3.6.0. "
+                                  "Starting remove-brick in this state "
+                                  "could lead to data loss.\nPlease "
+                                  "disconnect those clients before "
+                                  "attempting this command again.",
+                                  volname);
+                goto out;
+            }
+
+            if (volinfo->snap_count > 0 ||
+                !cds_list_empty(&volinfo->snap_volumes)) {
+                snprintf(msg, sizeof(msg),
+                         "Volume %s  has %" PRIu64
+                         " snapshots. "
+                         "Changing the volume configuration will not effect "
+                         "snapshots."
+                         "But the snapshot brick mount should be intact to "
+                         "make them function.",
+                         volname, volinfo->snap_count);
+                gf_msg("glusterd", GF_LOG_WARNING, 0, GD_MSG_SNAP_WARN, "%s",
+                       msg);
+                msg[0] = '\0';
+            }
+
+            ret = glusterd_remove_brick_validate_bricks(
+                cmd, brick_count, dict, volinfo, &errstr, GF_DEFRAG_CMD_NONE);
+            if (ret)
+                goto out;
+
+            if (is_origin_glusterd(dict)) {
+                ret = glusterd_generate_and_set_task_id(
+                    dict, GF_REMOVE_BRICK_TID_KEY,
+                    SLEN(GF_REMOVE_BRICK_TID_KEY));
+                if (ret) {
+                    gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_TASKID_GEN_FAIL,
+                           "Failed to generate task-id");
+                    goto out;
+                }
+            } else {
+                ret = dict_get_strn(dict, GF_REMOVE_BRICK_TID_KEY,
+                                    SLEN(GF_REMOVE_BRICK_TID_KEY),
+                                    &task_id_str);
+                if (ret) {
+                    gf_msg(this->name, GF_LOG_WARNING, errno,
+                           GD_MSG_DICT_GET_FAILED, "Missing remove-brick-id");
+                    ret = 0;
+                }
+            }
+            break;
+        }
+
+        case GF_OP_CMD_STOP:
+            ret = 0;
+            break;
+
+        case GF_OP_CMD_COMMIT:
+            if (volinfo->decommission_in_progress) {
+                errstr = gf_strdup(
+                    "use 'force' option as migration "
+                    "is in progress");
+                gf_smsg(this->name, GF_LOG_WARNING, 0, GD_MSG_MIGRATION_PROG,
+                        "Use 'force' option", NULL);
+                goto out;
+            }
+
+            if (volinfo->rebal.defrag_status == GF_DEFRAG_STATUS_FAILED) {
+                errstr = gf_strdup(
+                    "use 'force' option as migration "
+                    "has failed");
+                gf_smsg(this->name, GF_LOG_WARNING, 0, GD_MSG_MIGRATION_FAIL,
+                        "Use 'force' option", NULL);
+                goto out;
+            }
+
+            if (volinfo->rebal.defrag_status == GF_DEFRAG_STATUS_COMPLETE) {
+                if (volinfo->rebal.rebalance_failures > 0 ||
+                    volinfo->rebal.skipped_files > 0) {
+                    errstr = gf_strdup(
+                        "use 'force' option as migration "
+                        "of some files might have been skipped or "
+                        "has failed");
+                    gf_smsg(this->name, GF_LOG_WARNING, 0,
+                            GD_MSG_MIGRATION_FAIL,
+                            "Use 'force' option, some files might have been "
+                            "skipped",
+                            NULL);
+                    goto out;
+                }
+            }
+
+            ret = glusterd_remove_brick_validate_bricks(
+                cmd, brick_count, dict, volinfo, &errstr, GF_DEFRAG_CMD_NONE);
+            if (ret)
+                goto out;
+
+            /* If geo-rep is configured, for this volume, it should be
+             * stopped.
+             */
+            param.volinfo = volinfo;
+            ret = glusterd_check_geo_rep_running(&param, op_errstr);
+            if (ret || param.is_active) {
+                ret = -1;
+                goto out;
+            }
+
+            break;
+
+        case GF_OP_CMD_COMMIT_FORCE:
+        case GF_OP_CMD_DETACH_START:
+        case GF_OP_CMD_DETACH_COMMIT:
+        case GF_OP_CMD_DETACH_COMMIT_FORCE:
+        case GF_OP_CMD_STOP_DETACH_TIER:
+            break;
+    }
+    ret = 0;
+
+out:
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+    if (ret && errstr) {
+        if (op_errstr)
+            *op_errstr = errstr;
+    }
+    if (!op_errstr && errstr)
+        GF_FREE(errstr);
+    return ret;
+}
+
+int
+glusterd_remove_brick_migrate_cbk(glusterd_volinfo_t *volinfo,
+                                  gf_defrag_status_t status)
+{
+    int ret = 0;
+
+#if 0 /* TODO: enable this behavior once cluster-wide awareness comes for      \
+         defrag cbk function */
+        glusterd_brickinfo_t *brickinfo = NULL;
+        glusterd_brickinfo_t *tmp = NULL;
+
+        switch (status) {
+        case GF_DEFRAG_STATUS_PAUSED:
+        case GF_DEFRAG_STATUS_FAILED:
+                /* No changes required in the volume file.
+                   everything should remain as is */
+                break;
+        case GF_DEFRAG_STATUS_STOPPED:
+                /* Fall back to the old volume file */
+                cds_list_for_each_entry_safe (brickinfo, tmp, &volinfo->bricks,
+                                              brick_list) {
+                        if (!brickinfo->decommissioned)
+                                continue;
+                        brickinfo->decommissioned = 0;
+                }
+                break;
+
+        case GF_DEFRAG_STATUS_COMPLETE:
+                /* Done with the task, you can remove the brick from the
+                   volume file */
+                cds_list_for_each_entry_safe (brickinfo, tmp, &volinfo->bricks,
+                                              brick_list) {
+                        if (!brickinfo->decommissioned)
+                                continue;
+                        gf_log (THIS->name, GF_LOG_INFO, "removing the brick %s",
+                                brickinfo->path);
+                        brickinfo->decommissioned = 0;
+                        if (GLUSTERD_STATUS_STARTED == volinfo->status) {
+                            /*TODO: use the 'atomic' flavour of brick_stop*/
+                                ret = glusterd_brick_stop (volinfo, brickinfo);
+                                if (ret) {
+                                        gf_log (THIS->name, GF_LOG_ERROR,
+                                                "Unable to stop glusterfs (%d)", ret);
+                                }
+                        }
+                        glusterd_delete_brick (volinfo, brickinfo);
+                }
+                break;
+
+        default:
+                GF_ASSERT (!"cbk function called with wrong status");
+                break;
+        }
+
+        ret = glusterd_create_volfiles_and_notify_services (volinfo);
+        if (ret)
+                gf_log (THIS->name, GF_LOG_ERROR,
+                        "Unable to write volume files (%d)", ret);
+
+        ret = glusterd_store_volinfo (volinfo, GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+        if (ret)
+                gf_log (THIS->name, GF_LOG_ERROR,
+                        "Unable to store volume info (%d)", ret);
+
+
+        if (GLUSTERD_STATUS_STARTED == volinfo->status) {
+                ret = glusterd_check_generate_start_nfs ();
+                if (ret)
+                        gf_log (THIS->name, GF_LOG_ERROR,
+                                "Unable to start nfs process (%d)", ret);
+        }
+
+#endif
+
+    volinfo->decommission_in_progress = 0;
+    return ret;
+}
+
+int
+glusterd_op_add_brick(dict_t *dict, char **op_errstr)
+{
+    int ret = 0;
+    char *volname = NULL;
+    glusterd_conf_t *priv = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    xlator_t *this = NULL;
+    char *bricks = NULL;
+    int32_t count = 0;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_FOUND,
+               "Unable to allocate memory");
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "count", SLEN("count"), &count);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get count");
+        goto out;
+    }
+
+    ret = dict_get_strn(dict, "bricks", SLEN("bricks"), &bricks);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get bricks");
+        goto out;
+    }
+
+    ret = glusterd_op_perform_add_bricks(volinfo, count, bricks, dict);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL,
+               "Unable to add bricks");
+        goto out;
+    }
+    if (priv->op_version <= GD_OP_VERSION_3_7_5) {
+        ret = glusterd_store_volinfo(volinfo,
+                                     GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+        if (ret)
+            goto out;
+    } else {
+        /*
+         * The cluster is operating at version greater than
+         * gluster-3.7.5. So no need to store volfiles
+         * in commit phase, the same will be done
+         * in post validate phase with v3 framework.
+         */
+    }
+
+    if (GLUSTERD_STATUS_STARTED == volinfo->status)
+        ret = glusterd_svcs_manager(volinfo);
+
+out:
+    return ret;
+}
+
+int
+glusterd_post_commit_add_brick(dict_t *dict, char **op_errstr)
+{
+    int ret = 0;
+    char *volname = NULL;
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+
+    if (ret) {
+        gf_msg(THIS->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+    ret = glusterd_replace_old_auth_allow_list(volname);
+out:
+    return ret;
+}
+
+int
+glusterd_post_commit_replace_brick(dict_t *dict, char **op_errstr)
+{
+    int ret = 0;
+    char *volname = NULL;
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+
+    if (ret) {
+        gf_msg(THIS->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+    ret = glusterd_replace_old_auth_allow_list(volname);
+out:
+    return ret;
+}
+
+int
+glusterd_set_rebalance_id_for_remove_brick(dict_t *req_dict, dict_t *rsp_dict)
+{
+    int ret = -1;
+    char *volname = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    char msg[2048] = {0};
+    char *task_id_str = NULL;
+    xlator_t *this = NULL;
+    int32_t cmd = 0;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    GF_ASSERT(rsp_dict);
+    GF_ASSERT(req_dict);
+
+    ret = dict_get_strn(rsp_dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        gf_msg_debug(this->name, 0, "volname not found");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_FOUND,
+               "Unable to allocate memory");
+        goto out;
+    }
+
+    ret = dict_get_int32n(rsp_dict, "command", SLEN("command"), &cmd);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get command");
+        goto out;
+    }
+
+    /* remove brick task id is generted in glusterd_op_stage_remove_brick(),
+     * but rsp_dict is unavailable there. So copying it to rsp_dict from
+     * req_dict here. */
+
+    if (is_origin_glusterd(rsp_dict)) {
+        ret = dict_get_strn(req_dict, GF_REMOVE_BRICK_TID_KEY,
+                            SLEN(GF_REMOVE_BRICK_TID_KEY), &task_id_str);
+        if (ret) {
+            snprintf(msg, sizeof(msg), "Missing rebalance id for remove-brick");
+            gf_msg(this->name, GF_LOG_WARNING, 0, GD_MSG_REBALANCE_ID_MISSING,
+                   "%s", msg);
+            ret = 0;
+        } else {
+            gf_uuid_parse(task_id_str, volinfo->rebal.rebalance_id);
+
+            ret = glusterd_copy_uuid_to_dict(volinfo->rebal.rebalance_id,
+                                             rsp_dict, GF_REMOVE_BRICK_TID_KEY,
+                                             SLEN(GF_REMOVE_BRICK_TID_KEY));
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, 0,
+                       GD_MSG_REMOVE_BRICK_ID_SET_FAIL,
+                       "Failed to set remove-brick-id");
+                goto out;
+            }
+        }
+    }
+    if (!gf_uuid_is_null(volinfo->rebal.rebalance_id) &&
+        GD_OP_REMOVE_BRICK == volinfo->rebal.op) {
+        ret = glusterd_copy_uuid_to_dict(volinfo->rebal.rebalance_id, rsp_dict,
+                                         GF_REMOVE_BRICK_TID_KEY,
+                                         SLEN(GF_REMOVE_BRICK_TID_KEY));
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
+                   "Failed to set task-id for volume %s", volname);
+            goto out;
+        }
+    }
+out:
+    return ret;
+}
+int
+glusterd_op_remove_brick(dict_t *dict, char **op_errstr)
+{
+    int ret = -1;
+    char *volname = NULL;
+    glusterd_volinfo_t *volinfo = NULL;
+    char *brick = NULL;
+    int32_t count = 0;
+    int32_t i = 1;
+    char key[64] = "";
+    int keylen;
+    int32_t flag = 0;
+    int need_rebalance = 0;
+    int force = 0;
+    gf1_op_commands cmd = 0;
+    int32_t replica_count = 0;
+    char *task_id_str = NULL;
+    xlator_t *this = NULL;
+    dict_t *bricks_dict = NULL;
+    char *brick_tmpstr = NULL;
+    int start_remove = 0;
+    uint32_t commit_hash = 0;
+    int defrag_cmd = 0;
+    glusterd_conf_t *conf = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+    conf = this->private;
+    GF_VALIDATE_OR_GOTO(this->name, conf, out);
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_BRICK_ADD_FAIL,
+               "Unable to get volume name");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_FOUND,
+               "Unable to allocate memory");
+        goto out;
+    }
+
+    ret = dict_get_int32n(dict, "command", SLEN("command"), &flag);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get command");
+        goto out;
+    }
+    cmd = flag;
+
+    if (GF_OP_CMD_START == cmd)
+        start_remove = 1;
+
+    /* Set task-id, if available, in ctx dict for operations other than
+     * start
+     */
+
+    if (is_origin_glusterd(dict) && (!start_remove)) {
+        if (!gf_uuid_is_null(volinfo->rebal.rebalance_id)) {
+            ret = glusterd_copy_uuid_to_dict(volinfo->rebal.rebalance_id, dict,
+                                             GF_REMOVE_BRICK_TID_KEY,
+                                             SLEN(GF_REMOVE_BRICK_TID_KEY));
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, 0,
+                       GD_MSG_REMOVE_BRICK_ID_SET_FAIL,
+                       "Failed to set remove-brick-id");
+                goto out;
+            }
+        }
+    }
+
+    /* Clear task-id, rebal.op and stored bricks on commmitting/stopping
+     * remove-brick */
+    if ((!start_remove) && (cmd != GF_OP_CMD_STATUS)) {
+        gf_uuid_clear(volinfo->rebal.rebalance_id);
+        volinfo->rebal.op = GD_OP_NONE;
+        dict_unref(volinfo->rebal.dict);
+        volinfo->rebal.dict = NULL;
+    }
+
+    ret = -1;
+    switch (cmd) {
+        case GF_OP_CMD_NONE:
+            goto out;
+
+        case GF_OP_CMD_STATUS:
+            ret = 0;
+            goto out;
+
+        case GF_OP_CMD_STOP:
+        case GF_OP_CMD_START:
+            /* Reset defrag status to 'NOT STARTED' whenever a
+             * remove-brick/rebalance command is issued to remove
+             * stale information from previous run.
+             * Update defrag_cmd as well or it will only be done
+             * for nodes on which the brick to be removed exists.
+             */
+            /* coverity[MIXED_ENUMS] */
+            volinfo->rebal.defrag_cmd = cmd;
+            volinfo->rebal.defrag_status = GF_DEFRAG_STATUS_NOT_STARTED;
+            ret = dict_get_strn(dict, GF_REMOVE_BRICK_TID_KEY,
+                                SLEN(GF_REMOVE_BRICK_TID_KEY), &task_id_str);
+            if (ret) {
+                gf_msg_debug(this->name, errno, "Missing remove-brick-id");
+                ret = 0;
+            } else {
+                gf_uuid_parse(task_id_str, volinfo->rebal.rebalance_id);
+                volinfo->rebal.op = GD_OP_REMOVE_BRICK;
+            }
+            force = 0;
+            break;
+
+        case GF_OP_CMD_COMMIT:
+            force = 1;
+            break;
+
+        case GF_OP_CMD_COMMIT_FORCE:
+
+            if (volinfo->decommission_in_progress) {
+                if (volinfo->rebal.defrag) {
+                    LOCK(&volinfo->rebal.defrag->lock);
+                    /* Fake 'rebalance-complete' so the graph change
+                       happens right away */
+                    volinfo->rebal.defrag_status = GF_DEFRAG_STATUS_COMPLETE;
+
+                    UNLOCK(&volinfo->rebal.defrag->lock);
+                }
+                /* Graph change happens in rebalance _cbk function,
+                   no need to do anything here */
+                /* TODO: '_cbk' function is not doing anything for now */
+            }
+
+            ret = 0;
+            force = 1;
+            break;
+        case GF_OP_CMD_DETACH_START:
+        case GF_OP_CMD_DETACH_COMMIT_FORCE:
+        case GF_OP_CMD_DETACH_COMMIT:
+        case GF_OP_CMD_STOP_DETACH_TIER:
+            break;
+    }
+
+    ret = dict_get_int32n(dict, "count", SLEN("count"), &count);
+    if (ret) {
+        gf_msg("glusterd", GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get count");
+        goto out;
+    }
+    /* Save the list of bricks for later usage only on starting a
+     * remove-brick. Right now this is required for displaying the task
+     * parameters with task status in volume status.
+     */
+
+    if (start_remove) {
+        bricks_dict = dict_new();
+        if (!bricks_dict) {
+            ret = -1;
+            goto out;
+        }
+        ret = dict_set_int32n(bricks_dict, "count", SLEN("count"), count);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+                   "Failed to save remove-brick count");
+            goto out;
+        }
+    }
+
+    while (i <= count) {
+        keylen = snprintf(key, sizeof(key), "brick%d", i);
+        ret = dict_get_strn(dict, key, keylen, &brick);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+                   "Unable to get %s", key);
+            goto out;
+        }
+
+        if (start_remove) {
+            brick_tmpstr = gf_strdup(brick);
+            if (!brick_tmpstr) {
+                ret = -1;
+                gf_msg(this->name, GF_LOG_ERROR, ENOMEM, GD_MSG_NO_MEMORY,
+                       "Failed to duplicate brick name");
+                goto out;
+            }
+            ret = dict_set_dynstrn(bricks_dict, key, keylen, brick_tmpstr);
+            if (ret) {
+                gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+                       "Failed to add brick to dict");
+                goto out;
+            }
+            brick_tmpstr = NULL;
+        }
+
+        ret = glusterd_op_perform_remove_brick(volinfo, brick, force,
+                                               &need_rebalance);
+        if (ret)
+            goto out;
+        i++;
+    }
+
+    if (start_remove)
+        volinfo->rebal.dict = dict_ref(bricks_dict);
+
+    ret = dict_get_int32n(dict, "replica-count", SLEN("replica-count"),
+                          &replica_count);
+    if (!ret) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_DICT_GET_FAILED,
+               "changing replica count %d to %d on volume %s",
+               volinfo->replica_count, replica_count, volinfo->volname);
+        volinfo->replica_count = replica_count;
+        /* A reduction in replica count implies an arbiter volume
+         * earlier is now no longer one. */
+        if (volinfo->arbiter_count)
+            volinfo->arbiter_count = 0;
+        volinfo->sub_count = replica_count;
+        volinfo->dist_leaf_count = glusterd_get_dist_leaf_count(volinfo);
+
+        /*
+         * volinfo->type and sub_count have already been set for
+         * volumes undergoing a detach operation, they should not
+         * be modified here.
+         */
+        if (replica_count == 1) {
+            if (volinfo->type == GF_CLUSTER_TYPE_REPLICATE) {
+                volinfo->type = GF_CLUSTER_TYPE_NONE;
+                /* backward compatibility */
+                volinfo->sub_count = 0;
+            }
+        }
+    }
+    volinfo->subvol_count = (volinfo->brick_count / volinfo->dist_leaf_count);
+
+    if (!glusterd_is_volume_replicate(volinfo) &&
+        conf->op_version >= GD_OP_VERSION_3_12_2) {
+        ret = dict_set_nstrn(volinfo->dict, "performance.client-io-threads",
+                             SLEN("performance.client-io-threads"), "on",
+                             SLEN("on"));
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_SET_FAILED,
+                   "Failed to set "
+                   "performance.client-io-threads to on");
+            goto out;
+        }
+    }
+
+    ret = glusterd_create_volfiles_and_notify_services(volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_WARNING, 0, GD_MSG_VOLFILE_CREATE_FAIL,
+               "failed to create volfiles");
+        goto out;
+    }
+
+    ret = glusterd_store_volinfo(volinfo, GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_WARNING, 0, GD_MSG_VOLINFO_STORE_FAIL,
+               "failed to store volinfo");
+        goto out;
+    }
+
+    if (start_remove && volinfo->status == GLUSTERD_STATUS_STARTED) {
+        ret = glusterd_svcs_reconfigure(volinfo);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_WARNING, 0, GD_MSG_NFS_RECONF_FAIL,
+                   "Unable to reconfigure NFS-Server");
+            goto out;
+        }
+    }
+
+    /* Need to reset the defrag/rebalance status accordingly */
+    switch (volinfo->rebal.defrag_status) {
+        case GF_DEFRAG_STATUS_FAILED:
+        case GF_DEFRAG_STATUS_COMPLETE:
+            volinfo->rebal.defrag_status = 0;
+        /* FALLTHROUGH */
+        default:
+            break;
+    }
+    if (!force && need_rebalance) {
+        if (dict_get_uint32(dict, "commit-hash", &commit_hash) == 0) {
+            volinfo->rebal.commit_hash = commit_hash;
+        }
+        /* perform the rebalance operations */
+        defrag_cmd = GF_DEFRAG_CMD_START_FORCE;
+        /*
+         * We need to set this *before* we issue commands to the
+         * bricks, or else we might end up setting it after the bricks
+         * have responded.  If we fail to send the request(s) we'll
+         * clear it ourselves because nobody else will.
+         */
+        volinfo->decommission_in_progress = 1;
+        char err_str[4096] = "";
+        ret = glusterd_handle_defrag_start(
+            volinfo, err_str, sizeof(err_str), defrag_cmd,
+            glusterd_remove_brick_migrate_cbk, GD_OP_REMOVE_BRICK);
+
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_REBALANCE_START_FAIL,
+                   "failed to start the rebalance");
+            /* TBD: shouldn't we do more than print a message? */
+            volinfo->decommission_in_progress = 0;
+            if (op_errstr)
+                *op_errstr = gf_strdup(err_str);
+        }
+    } else {
+        if (GLUSTERD_STATUS_STARTED == volinfo->status)
+            ret = glusterd_svcs_manager(volinfo);
+    }
+out:
+    GF_FREE(brick_tmpstr);
+    if (bricks_dict)
+        dict_unref(bricks_dict);
+    gf_msg_debug(this->name, 0, "returning %d ", ret);
+    return ret;
+}
+
+int
+glusterd_op_stage_barrier(dict_t *dict, char **op_errstr)
+{
+    int ret = -1;
+    xlator_t *this = NULL;
+    char *volname = NULL;
+    glusterd_volinfo_t *vol = NULL;
+    char *barrier_op = NULL;
+
+    GF_ASSERT(dict);
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Volname not present in "
+               "dict");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &vol);
+    if (ret) {
+        gf_asprintf(op_errstr, "Volume %s does not exist", volname);
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOL_NOT_FOUND, "%s",
+               *op_errstr);
+        goto out;
+    }
+
+    if (!glusterd_is_volume_started(vol)) {
+        gf_asprintf(op_errstr, "Volume %s is not started", volname);
+        ret = -1;
+        goto out;
+    }
+
+    ret = dict_get_strn(dict, "barrier", SLEN("barrier"), &barrier_op);
+    if (ret == -1) {
+        gf_asprintf(op_errstr,
+                    "Barrier op for volume %s not present "
+                    "in dict",
+                    volname);
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               *op_errstr);
+        goto out;
+    }
+    ret = 0;
+out:
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+    return ret;
+}
+
+int
+glusterd_op_barrier(dict_t *dict, char **op_errstr)
+{
+    int ret = -1;
+    xlator_t *this = NULL;
+    char *volname = NULL;
+    glusterd_volinfo_t *vol = NULL;
+    char *barrier_op = NULL;
+
+    GF_ASSERT(dict);
+    this = THIS;
+    GF_ASSERT(this);
+
+    ret = dict_get_strn(dict, "volname", SLEN("volname"), &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Volname not present in "
+               "dict");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &vol);
+    if (ret) {
+        gf_asprintf(op_errstr, "Volume %s does not exist", volname);
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOL_NOT_FOUND, "%s",
+               *op_errstr);
+        goto out;
+    }
+
+    ret = dict_get_strn(dict, "barrier", SLEN("barrier"), &barrier_op);
+    if (ret) {
+        gf_asprintf(op_errstr,
+                    "Barrier op for volume %s not present "
+                    "in dict",
+                    volname);
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED, "%s",
+               *op_errstr);
+        goto out;
+    }
+
+    ret = dict_set_dynstr_with_alloc(vol->dict, "features.barrier", barrier_op);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set barrier op in"
+               " volume option dict");
+        goto out;
+    }
+
+    gd_update_volume_op_versions(vol);
+    ret = glusterd_create_volfiles(vol);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_VOLFILE_CREATE_FAIL,
+               "Failed to create volfiles");
+        goto out;
+    }
+    ret = glusterd_store_volinfo(vol, GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+
+out:
+    gf_msg_debug(this->name, 0, "Returning %d", ret);
+    return ret;
+}
+
+int
+glusterd_handle_add_tier_brick(rpcsvc_request_t *req)
+{
+    return 0;
+}
+
+int
+glusterd_handle_attach_tier(rpcsvc_request_t *req)
+{
+    return 0;
+}
+
+int
+glusterd_handle_detach_tier(rpcsvc_request_t *req)
+{
+    return 0;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-conn-helper.c b/xlators/mgmt/glusterd/src/glusterd-conn-helper.c
new file mode 100644
index 00000000000..a7f54ec24b7
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-conn-helper.c
@@ -0,0 +1,21 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include "glusterd-conn-mgmt.h"
+#include "glusterd-svc-mgmt.h"
+
+#define _LGPL_SOURCE
+#include <urcu/rculist.h>
+
+glusterd_svc_t *
+glusterd_conn_get_svc_object(glusterd_conn_t *conn)
+{
+    return cds_list_entry(conn, glusterd_svc_t, conn);
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-conn-helper.h b/xlators/mgmt/glusterd/src/glusterd-conn-helper.h
new file mode 100644
index 00000000000..6f500309175
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-conn-helper.h
@@ -0,0 +1,21 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef _GLUSTERD_CONN_HELPER_H_
+#define _GLUSTERD_CONN_HELPER_H_
+
+#include "rpc-clnt.h"
+
+#include "glusterd-conn-mgmt.h"
+
+glusterd_svc_t *
+glusterd_conn_get_svc_object(glusterd_conn_t *conn);
+
+#endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.c b/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.c
new file mode 100644
index 00000000000..5c01f0c70b6
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.c
@@ -0,0 +1,191 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include <glusterfs/xlator.h>
+#include "rpc-clnt.h"
+#include "glusterd.h"
+#include "glusterd-conn-mgmt.h"
+#include "glusterd-conn-helper.h"
+#include "glusterd-utils.h"
+#include "glusterd-messages.h"
+
+int
+glusterd_conn_init(glusterd_conn_t *conn, char *sockpath, int frame_timeout,
+                   glusterd_conn_notify_t notify)
+{
+    int ret = -1;
+    dict_t *options = NULL;
+    struct rpc_clnt *rpc = NULL;
+    xlator_t *this = THIS;
+    glusterd_svc_t *svc = NULL;
+
+    if (!this) {
+        gf_smsg(THIS->name, GF_LOG_ERROR, errno, GD_MSG_XLATOR_NOT_DEFINED,
+                NULL);
+        goto out;
+    }
+
+    options = dict_new();
+    if (!options) {
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_CREATE_FAIL, NULL);
+        goto out;
+    }
+
+    svc = glusterd_conn_get_svc_object(conn);
+    if (!svc) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SVC_GET_FAIL,
+               "Failed to get the service");
+        goto out;
+    }
+
+    ret = rpc_transport_unix_options_build(options, sockpath, frame_timeout);
+    if (ret)
+        goto out;
+
+    ret = dict_set_int32n(options, "transport.socket.ignore-enoent",
+                          SLEN("transport.socket.ignore-enoent"), 1);
+    if (ret) {
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+                "Key=transport.socket.ignore-enoent", NULL);
+        goto out;
+    }
+
+    /* @options is free'd by rpc_transport when destroyed */
+    rpc = rpc_clnt_new(options, this, (char *)svc->name, 16);
+    if (!rpc) {
+        ret = -1;
+        goto out;
+    }
+
+    ret = rpc_clnt_register_notify(rpc, glusterd_conn_common_notify, conn);
+    if (ret)
+        goto out;
+
+    ret = snprintf(conn->sockpath, sizeof(conn->sockpath), "%s", sockpath);
+    if (ret < 0) {
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_COPY_FAIL, NULL);
+        goto out;
+    } else
+        ret = 0;
+
+    conn->frame_timeout = frame_timeout;
+    conn->rpc = rpc;
+    conn->notify = notify;
+out:
+    if (options)
+        dict_unref(options);
+    if (ret) {
+        if (rpc) {
+            rpc_clnt_unref(rpc);
+            rpc = NULL;
+        }
+    }
+    return ret;
+}
+
+int
+glusterd_conn_term(glusterd_conn_t *conn)
+{
+    rpc_clnt_unref(conn->rpc);
+    return 0;
+}
+
+int
+glusterd_conn_connect(glusterd_conn_t *conn)
+{
+    return rpc_clnt_start(conn->rpc);
+}
+
+int
+glusterd_conn_disconnect(glusterd_conn_t *conn)
+{
+    rpc_clnt_disable(conn->rpc);
+
+    return 0;
+}
+
+int
+__glusterd_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                              rpc_clnt_event_t event, void *data)
+{
+    glusterd_conn_t *conn = mydata;
+
+    /* Silently ignoring this error, exactly like the current
+     * implementation */
+    if (!conn)
+        return 0;
+
+    return conn->notify(conn, event);
+}
+
+int
+glusterd_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                            rpc_clnt_event_t event, void *data)
+{
+    return glusterd_big_locked_notify(rpc, mydata, event, data,
+                                      __glusterd_conn_common_notify);
+}
+
+int32_t
+glusterd_conn_build_socket_filepath(char *rundir, uuid_t uuid, char *socketpath,
+                                    int len)
+{
+    char sockfilepath[PATH_MAX] = {
+        0,
+    };
+
+    snprintf(sockfilepath, sizeof(sockfilepath), "%s/run-%s", rundir,
+             uuid_utoa(uuid));
+
+    glusterd_set_socket_filepath(sockfilepath, socketpath, len);
+    return 0;
+}
+
+int
+__glusterd_muxsvc_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                                     rpc_clnt_event_t event, void *data)
+{
+    glusterd_conf_t *conf = THIS->private;
+    glusterd_svc_proc_t *mux_proc = mydata;
+    int ret = -1;
+
+    /* Silently ignoring this error, exactly like the current
+     * implementation */
+    if (!mux_proc)
+        return 0;
+
+    if (event == RPC_CLNT_DESTROY) {
+        /*RPC_CLNT_DESTROY will only called after mux_proc detached from the
+         * list. So it is safe to call without lock. Processing
+         * RPC_CLNT_DESTROY under a lock will lead to deadlock.
+         */
+        if (mux_proc->data) {
+            glusterd_volinfo_unref(mux_proc->data);
+            mux_proc->data = NULL;
+        }
+        GF_FREE(mux_proc);
+        ret = 0;
+    } else {
+        pthread_mutex_lock(&conf->attach_lock);
+        {
+            ret = mux_proc->notify(mux_proc, event);
+        }
+        pthread_mutex_unlock(&conf->attach_lock);
+    }
+    return ret;
+}
+
+int
+glusterd_muxsvc_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                                   rpc_clnt_event_t event, void *data)
+{
+    return glusterd_big_locked_notify(rpc, mydata, event, data,
+                                      __glusterd_muxsvc_conn_common_notify);
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.h b/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.h
new file mode 100644
index 00000000000..1b225621ab1
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-conn-mgmt.h
@@ -0,0 +1,53 @@
+/*
+   Copyright (c) 2014 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#ifndef _GLUSTERD_CONN_MGMT_H_
+#define _GLUSTERD_CONN_MGMT_H_
+
+#include "rpc-clnt.h"
+
+typedef struct glusterd_conn_ glusterd_conn_t;
+
+typedef int (*glusterd_conn_notify_t)(glusterd_conn_t *conn,
+                                      rpc_clnt_event_t event);
+
+struct glusterd_conn_ {
+    struct rpc_clnt *rpc;
+    /* Existing daemons tend to specialize their respective
+     * notify implementations, so ... */
+    glusterd_conn_notify_t notify;
+    int frame_timeout;
+    char sockpath[PATH_MAX];
+};
+
+int
+glusterd_conn_init(glusterd_conn_t *conn, char *sockpath, int frame_timeout,
+                   glusterd_conn_notify_t notify);
+
+int
+glusterd_conn_term(glusterd_conn_t *conn);
+
+int
+glusterd_conn_connect(glusterd_conn_t *conn);
+
+int
+glusterd_conn_disconnect(glusterd_conn_t *conn);
+
+int
+glusterd_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                            rpc_clnt_event_t event, void *data);
+int
+glusterd_muxsvc_conn_common_notify(struct rpc_clnt *rpc, void *mydata,
+                                   rpc_clnt_event_t event, void *data);
+
+int32_t
+glusterd_conn_build_socket_filepath(char *rundir, uuid_t uuid, char *socketpath,
+                                    int len);
+#endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-errno.h b/xlators/mgmt/glusterd/src/glusterd-errno.h
new file mode 100644
index 00000000000..c74070e0e8d
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-errno.h
@@ -0,0 +1,33 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#ifndef _GLUSTERD_ERRNO_H
+#define _GLUSTERD_ERRNO_H
+
+enum glusterd_op_errno {
+    EG_INTRNL = 30800,    /* Internal Error                    */
+    EG_OPNOTSUP = 30801,  /* Gluster Op Not Supported          */
+    EG_ANOTRANS = 30802,  /* Another Transaction in Progress   */
+    EG_BRCKDWN = 30803,   /* One or more brick is down         */
+    EG_NODEDWN = 30804,   /* One or more node is down          */
+    EG_HRDLMT = 30805,    /* Hard Limit is reached             */
+    EG_NOVOL = 30806,     /* Volume does not exist             */
+    EG_NOSNAP = 30807,    /* Snap does not exist               */
+    EG_RBALRUN = 30808,   /* Rebalance is running              */
+    EG_VOLRUN = 30809,    /* Volume is running                 */
+    EG_VOLSTP = 30810,    /* Volume is not running             */
+    EG_VOLEXST = 30811,   /* Volume exists                     */
+    EG_SNAPEXST = 30812,  /* Snapshot exists                   */
+    EG_ISSNAP = 30813,    /* Volume is a snap volume           */
+    EG_GEOREPRUN = 30814, /* Geo-Replication is running        */
+    EG_NOTTHINP = 30815,  /* Bricks are not thinly provisioned */
+    EG_NOGANESHA = 30816, /* Global ganesha is not enabled   */
+};
+
+#endif
diff --git a/xlators/mgmt/glusterd/src/glusterd-ganesha.c b/xlators/mgmt/glusterd/src/glusterd-ganesha.c
new file mode 100644
index 00000000000..f08bd6cebee
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-ganesha.c
@@ -0,0 +1,927 @@
+/*
+   Copyright (c) 2015 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include <glusterfs/common-utils.h>
+#include "glusterd.h"
+#include "glusterd-op-sm.h"
+#include "glusterd-store.h"
+#include "glusterd-utils.h"
+#include "glusterd-volgen.h"
+#include "glusterd-messages.h"
+#include <glusterfs/syscall.h>
+
+#include <ctype.h>
+
+int
+start_ganesha(char **op_errstr);
+
+typedef struct service_command {
+    char *binary;
+    char *service;
+    int (*action)(struct service_command *, char *);
+} service_command;
+
+/* parsing_ganesha_ha_conf will allocate the returned string
+ * to be freed (GF_FREE) by the caller
+ * return NULL if error or not found */
+static char *
+parsing_ganesha_ha_conf(const char *key)
+{
+#define MAX_LINE 1024
+    char scratch[MAX_LINE * 2] = {
+        0,
+    };
+    char *value = NULL, *pointer = NULL, *end_pointer = NULL;
+    FILE *fp;
+
+    fp = fopen(GANESHA_HA_CONF, "r");
+    if (fp == NULL) {
+        gf_msg(THIS->name, GF_LOG_ERROR, errno, GD_MSG_FILE_OP_FAILED,
+               "couldn't open the file %s", GANESHA_HA_CONF);
+        goto end_ret;
+    }
+    while ((pointer = fgets(scratch, MAX_LINE, fp)) != NULL) {
+        /* Read config file until we get matching "^[[:space:]]*key" */
+        if (*pointer == '#') {
+            continue;
+        }
+        while (isblank(*pointer)) {
+            pointer++;
+        }
+        if (strncmp(pointer, key, strlen(key))) {
+            continue;
+        }
+        pointer += strlen(key);
+        /* key found : if we fail to parse, we'll return an error
+         * rather than trying next one
+         * - supposition : conf file is bash compatible : no space
+         *   around the '=' */
+        if (*pointer != '=') {
+            gf_msg(THIS->name, GF_LOG_ERROR, errno,
+                   GD_MSG_GET_CONFIG_INFO_FAILED, "Parsing %s failed at key %s",
+                   GANESHA_HA_CONF, key);
+            goto end_close;
+        }
+        pointer++; /* jump the '=' */
+
+        if (*pointer == '"' || *pointer == '\'') {
+            /* dont get the quote */
+            pointer++;
+        }
+        end_pointer = pointer;
+        /* stop at the next closing quote or  blank/newline */
+        do {
+            end_pointer++;
+        } while (!(*end_pointer == '\'' || *end_pointer == '"' ||
+                   isspace(*end_pointer) || *end_pointer == '\0'));
+        *end_pointer = '\0';
+
+        /* got it. copy it and return */
+        value = gf_strdup(pointer);
+        break;
+    }
+
+end_close:
+    fclose(fp);
+end_ret:
+    return value;
+}
+
+static int
+sc_systemctl_action(struct service_command *sc, char *command)
+{
+    runner_t runner = {
+        0,
+    };
+
+    runinit(&runner);
+    runner_add_args(&runner, sc->binary, command, sc->service, NULL);
+    return runner_run(&runner);
+}
+
+static int
+sc_service_action(struct service_command *sc, char *command)
+{
+    runner_t runner = {
+        0,
+    };
+
+    runinit(&runner);
+    runner_add_args(&runner, sc->binary, sc->service, command, NULL);
+    return runner_run(&runner);
+}
+
+static int
+manage_service(char *action)
+{
+    int i = 0;
+    int ret = 0;
+    struct service_command sc_list[] = {{.binary = "/bin/systemctl",
+                                         .service = "nfs-ganesha",
+                                         .action = sc_systemctl_action},
+                                        {.binary = "/sbin/invoke-rc.d",
+                                         .service = "nfs-ganesha",
+                                         .action = sc_service_action},
+                                        {.binary = "/sbin/service",
+                                         .service = "nfs-ganesha",
+                                         .action = sc_service_action},
+                                        {.binary = NULL}};
+
+    while (sc_list[i].binary != NULL) {
+        ret = sys_access(sc_list[i].binary, X_OK);
+        if (ret == 0) {
+            gf_msg_debug(THIS->name, 0, "%s found.", sc_list[i].binary);
+            return sc_list[i].action(&sc_list[i], action);
+        }
+        i++;
+    }
+    gf_msg(THIS->name, GF_LOG_ERROR, 0, GD_MSG_UNRECOGNIZED_SVC_MNGR,
+           "Could not %s NFS-Ganesha.Service manager for distro"
+           " not recognized.",
+           action);
+    return ret;
+}
+
+/*
+ * Check if the cluster is a ganesha cluster or not *
+ */
+gf_boolean_t
+glusterd_is_ganesha_cluster()
+{
+    int ret = -1;
+    glusterd_conf_t *priv = NULL;
+    xlator_t *this = NULL;
+    gf_boolean_t ret_bool = _gf_false;
+
+    this = THIS;
+    GF_VALIDATE_OR_GOTO("ganesha", this, out);
+    priv = this->private;
+    GF_VALIDATE_OR_GOTO(this->name, priv, out);
+
+    ret = dict_get_str_boolean(priv->opts, GLUSTERD_STORE_KEY_GANESHA_GLOBAL,
+                               _gf_false);
+    if (ret == _gf_true) {
+        ret_bool = _gf_true;
+        gf_msg_debug(this->name, 0, "nfs-ganesha is enabled for the cluster");
+    } else
+        gf_msg_debug(this->name, 0, "nfs-ganesha is disabled for the cluster");
+
+out:
+    return ret_bool;
+}
+
+/* Check if ganesha.enable is set to 'on', that checks if
+ * a  particular volume is exported via NFS-Ganesha */
+gf_boolean_t
+glusterd_check_ganesha_export(glusterd_volinfo_t *volinfo)
+{
+    char *value = NULL;
+    gf_boolean_t is_exported = _gf_false;
+    int ret = 0;
+
+    ret = glusterd_volinfo_get(volinfo, "ganesha.enable", &value);
+    if ((ret == 0) && value) {
+        if (strcmp(value, "on") == 0) {
+            gf_msg_debug(THIS->name, 0,
+                         "ganesha.enable set"
+                         " to %s",
+                         value);
+            is_exported = _gf_true;
+        }
+    }
+    return is_exported;
+}
+
+/* *
+ * The below function is called as part of commit phase for volume set option
+ * "ganesha.enable". If the value is "on", it creates export configuration file
+ * and then export the volume via dbus command. Incase of "off", the volume
+ * will be already unexported during stage phase, so it will remove the conf
+ * file from shared storage
+ */
+int
+glusterd_check_ganesha_cmd(char *key, char *value, char **errstr, dict_t *dict)
+{
+    int ret = 0;
+    char *volname = NULL;
+
+    GF_ASSERT(key);
+    GF_ASSERT(value);
+    GF_ASSERT(dict);
+
+    if ((strcmp(key, "ganesha.enable") == 0)) {
+        if ((strcmp(value, "on")) && (strcmp(value, "off"))) {
+            gf_asprintf(errstr,
+                        "Invalid value"
+                        " for volume set command. Use on/off only.");
+            ret = -1;
+            goto out;
+        }
+        if (strcmp(value, "on") == 0) {
+            ret = glusterd_handle_ganesha_op(dict, errstr, key, value);
+
+        } else if (is_origin_glusterd(dict)) {
+            ret = dict_get_str(dict, "volname", &volname);
+            if (ret) {
+                gf_msg("glusterd-ganesha", GF_LOG_ERROR, errno,
+                       GD_MSG_DICT_GET_FAILED, "Unable to get volume name");
+                goto out;
+            }
+            ret = manage_export_config(volname, "off", errstr);
+        }
+    }
+out:
+    if (ret) {
+        gf_msg("glusterd-ganesha", GF_LOG_ERROR, 0,
+               GD_MSG_NFS_GNS_OP_HANDLE_FAIL,
+               "Handling NFS-Ganesha"
+               " op failed.");
+    }
+    return ret;
+}
+
+int
+glusterd_op_stage_set_ganesha(dict_t *dict, char **op_errstr)
+{
+    int ret = -1;
+    char *value = NULL;
+    char *str = NULL;
+    glusterd_conf_t *priv = NULL;
+    xlator_t *this = NULL;
+
+    GF_ASSERT(dict);
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = dict_get_str(dict, "value", &value);
+    if (value == NULL) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "value not present.");
+        goto out;
+    }
+    /* This dict_get will fail if the user had never set the key before */
+    /*Ignoring the ret value and proceeding */
+    ret = dict_get_str(priv->opts, GLUSTERD_STORE_KEY_GANESHA_GLOBAL, &str);
+    if (str ? strcmp(value, str) == 0 : strcmp(value, "disable") == 0) {
+        gf_asprintf(op_errstr, "nfs-ganesha is already %sd.", value);
+        ret = -1;
+        goto out;
+    }
+
+    if (strcmp(value, "enable") == 0) {
+        ret = start_ganesha(op_errstr);
+        if (ret) {
+            gf_msg(THIS->name, GF_LOG_ERROR, 0, GD_MSG_NFS_GNS_START_FAIL,
+                   "Could not start NFS-Ganesha");
+        }
+    } else {
+        ret = stop_ganesha(op_errstr);
+        if (ret)
+            gf_msg_debug(THIS->name, 0,
+                         "Could not stop "
+                         "NFS-Ganesha.");
+    }
+
+out:
+
+    if (ret) {
+        if (!(*op_errstr)) {
+            *op_errstr = gf_strdup("Error, Validation Failed");
+            gf_msg_debug(this->name, 0, "Error, Cannot Validate option :%s",
+                         GLUSTERD_STORE_KEY_GANESHA_GLOBAL);
+        } else {
+            gf_msg_debug(this->name, 0, "Error, Cannot Validate option");
+        }
+    }
+    return ret;
+}
+
+int
+glusterd_op_set_ganesha(dict_t *dict, char **errstr)
+{
+    int ret = 0;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+    char *key = NULL;
+    char *value = NULL;
+    char *next_version = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+    GF_ASSERT(dict);
+
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = dict_get_str(dict, "key", &key);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Couldn't get key in global option set");
+        goto out;
+    }
+
+    ret = dict_get_str(dict, "value", &value);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Couldn't get value in global option set");
+        goto out;
+    }
+
+    ret = glusterd_handle_ganesha_op(dict, errstr, key, value);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_NFS_GNS_SETUP_FAIL,
+               "Initial NFS-Ganesha set up failed");
+        ret = -1;
+        goto out;
+    }
+    ret = dict_set_dynstr_with_alloc(priv->opts,
+                                     GLUSTERD_STORE_KEY_GANESHA_GLOBAL, value);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_WARNING, errno, GD_MSG_DICT_SET_FAILED,
+               "Failed to set"
+               " nfs-ganesha in dict.");
+        goto out;
+    }
+    ret = glusterd_get_next_global_opt_version_str(priv->opts, &next_version);
+    if (ret) {
+        gf_msg_debug(THIS->name, 0,
+                     "Could not fetch "
+                     " global op version");
+        goto out;
+    }
+    ret = dict_set_str(priv->opts, GLUSTERD_GLOBAL_OPT_VERSION, next_version);
+    if (ret)
+        goto out;
+
+    ret = glusterd_store_options(this, priv->opts);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_STORE_FAIL,
+               "Failed to store options");
+        goto out;
+    }
+
+out:
+    gf_msg_debug(this->name, 0, "returning %d", ret);
+    return ret;
+}
+
+/* Following function parse GANESHA_HA_CONF
+ * The sample file looks like below,
+ * HA_NAME="ganesha-ha-360"
+ * HA_VOL_NAME="ha-state"
+ * HA_CLUSTER_NODES="server1,server2"
+ * VIP_rhs_1="10.x.x.x"
+ * VIP_rhs_2="10.x.x.x." */
+
+/* Check if the localhost is listed as one of nfs-ganesha nodes */
+gf_boolean_t
+check_host_list(void)
+{
+    glusterd_conf_t *priv = NULL;
+    char *hostname, *hostlist;
+    gf_boolean_t ret = _gf_false;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    priv = THIS->private;
+    GF_ASSERT(priv);
+
+    hostlist = parsing_ganesha_ha_conf("HA_CLUSTER_NODES");
+    if (hostlist == NULL) {
+        gf_msg(this->name, GF_LOG_INFO, errno, GD_MSG_GET_CONFIG_INFO_FAILED,
+               "couldn't get HA_CLUSTER_NODES from file %s", GANESHA_HA_CONF);
+        return _gf_false;
+    }
+
+    /* Hostlist is a comma separated list now */
+    hostname = strtok(hostlist, ",");
+    while (hostname != NULL) {
+        ret = gf_is_local_addr(hostname);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_INFO, 0, GD_MSG_NFS_GNS_HOST_FOUND,
+                   "ganesha host found "
+                   "Hostname is %s",
+                   hostname);
+            break;
+        }
+        hostname = strtok(NULL, ",");
+    }
+
+    GF_FREE(hostlist);
+    return ret;
+}
+
+int
+gd_ganesha_send_dbus(char *volname, char *value)
+{
+    runner_t runner = {
+        0,
+    };
+    int ret = -1;
+    runinit(&runner);
+
+    GF_VALIDATE_OR_GOTO("glusterd-ganesha", volname, out);
+    GF_VALIDATE_OR_GOTO("glusterd-ganesha", value, out);
+
+    ret = 0;
+    if (check_host_list()) {
+        /* Check whether ganesha is running on this node */
+        if (manage_service("status")) {
+            gf_msg("glusterd-ganesha", GF_LOG_WARNING, 0,
+                   GD_MSG_GANESHA_NOT_RUNNING,
+                   "Export failed, NFS-Ganesha is not running");
+        } else {
+            runner_add_args(&runner, GANESHA_PREFIX "/dbus-send.sh", CONFDIR,
+                            value, volname, NULL);
+            ret = runner_run(&runner);
+        }
+    }
+out:
+    return ret;
+}
+
+int
+manage_export_config(char *volname, char *value, char **op_errstr)
+{
+    runner_t runner = {
+        0,
+    };
+    int ret = -1;
+
+    GF_ASSERT(volname);
+    runinit(&runner);
+    runner_add_args(&runner, GANESHA_PREFIX "/create-export-ganesha.sh",
+                    CONFDIR, value, volname, NULL);
+    ret = runner_run(&runner);
+
+    if (ret && op_errstr)
+        gf_asprintf(op_errstr,
+                    "Failed to create"
+                    " NFS-Ganesha export config file.");
+
+    return ret;
+}
+
+/* Exports and unexports a particular volume via NFS-Ganesha */
+int
+ganesha_manage_export(dict_t *dict, char *value,
+                      gf_boolean_t update_cache_invalidation, char **op_errstr)
+{
+    int ret = -1;
+    glusterd_volinfo_t *volinfo = NULL;
+    dict_t *vol_opts = NULL;
+    char *volname = NULL;
+    xlator_t *this = NULL;
+    glusterd_conf_t *priv = NULL;
+    gf_boolean_t option = _gf_false;
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+
+    GF_ASSERT(value);
+    GF_ASSERT(dict);
+    GF_ASSERT(priv);
+
+    ret = dict_get_str(dict, "volname", &volname);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_GET_FAILED,
+               "Unable to get volume name");
+        goto out;
+    }
+    ret = gf_string2boolean(value, &option);
+    if (ret == -1) {
+        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_INVALID_ENTRY,
+               "invalid value.");
+        goto out;
+    }
+
+    ret = glusterd_volinfo_find(volname, &volinfo);
+    if (ret) {
+        gf_msg(this->name, GF_LOG_ERROR, EINVAL, GD_MSG_VOL_NOT_FOUND,
+               FMTSTR_CHECK_VOL_EXISTS, volname);
+        goto out;
+    }
+
+    ret = glusterd_check_ganesha_export(volinfo);
+    if (ret && option) {
+        gf_asprintf(op_errstr,
+                    "ganesha.enable "
+                    "is already 'on'.");
+        ret = -1;
+        goto out;
+
+    } else if (!option && !ret) {
+        gf_asprintf(op_errstr,
+                    "ganesha.enable "
+                    "is already 'off'.");
+        ret = -1;
+        goto out;
+    }
+
+    /* Check if global option is enabled, proceed only then */
+    ret = dict_get_str_boolean(priv->opts, GLUSTERD_STORE_KEY_GANESHA_GLOBAL,
+                               _gf_false);
+    if (ret == -1) {
+        gf_msg_debug(this->name, 0,
+                     "Failed to get "
+                     "global option dict.");
+        gf_asprintf(op_errstr,
+                    "The option "
+                    "nfs-ganesha should be "
+                    "enabled before setting ganesha.enable.");
+        goto out;
+    }
+    if (!ret) {
+        gf_asprintf(op_errstr,
+                    "The option "
+                    "nfs-ganesha should be "
+                    "enabled before setting ganesha.enable.");
+        ret = -1;
+        goto out;
+    }
+
+    /* *
+     * Create the export file from the node where ganesha.enable "on"
+     * is executed
+     * */
+    if (option && is_origin_glusterd(dict)) {
+        ret = manage_export_config(volname, "on", op_errstr);
+        if (ret) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_EXPORT_FILE_CREATE_FAIL,
+                   "Failed to create"
+                   "export file for NFS-Ganesha\n");
+            goto out;
+        }
+    }
+    ret = gd_ganesha_send_dbus(volname, value);
+    if (ret) {
+        gf_asprintf(op_errstr,
+                    "Dynamic export addition/deletion failed."
+                    " Please see log file for details");
+        goto out;
+    }
+    if (update_cache_invalidation) {
+        vol_opts = volinfo->dict;
+        ret = dict_set_dynstr_with_alloc(vol_opts,
+                                         "features.cache-invalidation", value);
+        if (ret)
+            gf_asprintf(op_errstr,
+                        "Cache-invalidation could not"
+                        " be set to %s.",
+                        value);
+        ret = glusterd_store_volinfo(volinfo,
+                                     GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+        if (ret)
+            gf_asprintf(op_errstr, "failed to store volinfo for %s",
+                        volinfo->volname);
+    }
+out:
+    return ret;
+}
+
+int
+tear_down_cluster(gf_boolean_t run_teardown)
+{
+    int ret = 0;
+    runner_t runner = {
+        0,
+    };
+    struct stat st = {
+        0,
+    };
+    DIR *dir = NULL;
+    struct dirent *entry = NULL;
+    struct dirent scratch[2] = {
+        {
+            0,
+        },
+    };
+    char path[PATH_MAX] = {
+        0,
+    };
+
+    if (run_teardown) {
+        runinit(&runner);
+        runner_add_args(&runner, GANESHA_PREFIX "/ganesha-ha.sh", "teardown",
+                        CONFDIR, NULL);
+        ret = runner_run(&runner);
+        /* *
+         * Remove all the entries in CONFDIR expect ganesha.conf and
+         * ganesha-ha.conf
+         */
+        dir = sys_opendir(CONFDIR);
+        if (!dir) {
+            gf_msg_debug(THIS->name, 0,
+                         "Failed to open directory %s. "
+                         "Reason : %s",
+                         CONFDIR, strerror(errno));
+            ret = 0;
+            goto out;
+        }
+
+        while ((entry = sys_readdir(dir, scratch))) {
+            if (gf_irrelevant_entry(entry))
+                continue;
+            snprintf(path, PATH_MAX, "%s/%s", CONFDIR, entry->d_name);
+            ret = sys_lstat(path, &st);
+            if (ret == -1) {
+                gf_msg_debug(THIS->name, 0,
+                             "Failed to stat entry %s :"
+                             " %s",
+                             path, strerror(errno));
+                goto out;
+            }
+
+            if (strcmp(entry->d_name, "ganesha.conf") == 0 ||
+                strcmp(entry->d_name, "ganesha-ha.conf") == 0)
+                gf_msg_debug(THIS->name, 0,
+                             " %s is not required"
+                             " to remove",
+                             path);
+            else if (S_ISDIR(st.st_mode))
+                ret = recursive_rmdir(path);
+            else
+                ret = sys_unlink(path);
+
+            if (ret) {
+                gf_msg_debug(THIS->name, 0,
+                             " Failed to remove %s. "
+                             "Reason : %s",
+                             path, strerror(errno));
+            }
+
+            gf_msg_debug(THIS->name, 0, "%s %s",
+                         ret ? "Failed to remove" : "Removed", entry->d_name);
+        }
+
+        ret = sys_closedir(dir);
+        if (ret) {
+            gf_msg_debug(THIS->name, 0,
+                         "Failed to close dir %s. Reason :"
+                         " %s",
+                         CONFDIR, strerror(errno));
+        }
+        goto exit;
+    }
+
+out:
+    if (dir && sys_closedir(dir)) {
+        gf_msg_debug(THIS->name, 0,
+                     "Failed to close dir %s. Reason :"
+                     " %s",
+                     CONFDIR, strerror(errno));
+    }
+exit:
+    return ret;
+}
+
+int
+setup_cluster(gf_boolean_t run_setup)
+{
+    int ret = 0;
+    runner_t runner = {
+        0,
+    };
+
+    if (run_setup) {
+        runinit(&runner);
+        runner_add_args(&runner, GANESHA_PREFIX "/ganesha-ha.sh", "setup",
+                        CONFDIR, NULL);
+        ret = runner_run(&runner);
+    }
+    return ret;
+}
+
+static int
+teardown(gf_boolean_t run_teardown, char **op_errstr)
+{
+    runner_t runner = {
+        0,
+    };
+    int ret = 1;
+    glusterd_volinfo_t *volinfo = NULL;
+    glusterd_conf_t *priv = NULL;
+    dict_t *vol_opts = NULL;
+
+    priv = THIS->private;
+
+    ret = tear_down_cluster(run_teardown);
+    if (ret == -1) {
+        gf_asprintf(op_errstr,
+                    "Cleanup of NFS-Ganesha"
+                    " HA config failed.");
+        goto out;
+    }
+
+    runinit(&runner);
+    runner_add_args(&runner, GANESHA_PREFIX "/ganesha-ha.sh", "cleanup",
+                    CONFDIR, NULL);
+    ret = runner_run(&runner);
+    if (ret)
+        gf_msg_debug(THIS->name, 0,
+                     "Could not clean up"
+                     " NFS-Ganesha related config");
+
+    cds_list_for_each_entry(volinfo, &priv->volumes, vol_list)
+    {
+        vol_opts = volinfo->dict;
+        /* All the volumes exported via NFS-Ganesha will be
+        unexported, hence setting the appropriate keys */
+        ret = dict_set_str(vol_opts, "features.cache-invalidation", "off");
+        if (ret)
+            gf_msg(THIS->name, GF_LOG_WARNING, errno, GD_MSG_DICT_SET_FAILED,
+                   "Could not set features.cache-invalidation "
+                   "to off for %s",
+                   volinfo->volname);
+
+        ret = dict_set_str(vol_opts, "ganesha.enable", "off");
+        if (ret)
+            gf_msg(THIS->name, GF_LOG_WARNING, errno, GD_MSG_DICT_SET_FAILED,
+                   "Could not set ganesha.enable to off for %s",
+                   volinfo->volname);
+
+        ret = glusterd_store_volinfo(volinfo,
+                                     GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+        if (ret)
+            gf_msg(THIS->name, GF_LOG_WARNING, 0, GD_MSG_VOLINFO_SET_FAIL,
+                   "failed to store volinfo for %s", volinfo->volname);
+    }
+out:
+    return ret;
+}
+
+int
+stop_ganesha(char **op_errstr)
+{
+    int ret = 0;
+    runner_t runner = {
+        0,
+    };
+
+    if (check_host_list()) {
+        runinit(&runner);
+        runner_add_args(&runner, GANESHA_PREFIX "/ganesha-ha.sh",
+                        "--setup-ganesha-conf-files", CONFDIR, "no", NULL);
+        ret = runner_run(&runner);
+        if (ret) {
+            gf_asprintf(op_errstr,
+                        "removal of symlink ganesha.conf "
+                        "in /etc/ganesha failed");
+        }
+        ret = manage_service("stop");
+        if (ret)
+            gf_asprintf(op_errstr,
+                        "NFS-Ganesha service could not"
+                        "be stopped.");
+    }
+    return ret;
+}
+
+int
+start_ganesha(char **op_errstr)
+{
+    int ret = -1;
+    glusterd_volinfo_t *volinfo = NULL;
+    glusterd_conf_t *priv = NULL;
+    runner_t runner = {
+        0,
+    };
+
+    priv = THIS->private;
+    GF_ASSERT(priv);
+
+    cds_list_for_each_entry(volinfo, &priv->volumes, vol_list)
+    {
+#ifdef BUILD_GNFS
+        /* Gluster-nfs has to be disabled across the trusted pool */
+        /* before attempting to start nfs-ganesha */
+        ret = dict_set_str_sizen(volinfo->dict, NFS_DISABLE_MAP_KEY, "on");
+        if (ret)
+            goto out;
+#endif
+        ret = glusterd_store_volinfo(volinfo,
+                                     GLUSTERD_VOLINFO_VER_AC_INCREMENT);
+        if (ret) {
+            *op_errstr = gf_strdup(
+                "Failed to store the "
+                "Volume information");
+            goto out;
+        }
+    }
+
+    /* If the nfs svc is not initialized it means that the service is not
+     * running, hence we can skip the process of stopping gluster-nfs
+     * service
+     */
+#ifdef BUILD_GNFS
+    if (priv->nfs_svc.inited) {
+        ret = priv->nfs_svc.stop(&(priv->nfs_svc), SIGKILL);
+        if (ret) {
+            ret = -1;
+            gf_asprintf(op_errstr,
+                        "Gluster-NFS service could"
+                        "not be stopped, exiting.");
+            goto out;
+        }
+    }
+#endif
+
+    if (check_host_list()) {
+        runinit(&runner);
+        runner_add_args(&runner, GANESHA_PREFIX "/ganesha-ha.sh",
+                        "--setup-ganesha-conf-files", CONFDIR, "yes", NULL);
+        ret = runner_run(&runner);
+        if (ret) {
+            gf_asprintf(op_errstr,
+                        "creation of symlink ganesha.conf "
+                        "in /etc/ganesha failed");
+            goto out;
+        }
+        ret = manage_service("start");
+        if (ret)
+            gf_asprintf(op_errstr,
+                        "NFS-Ganesha failed to start."
+                        "Please see log file for details");
+    }
+
+out:
+    return ret;
+}
+
+static int
+pre_setup(gf_boolean_t run_setup, char **op_errstr)
+{
+    int ret = 0;
+    if (run_setup) {
+        if (!check_host_list()) {
+            gf_asprintf(op_errstr,
+                        "Running nfs-ganesha setup command "
+                        "from node which is not part of ganesha cluster");
+            return -1;
+        }
+    }
+    ret = setup_cluster(run_setup);
+    if (ret == -1)
+        gf_asprintf(op_errstr,
+                    "Failed to set up HA "
+                    "config for NFS-Ganesha. "
+                    "Please check the log file for details");
+    return ret;
+}
+
+int
+glusterd_handle_ganesha_op(dict_t *dict, char **op_errstr, char *key,
+                           char *value)
+{
+    int32_t ret = -1;
+    gf_boolean_t option = _gf_false;
+
+    GF_ASSERT(dict);
+    GF_ASSERT(op_errstr);
+    GF_ASSERT(key);
+    GF_ASSERT(value);
+
+    if (strcmp(key, "ganesha.enable") == 0) {
+        ret = ganesha_manage_export(dict, value, _gf_true, op_errstr);
+        if (ret < 0)
+            goto out;
+    }
+
+    /* It is possible that the key might not be set */
+    ret = gf_string2boolean(value, &option);
+    if (ret == -1) {
+        gf_asprintf(op_errstr, "Invalid value in key-value pair.");
+        goto out;
+    }
+
+    if (strcmp(key, GLUSTERD_STORE_KEY_GANESHA_GLOBAL) == 0) {
+        /* *
+         * The set up/teardown of pcs cluster should be performed only
+         * once. This will done on the node in which the cli command
+         * 'gluster nfs-ganesha <enable/disable>' got executed. So that
+         * node should part of ganesha HA cluster
+         */
+        if (option) {
+            ret = pre_setup(is_origin_glusterd(dict), op_errstr);
+            if (ret < 0)
+                goto out;
+        } else {
+            ret = teardown(is_origin_glusterd(dict), op_errstr);
+            if (ret < 0)
+                goto out;
+        }
+    }
+
+out:
+    return ret;
+}
diff --git a/xlators/mgmt/glusterd/src/glusterd-geo-rep.c b/xlators/mgmt/glusterd/src/glusterd-geo-rep.c
new file mode 100644
index 00000000000..bf062c87060
--- /dev/null
+++ b/xlators/mgmt/glusterd/src/glusterd-geo-rep.c
@@ -0,0 +1,6782 @@
+/*
+   Copyright (c) 2011-2012 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+#include <glusterfs/common-utils.h>
+#include "cli1-xdr.h"
+#include "xdr-generic.h"
+#include "glusterd.h"
+#include "glusterd-op-sm.h"
+#include "glusterd-geo-rep.h"
+#include "glusterd-store.h"
+#include "glusterd-utils.h"
+#include "glusterd-volgen.h"
+#include "glusterd-svc-helper.h"
+#include <glusterfs/run.h>
+#include <glusterfs/syscall.h>
+#include "glusterd-messages.h"
+
+#include <signal.h>
+
+static int
+dict_get_param(dict_t *dict, char *key, char **param);
+
+struct gsync_config_opt_vals_ gsync_confopt_vals[] = {
+    {
+        .op_name = "change_detector",
+        .no_of_pos_vals = 2,
+        .case_sensitive = _gf_true,
+        .values = {"xsync", "changelog"},
+    },
+    {.op_name = "special_sync_mode",
+     .no_of_pos_vals = 2,
+     .case_sensitive = _gf_true,
+     .values = {"partial", "recover"}},
+    {.op_name = "log-level",
+     .no_of_pos_vals = 5,
+     .case_sensitive = _gf_false,
+     .values = {"critical", "error", "warning", "info", "debug"}},
+    {.op_name = "use-tarssh",
+     .no_of_pos_vals = 6,
+     .case_sensitive = _gf_false,
+     .values = {"true", "false", "0", "1", "yes", "no"}},
+    {.op_name = "ignore_deletes",
+     .no_of_pos_vals = 6,
+     .case_sensitive = _gf_false,
+     .values = {"true", "false", "0", "1", "yes", "no"}},
+    {.op_name = "use_meta_volume",
+     .no_of_pos_vals = 6,
+     .case_sensitive = _gf_false,
+     .values = {"true", "false", "0", "1", "yes", "no"}},
+    {.op_name = "use-meta-volume",
+     .no_of_pos_vals = 6,
+     .case_sensitive = _gf_false,
+     .values = {"true", "false", "0", "1", "yes", "no"}},
+    {
+        .op_name = NULL,
+    },
+};
+
+static char *gsync_reserved_opts[] = {"gluster-command",
+                                      "pid-file",
+                                      "state-file",
+                                      "session-owner",
+                                      "state-socket-unencoded",
+                                      "socketdir",
+                                      "local-id",
+                                      "local-path",
+                                      "slave-id",
+                                      NULL};
+
+static char *gsync_no_restart_opts[] = {"checkpoint", "log_rsync_performance",
+                                        "log-rsync-performance", NULL};
+
+void
+set_gsyncd_inet6_arg(runner_t *runner)
+{
+    xlator_t *this = NULL;
+    char *af;
+    int ret;
+
+    this = THIS;
+    ret = dict_get_str(this->options, "transport.address-family", &af);
+    if (ret == 0)
+        runner_argprintf(runner, "--%s", af);
+}
+
+int
+__glusterd_handle_sys_exec(rpcsvc_request_t *req)
+{
+    int32_t ret = 0;
+    dict_t *dict = NULL;
+    gf_cli_req cli_req = {
+        {0},
+    };
+    glusterd_op_t cli_op = GD_OP_SYS_EXEC;
+    glusterd_conf_t *priv = NULL;
+    char *host_uuid = NULL;
+    char err_str[64] = {
+        0,
+    };
+    xlator_t *this = NULL;
+
+    GF_ASSERT(req);
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        req->rpc_err = GARBAGE_ARGS;
+        snprintf(err_str, sizeof(err_str), "Garbage args received");
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_GARBAGE_ARGS, NULL);
+        goto out;
+    }
+
+    if (cli_req.dict.dict_len) {
+        dict = dict_new();
+        if (!dict) {
+            gf_smsg(THIS->name, GF_LOG_ERROR, errno, GD_MSG_DICT_CREATE_FAIL,
+                    NULL);
+            goto out;
+        }
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to "
+                   "unserialize req-buffer to dictionary");
+            snprintf(err_str, sizeof(err_str),
+                     "Unable to decode "
+                     "the command");
+            goto out;
+        } else {
+            dict->extra_stdfree = cli_req.dict.dict_val;
+        }
+
+        host_uuid = gf_strdup(uuid_utoa(MY_UUID));
+        if (host_uuid == NULL) {
+            snprintf(err_str, sizeof(err_str),
+                     "Failed to get "
+                     "the uuid of local glusterd");
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_UUID_GET_FAIL,
+                    NULL);
+            ret = -1;
+            goto out;
+        }
+
+        ret = dict_set_dynstr(dict, "host-uuid", host_uuid);
+        if (ret) {
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_SET_FAILED,
+                    "Key=host-uuid", NULL);
+            goto out;
+        }
+    }
+
+    ret = glusterd_op_begin_synctask(req, cli_op, dict);
+
+out:
+    if (ret) {
+        if (err_str[0] == '\0')
+            snprintf(err_str, sizeof(err_str), "Operation failed");
+        ret = glusterd_op_send_cli_response(cli_op, ret, 0, req, dict, err_str);
+    }
+    return ret;
+}
+
+int
+__glusterd_handle_copy_file(rpcsvc_request_t *req)
+{
+    int32_t ret = 0;
+    dict_t *dict = NULL;
+    gf_cli_req cli_req = {
+        {0},
+    };
+    glusterd_op_t cli_op = GD_OP_COPY_FILE;
+    glusterd_conf_t *priv = NULL;
+    char *host_uuid = NULL;
+    char err_str[64] = {
+        0,
+    };
+    xlator_t *this = NULL;
+
+    GF_ASSERT(req);
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        req->rpc_err = GARBAGE_ARGS;
+        snprintf(err_str, sizeof(err_str), "Garbage args received");
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_GARBAGE_ARGS, NULL);
+        goto out;
+    }
+
+    if (cli_req.dict.dict_len) {
+        dict = dict_new();
+        if (!dict) {
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_CREATE_FAIL,
+                    NULL);
+            goto out;
+        }
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to"
+                   "unserialize req-buffer to dictionary");
+            snprintf(err_str, sizeof(err_str),
+                     "Unable to decode "
+                     "the command");
+            goto out;
+        } else {
+            dict->extra_stdfree = cli_req.dict.dict_val;
+        }
+
+        host_uuid = gf_strdup(uuid_utoa(MY_UUID));
+        if (host_uuid == NULL) {
+            snprintf(err_str, sizeof(err_str),
+                     "Failed to get "
+                     "the uuid of local glusterd");
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_UUID_GET_FAIL,
+                    NULL);
+            ret = -1;
+            goto out;
+        }
+
+        ret = dict_set_dynstr(dict, "host-uuid", host_uuid);
+        if (ret)
+            goto out;
+    }
+
+    ret = glusterd_op_begin_synctask(req, cli_op, dict);
+
+out:
+    if (ret) {
+        if (err_str[0] == '\0')
+            snprintf(err_str, sizeof(err_str), "Operation failed");
+        ret = glusterd_op_send_cli_response(cli_op, ret, 0, req, dict, err_str);
+    }
+    return ret;
+}
+
+int
+__glusterd_handle_gsync_set(rpcsvc_request_t *req)
+{
+    int32_t ret = 0;
+    dict_t *dict = NULL;
+    gf_cli_req cli_req = {
+        {0},
+    };
+    glusterd_op_t cli_op = GD_OP_GSYNC_SET;
+    char *master = NULL;
+    char *slave = NULL;
+    char operation[64] = {
+        0,
+    };
+    int type = 0;
+    glusterd_conf_t *priv = NULL;
+    char *host_uuid = NULL;
+    char err_str[64] = {
+        0,
+    };
+    xlator_t *this = NULL;
+
+    GF_ASSERT(req);
+
+    this = THIS;
+    GF_ASSERT(this);
+    priv = this->private;
+    GF_ASSERT(priv);
+
+    ret = xdr_to_generic(req->msg[0], &cli_req, (xdrproc_t)xdr_gf_cli_req);
+    if (ret < 0) {
+        req->rpc_err = GARBAGE_ARGS;
+        snprintf(err_str, sizeof(err_str), "Garbage args received");
+        gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_GARBAGE_ARGS, NULL);
+        goto out;
+    }
+
+    if (cli_req.dict.dict_len) {
+        dict = dict_new();
+        if (!dict) {
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_DICT_CREATE_FAIL,
+                    NULL);
+            goto out;
+        }
+
+        ret = dict_unserialize(cli_req.dict.dict_val, cli_req.dict.dict_len,
+                               &dict);
+        if (ret < 0) {
+            gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_UNSERIALIZE_FAIL,
+                   "failed to "
+                   "unserialize req-buffer to dictionary");
+            snprintf(err_str, sizeof(err_str),
+                     "Unable to decode "
+                     "the command");
+            goto out;
+        } else {
+            dict->extra_stdfree = cli_req.dict.dict_val;
+        }
+
+        host_uuid = gf_strdup(uuid_utoa(MY_UUID));
+        if (host_uuid == NULL) {
+            snprintf(err_str, sizeof(err_str),
+                     "Failed to get "
+                     "the uuid of local glusterd");
+            gf_smsg(this->name, GF_LOG_ERROR, errno, GD_MSG_UUID_GET_FAIL,
+                    NULL);
+            ret = -1;
+            goto out;
+        }
+        ret = dict_set_dynstr(dict, "host-uuid", host_uuid);
+        if (ret)
+            goto out;
+    }
+
+    ret = dict_get_str(dict, "master", &master);
+    if (ret < 0) {
+        gf_msg(this->name, GF_LOG_INFO, 0, GD_MSG_DICT_GET_FAILED,
+               "master not found, while handling " GEOREP " options");
+        master = "(No Master)";
+    }
+
+    ret = dict_get_str(dict, "slave", &slave);
+    if (ret < 0) {
+        gf_msg(this->name, GF_LOG_INFO, 0, GD_MSG_DICT_GET_FAILED,
+               "slave not found, while handling " GEOREP " options");
+        slave = "(No Slave)";
+    }
+
+    ret = dict_get_int32(dict, "type", &type);
+    if (ret < 0) {
+        snprintf(err_str, sizeof(err_str),
+                 "Command type not found "
+                 "while handling " GEOREP " options");
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_DICT_GET_FAILED, "%s",
+               err_str);
+        goto out;
+    }
+
+    switch (type) {
+        case GF_GSYNC_OPTION_TYPE_CREATE:
+            snprintf(operation, sizeof(operation), "create");
+            cli_op = GD_OP_GSYNC_CREATE;
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_START:
+            snprintf(operation, sizeof(operation), "start");
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_STOP:
+            snprintf(operation, sizeof(operation), "stop");
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_PAUSE:
+            snprintf(operation, sizeof(operation), "pause");
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_RESUME:
+            snprintf(operation, sizeof(operation), "resume");
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_CONFIG:
+            snprintf(operation, sizeof(operation), "config");
+            break;
+
+        case GF_GSYNC_OPTION_TYPE_STATUS:
+            snprintf(operation, sizeof(operation), "status");
+            break;
+    }
+
+    ret = glusterd_op_begin_synctask(req, cli_op, dict);
+
+out:
+    if (ret) {
+        if (err_str[0] == '\0')
+            snprintf(err_str, sizeof(err_str), "Operation failed");
+        ret = glusterd_op_send_cli_response(cli_op, ret, 0, req, dict, err_str);
+    }
+    return ret;
+}
+
+int
+glusterd_handle_sys_exec(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_sys_exec);
+}
+
+int
+glusterd_handle_copy_file(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_copy_file);
+}
+
+int
+glusterd_handle_gsync_set(rpcsvc_request_t *req)
+{
+    return glusterd_big_locked_handler(req, __glusterd_handle_gsync_set);
+}
+
+/*****
+ *
+ * glusterd_urltransform* internal API
+ *
+ *****/
+
+static void
+glusterd_urltransform_init(runner_t *runner, const char *transname)
+{
+    runinit(runner);
+    runner_add_arg(runner, GSYNCD_PREFIX "/gsyncd");
+    set_gsyncd_inet6_arg(runner);
+    runner_argprintf(runner, "--%s-url", transname);
+}
+
+static void
+glusterd_urltransform_add(runner_t *runner, const char *url)
+{
+    runner_add_arg(runner, url);
+}
+
+/* Helper routine to terminate just before slave_voluuid */
+static int32_t
+parse_slave_url(char *slv_url, char **slave)
+{
+    char *tmp = NULL;
+    xlator_t *this = NULL;
+    int32_t ret = -1;
+
+    this = THIS;
+
+    /* slave format:
+     * master_node_uuid:ssh://slave_host::slave_vol:slave_voluuid */
+    *slave = strchr(slv_url, ':');
+    if (!(*slave)) {
+        goto out;
+    }
+    (*slave)++;
+
+    /* To terminate at : before slave volume uuid */
+    tmp = strstr(*slave, "::");
+    if (!tmp) {
+        goto out;
+    }
+    tmp += 2;
+    tmp = strchr(tmp, ':');
+    if (!tmp)
+        gf_msg_debug(this->name, 0, "old slave: %s!", *slave);
+    else
+        *tmp = '\0';
+
+    ret = 0;
+    gf_msg_debug(this->name, 0, "parsed slave: %s!", *slave);
+out:
+    return ret;
+}
+
+static int
+_glusterd_urltransform_add_iter(dict_t *dict, char *key, data_t *value,
+                                void *data)
+{
+    runner_t *runner = (runner_t *)data;
+    char slv_url[VOLINFO_SLAVE_URL_MAX] = {0};
+    char *slave = NULL;
+    xlator_t *this = NULL;
+    int32_t ret = -1;
+
+    this = THIS;
+    GF_VALIDATE_OR_GOTO("glusterd", this, out);
+
+    gf_msg_debug(this->name, 0, "value->data %s", value->data);
+
+    if (snprintf(slv_url, sizeof(slv_url), "%s", value->data) >=
+        sizeof(slv_url)) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SLAVE_VOL_PARSE_FAIL,
+               "Error in copying slave: %s!", value->data);
+        goto out;
+    }
+
+    ret = parse_slave_url(slv_url, &slave);
+    if (ret == -1) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SLAVE_VOL_PARSE_FAIL,
+               "Error in parsing slave: %s!", value->data);
+        goto out;
+    }
+
+    runner_add_arg(runner, slave);
+    ret = 0;
+out:
+    return ret;
+}
+
+static void
+glusterd_urltransform_free(char **linearr, unsigned n)
+{
+    int i = 0;
+
+    for (; i < n; i++)
+        GF_FREE(linearr[i]);
+
+    GF_FREE(linearr);
+}
+
+static int
+glusterd_urltransform(runner_t *runner, char ***linearrp)
+{
+    char **linearr = NULL;
+    char *line = NULL;
+    unsigned arr_len = 32;
+    unsigned arr_idx = 0;
+    gf_boolean_t error = _gf_false;
+    xlator_t *this = NULL;
+
+    this = THIS;
+    GF_ASSERT(this);
+
+    linearr = GF_CALLOC(arr_len, sizeof(char *), gf_gld_mt_linearr);
+    if (!linearr) {
+        error = _gf_true;
+        goto out;
+    }
+
+    runner_redir(runner, STDOUT_FILENO, RUN_PIPE);
+    if (runner_start(runner) != 0) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_SPAWNING_CHILD_FAILED,
+               "spawning child failed");
+
+        error = _gf_true;
+        goto out;
+    }
+
+    arr_idx = 0;
+    for (;;) {
+        size_t len;
+        line = GF_MALLOC(1024, gf_gld_mt_linebuf);
+        if (!line) {
+            error = _gf_true;
+            goto out;
+        }
+
+        if (fgets(line, 1024, runner_chio(runner, STDOUT_FILENO)) == NULL) {
+            GF_FREE(line);
+            break;
+        }
+
+        len = strlen(line);
+        if (len == 0 || line[len - 1] != '\n') {
+            GF_FREE(line);
+            error = _gf_true;
+            goto out;
+        }
+        line[len - 1] = '\0';
+
+        if (arr_idx == arr_len) {
+            void *p = linearr;
+            arr_len <<= 1;
+            p = GF_REALLOC(linearr, arr_len);
+            if (!p) {
+                GF_FREE(line);
+                error = _gf_true;
+                goto out;
+            }
+            linearr = p;
+        }
+        linearr[arr_idx] = line;
+
+        arr_idx++;
+    }
+
+out:
+
+    /* XXX chpid field is not exported by run API
+     * but runner_end() does not abort the invoked
+     * process (ie. it might block in waitpid(2))
+     * so we resort to a manual kill a the private field
+     */
+    if (error && runner->chpid > 0)
+        kill(runner->chpid, SIGKILL);
+
+    if (runner_end(runner) != 0)
+        error = _gf_true;
+
+    if (error) {
+        gf_msg(this->name, GF_LOG_ERROR, 0, GD_MSG_READ_CHILD_DATA_FAILED,
+               "reading data from child failed");
+        glusterd_urltransform_free(linearr, arr_idx);
+        return -1;
+    }
+
+    *linearrp = linearr;
+    return arr_idx;
+}
+
+static int
+glusterd_urltransform_single(const char *url, const char *transname,
+                             char ***linearrp)
+{
+    runner_t runner = {
+        0,
+    };
+
+    glusterd_urltransform_init(&runner, transname);
+    glusterd_urltransform_add(&runner, url);
+    return glusterd_urltransform(&runner, linearrp);
+}
+
+struct dictidxmark {
+    unsigned isrch;
+    unsigned ithis;
+    char *ikey;
+};
+
+struct slave_vol_config {
+    char old_slvhost[_POSIX_HOST_NAME_MAX + 1];
+    char old_slvuser[LOGIN_NAME_MAX];
+    unsigned old_slvidx;
+    char slave_voluuid[UUID_CANONICAL_FORM_LEN + 1];
+};
+
+static int
+_dict_mark_atindex(dict_t *dict, char *key, data_t *value, void *data)
+{
+    struct dictidxmark *dim = data;
+
+    if (dim->isrch == dim->ithis)
+        dim->ikey = key;
+
+    dim->ithis++;
+    return 0;
+}
+
+static char *
+dict_get_by_index(dict_t *dict, unsigned i)
+{
+    struct dictidxmark dim = {
+        0,
+    };