BD Backend: CLI commands to create/delete image

Cli commands  added to create/delete a LV device.

The following command creates lv in a given vg.
$ gluster bd create <volname>:<vgname>/<lvname> <size>

The following command deletes lv in a given vg.
$ gluster bd delete <volname>:<vgname>/<lvname>

BUG: 805138
Change-Id: Ie4e100eca14e2ee32cf2bb4dd064b17230d673bf
Signed-off-by: M. Mohan Kumar <mohan@in.ibm.com>
Reviewed-on: http://review.gluster.org/3718
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

17 files changed, 1047 insertions, 81 deletions

diff --git a/cli/src/Makefile.am b/cli/src/Makefile.am
index 393077688be..7054f1120c8 100644
--- a/cli/src/Makefile.am
+++ b/cli/src/Makefile.am
@@ -3,6 +3,9 @@ sbin_PROGRAMS = gluster
 gluster_SOURCES = cli.c registry.c input.c cli-cmd.c cli-rl.c \
 	 cli-cmd-volume.c cli-cmd-peer.c cli-rpc-ops.c cli-cmd-parser.c\
 	 cli-cmd-system.c cli-cmd-misc.c cli-xml-output.c
+if ENABLE_BD_XLATOR
+gluster_SOURCES += cli-cmd-volume-bdevice.c
+endif
 
 gluster_LDADD = $(top_builddir)/libglusterfs/src/libglusterfs.la $(GF_LDADD)\
 		$(RLLIBS) $(top_builddir)/rpc/xdr/src/libgfxdr.la \
diff --git a/cli/src/cli-cmd-misc.c b/cli/src/cli-cmd-misc.c
index f3ef12147eb..5f3a77ca67f 100644
--- a/cli/src/cli-cmd-misc.c
+++ b/cli/src/cli-cmd-misc.c
@@ -31,6 +31,7 @@ extern struct cli_cmd volume_cmds[];
 extern struct cli_cmd cli_probe_cmds[];
 extern struct cli_cmd cli_log_cmds[];
 extern struct cli_cmd cli_system_cmds[];
+extern struct cli_cmd cli_bd_cmds[];
 struct cli_cmd cli_misc_cmds[];
 
 int
@@ -45,7 +46,11 @@ cli_cmd_display_help (struct cli_state *state, struct cli_cmd_word *in_word,
                       const char **words, int wordcount)
 {
         struct cli_cmd        *cmd[] = {volume_cmds, cli_probe_cmds,
-                                       cli_misc_cmds, NULL};
+                                       cli_misc_cmds,
+#ifdef HAVE_BD_XLATOR
+                                       cli_bd_cmds,
+#endif
+                                       NULL};
         struct cli_cmd        *cmd_ind = NULL;
         int                   i = 0;
 
diff --git a/cli/src/cli-cmd-volume-bdevice.c b/cli/src/cli-cmd-volume-bdevice.c
new file mode 100644
index 00000000000..ea7edab6502
--- /dev/null
+++ b/cli/src/cli-cmd-volume-bdevice.c
@@ -0,0 +1,225 @@
+/*
+  CLI for BD translator
+
+  Copyright IBM, Corp. 2012
+
+  This file is part of GlusterFS.
+
+  Author:
+  M. Mohan Kumar <mohan@in.ibm.com>
+
+  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 _CONFIG_H
+#define _CONFIG_H
+#include "config.h"
+#endif
+
+#include "cli.h"
+#include "cli-cmd.h"
+#include <string.h>
+
+extern rpc_clnt_prog_t *cli_rpc_prog;
+
+int
+cli_cmd_bd_help_cbk (struct cli_state *state, struct cli_cmd_word *in_word,
+                      const char **words, int wordcount);
+
+int32_t
+cli_cmd_bd_parse (dict_t *dict, const char **words)
+{
+        char          *volname  = NULL;
+        char          *buff     = NULL;
+        char          *buffp    = NULL;
+        int            ret      = -1;
+        char          *save     = NULL;
+        char          *path     = NULL;
+        char          *size     = NULL;
+        char          *eptr     = NULL;
+        gf_xl_bd_op_t  bd_op    = GF_BD_OP_INVALID;
+
+        /* volname:/path */
+        if (!strchr (words[2], ':') || !strchr (words[2], '/')) {
+                cli_out ("invalid parameter %s, needs <volname:/path>",
+                                words[2]);
+                return -1;
+        }
+        buff = buffp = gf_strdup (words[2]);
+        volname = strtok_r (buff, ":", &save);
+        path = strtok_r (NULL, ":", &save);
+
+        ret = dict_set_dynstr (dict, "volname", gf_strdup (volname));
+        if (ret)
+                goto out;
+
+        ret = dict_set_dynstr (dict, "path", gf_strdup (path));
+        if (ret)
+                goto out;
+
+        if (!strcasecmp (words[1], "create"))
+                bd_op = GF_BD_OP_NEW_BD;
+        else if (!strcasecmp (words[1], "delete"))
+                bd_op = GF_BD_OP_DELETE_BD;
+        else
+                return -1;
+
+        ret = dict_set_int32 (dict, "bd-op", bd_op);
+        if (ret)
+                goto out;
+
+        if (bd_op == GF_BD_OP_NEW_BD) {
+                /* If no suffix given we will treat it as MB */
+                strtoull (words[3], &eptr, 0);
+                /* no suffix */
+                if (!eptr[0])
+                        gf_asprintf (&size, "%sMB", words[3]);
+                else
+                        size = gf_strdup (words[3]);
+
+                ret = dict_set_dynstr (dict, "size", size);
+                if (ret)
+                        goto out;
+        }
+
+        ret = 0;
+out:
+        GF_FREE (buffp);
+        return ret;
+}
+
+/*
+ * bd create <volname>:/path <size>
+ * bd delete <volname>:/path
+ */
+int32_t
+cli_cmd_bd_validate (const char **words, int wordcount, dict_t **options)
+{
+        dict_t  *dict = NULL;
+        int     ret   = -1;
+        char    *op[] = { "create", "delete", NULL };
+        int     index = 0;
+
+        for (index = 0; op[index]; index++)
+                if (!strcasecmp (words[1], op[index]))
+                        break;
+
+        if (!op[index])
+                return -1;
+
+        dict = dict_new ();
+        if (!dict)
+                goto out;
+
+        if (!strcasecmp (words[1], "create")) {
+                if (wordcount != 4)
+                        goto out;
+        } else if (!strcasecmp (words[1], "delete")) {
+                if (wordcount != 3)
+                        goto out;
+        } else {
+                ret = -1;
+                goto out;
+        }
+
+        ret = cli_cmd_bd_parse (dict, words);
+        if (ret < 0)
+                goto out;
+
+        *options = dict;
+        ret = 0;
+out:
+        if (ret)
+                dict_unref (dict);
+        return ret;
+}
+
+int
+cli_cmd_bd_cbk (struct cli_state *state, struct cli_cmd_word *word,
+                         const char **words, int wordcount)
+{
+        int                     ret         = -1;
+        rpc_clnt_procedure_t    *proc       = NULL;
+        call_frame_t            *frame      = NULL;
+        int                     sent        = 0;
+        int                     parse_error = 0;
+        dict_t                  *options    = NULL;
+        cli_local_t             *local = NULL;
+
+        proc = &cli_rpc_prog->proctable[GLUSTER_CLI_BD_OP];
+
+        frame = create_frame (THIS, THIS->ctx->pool);
+        if (!frame)
+                goto out;
+
+        ret = cli_cmd_bd_validate (words, wordcount, &options);
+        if (ret) {
+                cli_usage_out (word->pattern);
+                parse_error = 1;
+                goto out;
+        }
+
+        CLI_LOCAL_INIT (local, words, frame, options);
+        if (proc->fn) {
+                ret = proc->fn (frame, THIS, options);
+        }
+
+out:
+        if (options)
+                dict_unref (options);
+
+        if (ret) {
+                cli_cmd_sent_status_get (&sent);
+                if ((sent == 0) && (parse_error == 0))
+                        cli_out ("BD op failed!");
+        }
+
+        CLI_STACK_DESTROY (frame);
+
+        return ret;
+}
+
+struct cli_cmd cli_bd_cmds[] = {
+        { "bd help",
+           cli_cmd_bd_help_cbk,
+           "display help for bd command"},
+        { "bd create <volname>:<bd> <size>",
+          cli_cmd_bd_cbk,
+          "\n\tcreate a block device where size can be "
+          "suffixed with KB, MB etc. Default size is in MB"},
+        { "bd delete <volname>:<bd>",
+          cli_cmd_bd_cbk,
+          "Delete a block device"},
+        { NULL, NULL, NULL }
+};
+
+int
+cli_cmd_bd_help_cbk (struct cli_state *state, struct cli_cmd_word *in_word,
+                      const char **words, int wordcount)
+{
+        struct cli_cmd *cmd = NULL;
+
+        for (cmd = cli_bd_cmds; cmd->pattern; cmd++)
+                if (_gf_false == cmd->disable)
+                        cli_out ("%s - %s", cmd->pattern, cmd->desc);
+
+        return 0;
+}
+
+int
+cli_cmd_bd_register (struct cli_state *state)
+{
+        int  ret = 0;
+        struct cli_cmd *cmd = NULL;
+
+        for (cmd = cli_bd_cmds; cmd->pattern; cmd++) {
+                ret = cli_cmd_register (&state->tree, cmd);
+                if (ret)
+                        goto out;
+        }
+out:
+        return ret;
+}
diff --git a/cli/src/cli-cmd.c b/cli/src/cli-cmd.c
index 64aba5d9faf..7a697603e15 100644
--- a/cli/src/cli-cmd.c
+++ b/cli/src/cli-cmd.c
@@ -231,6 +231,11 @@ cli_cmds_register (struct cli_state *state)
         if (ret)
                 goto out;
 
+#ifdef HAVE_BD_XLATOR
+        ret = cli_cmd_bd_register (state);
+        if (ret)
+                goto out;
+#endif
 out:
         return ret;
 }
diff --git a/cli/src/cli-cmd.h b/cli/src/cli-cmd.h
index a30a6cfe418..0ad37d49843 100644
--- a/cli/src/cli-cmd.h
+++ b/cli/src/cli-cmd.h
@@ -118,4 +118,8 @@ cli_cmd_submit (void *req, call_frame_t *frame,
 gf_answer_t
 cli_cmd_get_confirmation (struct cli_state *state, const char *question);
 int cli_cmd_sent_status_get (int *status);
+
+#ifdef HAVE_BD_XLATOR
+int cli_cmd_bd_register (struct cli_state *state);
+#endif
 #endif /* __CLI_CMD_H__ */
diff --git a/cli/src/cli-rpc-ops.c b/cli/src/cli-rpc-ops.c
index 77e15ee296c..9e4e03d0754 100644
--- a/cli/src/cli-rpc-ops.c
+++ b/cli/src/cli-rpc-ops.c
@@ -2601,6 +2601,134 @@ out:
         return ret;
 }
 
+#ifdef HAVE_BD_XLATOR
+int
+gf_cli_bd_op_cbk (struct rpc_req *req, struct iovec *iov,
+                             int count, void *myframe)
+{
+        gf_cli_rsp              rsp         = {0,};
+        int                     ret         = -1;
+        cli_local_t             *local      = NULL;
+        dict_t                  *dict       = NULL;
+        dict_t                  *input_dict = NULL;
+        gf_xl_bd_op_t           bd_op       = GF_BD_OP_INVALID;
+        char                    *operation  = NULL;
+        call_frame_t            *frame      = NULL;
+
+        if (-1 == req->rpc_status)
+                goto out;
+
+        ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_cli_rsp);
+        if (ret < 0) {
+                gf_log ("", GF_LOG_ERROR, "error");
+                goto out;
+        }
+
+        dict = dict_new ();
+        if (!dict) {
+                ret = -1;
+                goto out;
+        }
+
+        frame = myframe;
+        if (frame)
+                local = frame->local;
+
+        if (local) {
+                input_dict = local->dict;
+                ret = dict_get_int32 (input_dict, "bd-op",
+                                      (int32_t *)&bd_op);
+        }
+
+        switch (bd_op) {
+        case GF_BD_OP_NEW_BD:
+                operation = gf_strdup ("create");
+                break;
+        case GF_BD_OP_DELETE_BD:
+                operation = gf_strdup ("delete");
+                break;
+        default:
+                break;
+        }
+
+        ret = dict_unserialize (rsp.dict.dict_val, rsp.dict.dict_len, &dict);
+        if (ret)
+                goto out;
+
+        gf_log ("cli", GF_LOG_INFO, "Received resp to %s bd op", operation);
+
+        if (global_state->mode & GLUSTER_MODE_XML) {
+                ret = cli_xml_output_dict ("BdOp", dict, rsp.op_ret,
+                                           rsp.op_errno, rsp.op_errstr);
+                if (ret)
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "Error outputting to xml");
+                goto out;
+        }
+
+        if (rsp.op_ret && strcmp (rsp.op_errstr, ""))
+                cli_err ("%s", rsp.op_errstr);
+        else
+                cli_out ("BD %s has been %s", operation,
+                                (rsp.op_ret) ? "unsuccessful":
+                                "successful.");
+        ret = rsp.op_ret;
+
+out:
+        cli_cmd_broadcast_response (ret);
+
+        if (dict)
+                dict_unref (dict);
+
+        if (operation)
+                GF_FREE (operation);
+
+        if (rsp.dict.dict_val)
+                free (rsp.dict.dict_val);
+        if (rsp.op_errstr)
+                free (rsp.op_errstr);
+        return ret;
+}
+
+int32_t
+gf_cli_bd_op (call_frame_t *frame, xlator_t *this,
+                      void *data)
+{
+        gf_cli_req      req    = { {0,} };
+        int             ret    = 0;
+        dict_t          *dict  = NULL;
+
+        if (!frame || !this || !data) {
+                ret = -1;
+                goto out;
+        }
+
+        dict = dict_ref ((dict_t *)data);
+        if (!dict)
+                goto out;
+
+        ret = dict_allocate_and_serialize (dict,
+                                           &req.dict.dict_val,
+                                           &req.dict.dict_len);
+
+
+        ret = cli_cmd_submit (&req, frame, cli_rpc_prog,
+                              GLUSTER_CLI_BD_OP, NULL,
+                              this, gf_cli_bd_op_cbk,
+                              (xdrproc_t) xdr_gf_cli_req);
+
+out:
+        if (dict)
+                dict_unref (dict);
+
+        if (req.dict.dict_val)
+                GF_FREE (req.dict.dict_val);
+
+        gf_log ("cli", GF_LOG_DEBUG, "Returning %d", ret);
+        return ret;
+}
+#endif
+
 int32_t
 gf_cli_create_volume (call_frame_t *frame, xlator_t *this,
                          void *data)
@@ -6227,6 +6355,9 @@ struct rpc_clnt_procedure gluster_cli_actors[GLUSTER_CLI_MAXVALUE] = {
         [GLUSTER_CLI_STATEDUMP_VOLUME] = {"STATEDUMP_VOLUME", gf_cli_statedump_volume},
         [GLUSTER_CLI_LIST_VOLUME]      = {"LIST_VOLUME", gf_cli_list_volume},
         [GLUSTER_CLI_CLRLOCKS_VOLUME]  = {"CLEARLOCKS_VOLUME", gf_cli_clearlocks_volume},
+#ifdef HAVE_BD_XLATOR
+        [GLUSTER_CLI_BD_OP]            = {"BD_OP", gf_cli_bd_op},
+#endif
 };
 
 struct rpc_clnt_program cli_prog = {
diff --git a/cli/src/cli.h b/cli/src/cli.h
index a5f85ec8835..6e05e099645 100644
--- a/cli/src/cli.h
+++ b/cli/src/cli.h
@@ -37,6 +37,19 @@
 #define CLI_TAB_LENGTH                         8
 #define CLI_BRICK_STATUS_LINE_LEN             78
 
+#define CLI_LOCAL_INIT(local, words, frame, dictionary) \
+        do {                                                 \
+                local = cli_local_get ();                    \
+                                                             \
+                if (local) {                                 \
+                        local->words = words;                \
+                        if (dictionary)                      \
+                                local->dict = dictionary;    \
+                        if (frame)                           \
+                                frame->local = local;        \
+                }                                            \
+        } while (0)
+
 enum argp_option_keys {
 	ARGP_DEBUG_KEY = 133,
 	ARGP_PORT_KEY = 'p',
diff --git a/glusterfsd/src/glusterfsd-mgmt.c b/glusterfsd/src/glusterfsd-mgmt.c
index d1742d61fb4..d2a91ed6897 100644
--- a/glusterfsd/src/glusterfsd-mgmt.c
+++ b/glusterfsd/src/glusterfsd-mgmt.c
@@ -1189,6 +1189,75 @@ out:
 }
 
 int
+glusterfs_handle_bd_op (void *data)
+{
+        int32_t                  ret        = -1;
+        gd1_mgmt_brick_op_req    xlator_req = {0,};
+        dict_t                   *input     = NULL;
+        xlator_t                 *xlator    = NULL;
+        xlator_t                 *any       = NULL;
+        dict_t                   *output    = NULL;
+        char                     *xname     = NULL;
+        glusterfs_ctx_t          *ctx       = NULL;
+        glusterfs_graph_t        *active    = NULL;
+        xlator_t                 *this      = NULL;
+        rpcsvc_request_t         *req       = data;
+        char                     *error     = NULL;
+
+        GF_ASSERT (req);
+        this = THIS;
+        GF_ASSERT (this);
+
+        if (!xdr_to_generic (req->msg[0], &xlator_req,
+                             (xdrproc_t)xdr_gd1_mgmt_brick_op_req)) {
+                /* failed to decode msg */
+                req->rpc_err = GARBAGE_ARGS;
+                goto out;
+        }
+
+        ctx = glusterfsd_ctx;
+        active = ctx->active;
+        any = active->first;
+        input = dict_new ();
+        ret = dict_unserialize (xlator_req.input.input_val,
+                                xlator_req.input.input_len,
+                                &input);
+        if (ret < 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to "
+                        "unserialize req-buffer to dictionary");
+                goto out;
+        } else {
+                input->extra_stdfree = xlator_req.input.input_val;
+        }
+
+        /* FIXME, hardcoded */
+        xlator = xlator_search_by_xl_type (any, "storage/bd_map");
+        if (!xlator) {
+                        gf_log (this->name, GF_LOG_ERROR, "xlator %s is not "
+                                "loaded", xname);
+                        goto out;
+        }
+        output = dict_new ();
+        XLATOR_NOTIFY (xlator, GF_EVENT_TRANSLATOR_OP, input, output);
+out:
+        if (ret < 0) {
+                int retval;
+                retval = dict_get_str (output, "error", &error);
+        }
+        glusterfs_xlator_op_response_send (req, ret, error, output);
+        if (input)
+                dict_unref (input);
+        if (output)
+                dict_unref (output);
+        if (xlator_req.name)
+                /* malloced by xdr */
+                free (xlator_req.name);
+
+        return 0;
+}
+
+int
 glusterfs_handle_rpc_msg (rpcsvc_request_t *req)
 {
         int             ret = -1;
@@ -1222,6 +1291,17 @@ glusterfs_handle_rpc_msg (rpcsvc_request_t *req)
                 break;
         case GLUSTERD_NODE_STATUS:
                 ret = glusterfs_handle_node_status (req);
+                break;
+#ifdef HAVE_BD_XLATOR
+        case GLUSTERD_BRICK_BD_OP:
+                frame = create_frame (this, this->ctx->pool);
+                if (!frame)
+                        goto out;
+                ret = synctask_new (this->ctx->env,
+                                    glusterfs_handle_bd_op,
+                                    glusterfs_command_done, frame, req);
+                break;
+#endif
         default:
                 break;
         }
@@ -1284,7 +1364,10 @@ rpcsvc_actor_t glusterfs_actors[] = {
         [GLUSTERD_BRICK_STATUS] = {"STATUS", GLUSTERD_BRICK_STATUS, glusterfs_handle_rpc_msg, NULL, 0},
         [GLUSTERD_BRICK_XLATOR_DEFRAG] = { "TRANSLATOR DEFRAG", GLUSTERD_BRICK_XLATOR_DEFRAG, glusterfs_handle_rpc_msg, NULL, 0},
         [GLUSTERD_NODE_PROFILE] = {"NFS PROFILE", GLUSTERD_NODE_PROFILE, glusterfs_handle_rpc_msg, NULL, 0},
-        [GLUSTERD_NODE_STATUS] = {"NFS STATUS", GLUSTERD_NODE_STATUS, glusterfs_handle_rpc_msg, NULL, 0}
+        [GLUSTERD_NODE_STATUS] = {"NFS STATUS", GLUSTERD_NODE_STATUS, glusterfs_handle_rpc_msg, NULL, 0},
+#ifdef HAVE_BD_XLATOR
+        [GLUSTERD_BRICK_BD_OP] = {"BD OP", GLUSTERD_BRICK_BD_OP, glusterfs_handle_rpc_msg, NULL, 0}
+#endif
 };
 
 struct rpcsvc_program glusterfs_mop_prog = {
diff --git a/libglusterfs/src/xlator.c b/libglusterfs/src/xlator.c
index 6db27e9a4c4..b42bc3bcc83 100644
--- a/libglusterfs/src/xlator.c
+++ b/libglusterfs/src/xlator.c
@@ -344,6 +344,28 @@ out:
         return search;
 }
 
+xlator_t *
+xlator_search_by_xl_type (xlator_t *any, const char *type)
+{
+        xlator_t *search = NULL;
+
+        GF_VALIDATE_OR_GOTO ("xlator", any, out);
+        GF_VALIDATE_OR_GOTO ("xlator", type, out);
+
+        search = any;
+
+        while (search->prev)
+                search = search->prev;
+
+        while (search) {
+                if (!strcmp (search->type, type))
+                        break;
+                search = search->next;
+        }
+
+out:
+        return search;
+}
 
 static int
 __xlator_init(xlator_t *xl)
diff --git a/libglusterfs/src/xlator.h b/libglusterfs/src/xlator.h
index ec58dc170f1..1e21b63c55d 100644
--- a/libglusterfs/src/xlator.h
+++ b/libglusterfs/src/xlator.h
@@ -865,6 +865,7 @@ void xlator_foreach (xlator_t *this,
                      void *data);
 
 xlator_t *xlator_search_by_name (xlator_t *any, const char *name);
+xlator_t *xlator_search_by_xl_type (xlator_t *any, const char *type);
 
 void inode_destroy_notify (inode_t *inode, const char *xlname);
 
diff --git a/rpc/rpc-lib/src/protocol-common.h b/rpc/rpc-lib/src/protocol-common.h
index e4a783a7fbb..38528bd5f97 100644
--- a/rpc/rpc-lib/src/protocol-common.h
+++ b/rpc/rpc-lib/src/protocol-common.h
@@ -154,6 +154,7 @@ enum gluster_cli_procnum {
         GLUSTER_CLI_LIST_VOLUME,
         GLUSTER_CLI_CLRLOCKS_VOLUME,
         GLUSTER_CLI_UUID_RESET,
+        GLUSTER_CLI_BD_OP,
         GLUSTER_CLI_MAXVALUE,
 };
 
@@ -185,6 +186,7 @@ enum glusterd_brick_procnum {
         GLUSTERD_BRICK_XLATOR_DEFRAG,
         GLUSTERD_NODE_PROFILE,
         GLUSTERD_NODE_STATUS,
+        GLUSTERD_BRICK_BD_OP,
         GLUSTERD_BRICK_MAXVALUE,
 };
 
@@ -205,6 +207,12 @@ typedef enum {
         GF_AFR_OP_SPLIT_BRAIN_FILES
 } gf_xl_afr_op_t ;
 
+typedef enum {
+        GF_BD_OP_INVALID,
+        GF_BD_OP_NEW_BD,
+        GF_BD_OP_DELETE_BD,
+} gf_xl_bd_op_t ;
+
 #define GLUSTER_HNDSK_PROGRAM    14398633 /* Completely random */
 #define GLUSTER_HNDSK_VERSION    2   /* 0.0.2 */
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-handler.c b/xlators/mgmt/glusterd/src/glusterd-handler.c
index 9433436d0ae..1b1d113b578 100644
--- a/xlators/mgmt/glusterd/src/glusterd-handler.c
+++ b/xlators/mgmt/glusterd/src/glusterd-handler.c
@@ -50,6 +50,10 @@
 #include "globals.h"
 #include "glusterd-syncop.h"
 
+#ifdef HAVE_BD_XLATOR
+#include <lvm2app.h>
+#endif
+
 static int
 glusterd_handle_friend_req (rpcsvc_request_t *req, uuid_t  uuid,
                             char *hostname, int port,
@@ -927,6 +931,73 @@ out:
         return ret;
 }
 
+#ifdef HAVE_BD_XLATOR
+int
+glusterd_handle_cli_bd_op (rpcsvc_request_t *req)
+{
+        int32_t          ret        = -1;
+        gf_cli_req       cli_req    = { {0,} };
+        dict_t           *dict      = NULL;
+        char             *volname   = NULL;
+        char             *op_errstr = NULL;
+        glusterd_op_t    cli_op     = GD_OP_BD_OP;
+
+        GF_ASSERT (req);
+
+        if (!xdr_to_generic (req->msg[0], &cli_req,
+                                (xdrproc_t)xdr_gf_cli_req)) {
+                /* failed to decode msg */
+                req->rpc_err = GARBAGE_ARGS;
+                goto out;
+        }
+
+        gf_log ("glusterd", GF_LOG_DEBUG, "Received bd op 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_log ("glusterd", GF_LOG_ERROR,
+                                "failed to "
+                                "unserialize req-buffer to dictionary");
+                        goto out;
+                } else {
+                        dict->extra_stdfree = cli_req.dict.dict_val;
+                }
+        }
+
+        ret = dict_get_str (dict, "volname", &volname);
+        if (ret) {
+                gf_log (THIS->name, GF_LOG_ERROR,
+                                "failed to get volname");
+                goto out;
+        }
+
+        ret = glusterd_op_begin (req, GD_OP_BD_OP, dict);
+        gf_cmd_log ("bd op: %s", ((ret == 0) ? "SUCCESS": "FAILED"));
+out:
+        if (ret && dict)
+                dict_unref (dict);
+
+        glusterd_friend_sm ();
+        glusterd_op_sm ();
+
+        if (ret) {
+                if (!op_errstr)
+                        op_errstr = gf_strdup ("operation failed");
+                ret = glusterd_op_send_cli_response (cli_op, ret, 0,
+                                req, NULL, op_errstr);
+                GF_FREE (op_errstr);
+        }
+
+        return ret;
+}
+#endif
+
 int
 glusterd_handle_cli_uuid_reset (rpcsvc_request_t *req)
 {
@@ -3165,6 +3236,9 @@ rpcsvc_actor_t gd_svc_cli_actors[] = {
         [GLUSTER_CLI_STATEDUMP_VOLUME] = {"STATEDUMP_VOLUME", GLUSTER_CLI_STATEDUMP_VOLUME, glusterd_handle_cli_statedump_volume, NULL, 0},
         [GLUSTER_CLI_LIST_VOLUME] = {"LIST_VOLUME", GLUSTER_CLI_LIST_VOLUME, glusterd_handle_cli_list_volume, NULL, 0},
         [GLUSTER_CLI_CLRLOCKS_VOLUME] = {"CLEARLOCKS_VOLUME", GLUSTER_CLI_CLRLOCKS_VOLUME, glusterd_handle_cli_clearlocks_volume, NULL, 0},
+#ifdef HAVE_BD_XLATOR
+        [GLUSTER_CLI_BD_OP]       = {"BD_OP", GLUSTER_CLI_BD_OP, glusterd_handle_cli_bd_op, NULL, 0},
+#endif
 };
 
 struct rpcsvc_program gd_svc_cli_prog = {
diff --git a/xlators/mgmt/glusterd/src/glusterd-op-sm.c b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
index 4ca70516163..ac53f2e463b 100644
--- a/xlators/mgmt/glusterd/src/glusterd-op-sm.c
+++ b/xlators/mgmt/glusterd/src/glusterd-op-sm.c
@@ -235,6 +235,20 @@ glusterd_brick_op_build_payload (glusterd_op_t op, glusterd_brickinfo_t *brickin
                 brick_req->name = gf_strdup (name);
 
                 break;
+
+#ifdef HAVE_BD_XLATOR
+        case GD_OP_BD_OP:
+        {
+                brick_req = GF_CALLOC (1, sizeof (*brick_req),
+                                       gf_gld_mt_mop_brick_req_t);
+                if (!brick_req)
+                        goto out;
+
+                brick_req->op = GLUSTERD_BRICK_BD_OP;
+                brick_req->name = "";
+        }
+                break;
+#endif
         default:
                 goto out;
         break;
@@ -2356,6 +2370,9 @@ glusterd_op_build_payload (dict_t **req, char **op_errstr, dict_t *op_ctx)
                 case GD_OP_STATEDUMP_VOLUME:
                 case GD_OP_CLEARLOCKS_VOLUME:
                 case GD_OP_DEFRAG_BRICK_VOLUME:
+#ifdef HAVE_BD_XLATOR
+                case GD_OP_BD_OP:
+#endif
                         {
                                 ret = dict_get_str (dict, "volname", &volname);
                                 if (ret) {
@@ -3489,7 +3506,11 @@ glusterd_op_stage_validate (glusterd_op_t op, dict_t *dict, char **op_errstr,
                         ret = glusterd_op_stage_clearlocks_volume (dict,
                                                                    op_errstr);
                         break;
-
+#ifdef HAVE_BD_XLATOR
+                case GD_OP_BD_OP:
+                        ret = glusterd_op_stage_bd (dict, op_errstr);
+                        break;
+#endif
                 default:
                         gf_log ("", GF_LOG_ERROR, "Unknown op %d",
                                 op);
@@ -3585,7 +3606,11 @@ glusterd_op_commit_perform (glusterd_op_t op, dict_t *dict, char **op_errstr,
                 case GD_OP_CLEARLOCKS_VOLUME:
                         ret = glusterd_op_clearlocks_volume (dict, op_errstr);
                         break;
-
+#ifdef HAVE_BD_XLATOR
+                case GD_OP_BD_OP:
+                        ret = 0;
+                        break;
+#endif
                 default:
                         gf_log ("", GF_LOG_ERROR, "Unknown op %d",
                                 op);
@@ -4325,6 +4350,62 @@ _select_rxlators_for_full_self_heal (xlator_t *this,
         return rxlator_count;
 }
 
+#ifdef HAVE_BD_XLATOR
+static int
+glusterd_bricks_select_bd (dict_t *dict, char **op_errstr)
+{
+        int                        ret           = -1;
+        glusterd_conf_t            *priv         = NULL;
+        xlator_t                   *this         = NULL;
+        glusterd_pending_node_t    *pending_node = NULL;
+        glusterd_volinfo_t         *volinfo      = NULL;
+        char                       *volname      = NULL;
+        glusterd_brickinfo_t       *brickinfo    = NULL;
+        int                         brick_index  = -1;
+
+        this = THIS;
+        GF_ASSERT (this);
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        ret = dict_get_str (dict, "volname", &volname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to get volname");
+                goto out;
+        }
+        ret = glusterd_volinfo_find (volname, &volinfo);
+        if (ret)
+                goto out;
+
+        pending_node = GF_CALLOC (1, sizeof (*pending_node),
+                                  gf_gld_mt_pending_node_t);
+        if (!pending_node) {
+                ret = -1;
+                goto out;
+        }
+
+        list_for_each_entry (brickinfo, &volinfo->bricks, brick_list) {
+                brick_index++;
+                if (uuid_compare (brickinfo->uuid, MY_UUID) ||
+                    !glusterd_is_brick_started (brickinfo)) {
+                        continue;
+                }
+                pending_node->node = brickinfo;
+                pending_node->type = GD_NODE_BRICK;
+                pending_node->index = brick_index;
+                list_add_tail (&pending_node->list,
+                               &opinfo.pending_bricks);
+                pending_node = NULL;
+        }
+
+        ret = 0;
+
+out:
+        gf_log (THIS->name, GF_LOG_DEBUG, "Returning ret %d", ret);
+        return ret;
+}
+#endif
+
 static int
 glusterd_bricks_select_heal_volume (dict_t *dict, char **op_errstr)
 {
@@ -4402,7 +4483,6 @@ out:
 
 }
 
-
 static int
 glusterd_bricks_select_rebalance_volume (dict_t *dict, char **op_errstr)
 {
@@ -4735,6 +4815,11 @@ glusterd_op_bricks_select (glusterd_op_t op, dict_t *dict, char **op_errstr)
         case GD_OP_DEFRAG_BRICK_VOLUME:
                 ret = glusterd_bricks_select_rebalance_volume (dict, op_errstr);
                 break;
+#ifdef HAVE_BD_XLATOR
+        case GD_OP_BD_OP:
+                ret = glusterd_bricks_select_bd (dict, op_errstr);
+                break;
+#endif
         default:
                 break;
          }
@@ -5298,6 +5383,9 @@ glusterd_op_free_ctx (glusterd_op_t op, void *ctx)
                 case GD_OP_STATEDUMP_VOLUME:
                 case GD_OP_CLEARLOCKS_VOLUME:
                 case GD_OP_DEFRAG_BRICK_VOLUME:
+#ifdef HAVE_BD_XLATOR
+                case GD_OP_BD_OP:
+#endif
                         dict_unref (ctx);
                         break;
                 default:
diff --git a/xlators/mgmt/glusterd/src/glusterd-rpc-ops.c b/xlators/mgmt/glusterd/src/glusterd-rpc-ops.c
index c18c2b5e19e..21fad7e93a2 100644
--- a/xlators/mgmt/glusterd/src/glusterd-rpc-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-rpc-ops.c
@@ -139,6 +139,7 @@ glusterd_op_send_cli_response (glusterd_op_t op, int32_t op_ret,
         case GD_OP_LIST_VOLUME:
         case GD_OP_CLEARLOCKS_VOLUME:
         case GD_OP_HEAL_VOLUME:
+        case GD_OP_BD_OP:
         {
                 /*nothing specific to be done*/
                 break;
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
index 8c76c8f09e1..b74bbec7c59 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-ops.c
@@ -1241,6 +1241,91 @@ out:
         return ret;
 }
 
+#ifdef HAVE_BD_XLATOR
+int
+glusterd_op_stage_bd (dict_t *dict, char **op_errstr)
+{
+        int                     ret       = -1;
+        char                    *volname  = NULL;
+        char                    *path     = NULL;
+        char                    *size     = NULL;
+        glusterd_volinfo_t      *volinfo  = NULL;
+        char                    msg[2048] = {0,};
+        gf_xl_bd_op_t           bd_op     = GF_BD_OP_INVALID;
+        uint64_t                bytes     = 0;
+
+        ret = dict_get_str (dict, "volname", &volname);
+        if (ret) {
+                snprintf (msg, sizeof(msg), "Failed to get volume name");
+                gf_log (THIS->name, GF_LOG_ERROR, "%s", msg);
+                *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        ret = dict_get_int32 (dict, "bd-op", (int32_t *)&bd_op);
+        if (ret) {
+                snprintf (msg, sizeof(msg), "Failed to get bd-op");
+                gf_log (THIS->name, GF_LOG_ERROR, "%s", msg);
+                *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "path", &path);
+        if (ret) {
+                snprintf (msg, sizeof(msg), "Failed to get path");
+                gf_log (THIS->name, GF_LOG_ERROR, "%s", msg);
+                *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        if (bd_op == GF_BD_OP_NEW_BD) {
+                ret = dict_get_str (dict, "size", &size);
+                if (ret) {
+                        snprintf (msg, sizeof(msg), "Failed to get size");
+                        gf_log ("", GF_LOG_ERROR, "%s", msg);
+                        *op_errstr = gf_strdup (msg);
+                        goto out;
+                }
+                if (gf_string2bytesize (size, &bytes) < 0) {
+                        snprintf (msg, sizeof(msg),
+                                  "Invalid size %s, suffix with KB, MB etc",
+                                   size);
+                        gf_log ("", GF_LOG_ERROR, "%s", msg);
+                        *op_errstr = gf_strdup (msg);
+                        ret = -1;
+                        goto out;
+                }
+        }
+
+        ret = glusterd_volinfo_find (volname, &volinfo);
+        if (ret) {
+                snprintf (msg, sizeof(msg), "Volume %s does not exist",
+                          volname);
+                gf_log ("", GF_LOG_ERROR, "%s", msg);
+                *op_errstr = gf_strdup (msg);
+                goto out;
+        }
+
+        ret = glusterd_validate_volume_id (dict, volinfo);
+        if (ret)
+                goto out;
+
+        if (!glusterd_is_volume_started (volinfo)) {
+                snprintf (msg, sizeof(msg), "Volume %s is not started",
+                          volname);
+                gf_log ("", GF_LOG_ERROR, "%s", msg);
+                *op_errstr = gf_strdup (msg);
+                ret = -1;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        gf_log ("", GF_LOG_DEBUG, "Returning %d", ret);
+        return ret;
+}
+
+#endif
 
 int
 glusterd_op_create_volume (dict_t *dict, char **op_errstr)
diff --git a/xlators/mgmt/glusterd/src/glusterd.h b/xlators/mgmt/glusterd/src/glusterd.h
index ea0ea6061d3..36cd8f8fc56 100644
--- a/xlators/mgmt/glusterd/src/glusterd.h
+++ b/xlators/mgmt/glusterd/src/glusterd.h
@@ -78,6 +78,7 @@ typedef enum glusterd_op_ {
         GD_OP_LIST_VOLUME,
         GD_OP_CLEARLOCKS_VOLUME,
         GD_OP_DEFRAG_BRICK_VOLUME,
+        GD_OP_BD_OP,
         GD_OP_MAX,
 } glusterd_op_t;
 
@@ -684,6 +685,7 @@ int glusterd_op_statedump_volume (dict_t *dict, char **op_errstr);
 
 int glusterd_op_stage_clearlocks_volume (dict_t *dict, char **op_errstr);
 int glusterd_op_clearlocks_volume (dict_t *dict, char **op_errstr);
+int glusterd_op_stage_bd (dict_t *dict, char **op_errstr);
 
 /* misc */
 void glusterd_do_replace_brick (void *data);
diff --git a/xlators/storage/bd_map/src/bd_map.c b/xlators/storage/bd_map/src/bd_map.c
index 2db5a13cd3d..a84ee29fba2 100644
--- a/xlators/storage/bd_map/src/bd_map.c
+++ b/xlators/storage/bd_map/src/bd_map.c
@@ -31,7 +31,7 @@
 #include "defaults.h"
 #include "glusterfs3-xdr.h"
 #include "run.h"
-
+#include "protocol-common.h"
 
 /* Regular fops */
 
@@ -156,6 +156,56 @@ out:
 }
 
 int32_t
+bd_delete_lv (bd_priv_t *priv, bd_entry_t *p_entry, bd_entry_t *lventry,
+              const char *path, int *op_errno)
+{
+        vg_t    vg       = NULL;
+        lv_t    lv       = NULL;
+        int     op_ret   = -1;
+
+        *op_errno = 0;
+        BD_WR_LOCK (&priv->lock);
+        vg = lvm_vg_open (priv->handle, p_entry->name, "w", 0);
+        if (!vg) {
+                *op_errno = ENOENT;
+                BD_UNLOCK (&priv->lock);
+                goto out;
+        }
+
+        lv = lvm_lv_from_name (vg, lventry->name);
+        if (!lv) {
+                lvm_vg_close (vg);
+                *op_errno = ENOENT;
+                BD_UNLOCK (&priv->lock);
+                goto out;
+        }
+        op_ret = lvm_vg_remove_lv (lv);
+        if (op_ret < 0) {
+                *op_errno = errno;
+                lvm_vg_close (vg);
+                BD_UNLOCK (&priv->lock);
+                goto out;
+        }
+        lvm_vg_close (vg);
+
+        op_ret = bd_entry_rm (path);
+        if (op_ret < 0) {
+                *op_errno = EIO;
+                BD_UNLOCK (&priv->lock);
+                goto out;
+        }
+        BD_ENTRY_UPDATE_MTIME (p_entry);
+
+        op_ret = 0;
+        op_errno = 0;
+
+        BD_UNLOCK (&priv->lock);
+        op_ret = 0;
+out:
+        return op_ret;
+}
+
+int32_t
 bd_unlink (call_frame_t *frame, xlator_t *this,
                 loc_t *loc, int xflag, dict_t *xdata)
 {
@@ -164,8 +214,6 @@ bd_unlink (call_frame_t *frame, xlator_t *this,
         struct iatt       preparent  = {0, };
         struct iatt       postparent = {0, };
         bd_priv_t         *priv      = NULL;
-        vg_t              vg         = NULL;
-        lv_t              lv         = NULL;
         bd_entry_t        *lventry   = NULL;
         bd_entry_t        *p_entry   = NULL;
         char              *vg_name   = NULL;
@@ -200,43 +248,8 @@ bd_unlink (call_frame_t *frame, xlator_t *this,
                 goto out;
 
         memcpy (&preparent, p_entry->attr, sizeof(preparent));
-
-        BD_WR_LOCK (&priv->lock);
-        vg = lvm_vg_open (priv->handle, p_entry->name, "w", 0);
-        if (!vg) {
-                op_errno = ENOENT;
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-
-        lv = lvm_lv_from_name (vg, lventry->name);
-        if (!lv) {
-                lvm_vg_close (vg);
-                op_errno = ENOENT;
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-        op_ret = lvm_vg_remove_lv (lv);
-        if (op_ret < 0) {
-                op_errno = errno;
-                lvm_vg_close (vg);
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-        lvm_vg_close (vg);
-        op_ret = bd_entry_rm (loc->path);
-        if (op_ret < 0) {
-                op_errno = EIO;
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-        BD_ENTRY_UPDATE_MTIME (p_entry);
+        op_ret = bd_delete_lv (priv, p_entry, lventry, loc->path, &op_errno);
         memcpy (&postparent, p_entry->attr, sizeof(postparent));
-        op_ret = 0;
-        op_errno = 0;
-
-        BD_UNLOCK (&priv->lock);
-
 out:
         if (p_entry)
                 BD_PUT_ENTRY (priv, p_entry);
@@ -970,6 +983,59 @@ err:
         return op_ret;
 }
 
+int bd_create_lv (bd_priv_t *priv, bd_entry_t *p_entry, const char *vg_name,
+                  const char *lv_name, char *size, mode_t mode)
+{
+        vg_t            vg       = NULL;
+        int             ret      = -1;
+        char            *path    = NULL;
+        struct iatt     iattr    = {0, };
+        bd_entry_t      *lventry = NULL;
+        uint64_t        extent   = 0;
+
+        BD_WR_LOCK (&priv->lock);
+        vg = lvm_vg_open (priv->handle, vg_name,  "w", 0);
+        if (!vg) {
+                ret = -1;
+                goto out;
+        }
+        extent = lvm_vg_get_extent_size (vg);
+        if (size)
+                gf_string2bytesize (size, &extent);
+
+        if (lvm_vg_create_lv_linear (vg, lv_name, extent) == NULL) {
+                ret = -EAGAIN;
+                lvm_vg_close (vg);
+                goto out;
+        }
+        lvm_vg_close (vg);
+
+        gf_asprintf (&path, "/dev/%s/%s", vg_name, lv_name);
+        if (!path) {
+                ret = -ENOMEM;
+                lvm_vg_close (vg);
+                goto out;
+        }
+        bd_entry_istat (path, &iattr, IA_IFREG);
+        iattr.ia_size = extent;
+        if (!mode)
+                mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
+
+        iattr.ia_type = ia_type_from_st_mode (mode);
+        iattr.ia_prot = ia_prot_from_st_mode (mode);
+        lventry = bd_entry_add (p_entry, lv_name, &iattr, IA_IFREG);
+        if (!lventry) {
+                ret = -EAGAIN;
+                goto out;
+        }
+        ret = 0;
+out:
+        BD_UNLOCK (&priv->lock);
+        if (path)
+                GF_FREE (path);
+        return ret;
+}
+
 int bd_create (call_frame_t *frame, xlator_t *this,
                 loc_t *loc, int32_t flags, mode_t mode,
                 mode_t umask, fd_t *fd, dict_t *params)
@@ -977,20 +1043,16 @@ int bd_create (call_frame_t *frame, xlator_t *this,
         int32_t            op_ret            = -1;
         int32_t            op_errno          = 0;
         int32_t            _fd               = -1;
-        uint64_t           extent            = 0;
         bd_priv_t          *priv             = NULL;
         struct iatt        stbuf             = {0, };
         struct iatt        preparent         = {0, };
         struct iatt        postparent        = {0, };
         bd_entry_t         *p_entry          = NULL;
         bd_entry_t         *lventry          = NULL;
-        vg_t               vg                = NULL;
-        lv_t               lv                = NULL;
         bd_fd_t            *pfd              = NULL;
         char               *vg_name          = NULL;
         char               *volume           = NULL;
         char               *path             = NULL;
-        struct iatt        iattr             = {0, };
 
         VALIDATE_OR_GOTO (frame, out);
         VALIDATE_OR_GOTO (this, out);
@@ -1024,39 +1086,10 @@ int bd_create (call_frame_t *frame, xlator_t *this,
 
         memcpy (&preparent, p_entry->attr, sizeof(preparent));
 
-        BD_WR_LOCK (&priv->lock);
-        vg = lvm_vg_open (priv->handle, p_entry->name, "w", 0);
-        if (!vg) {
-                op_errno = errno;
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-        extent = lvm_vg_get_extent_size (vg);
-        /* Create the LV */
-        lv = lvm_vg_create_lv_linear (vg, loc->name, extent);
-        if (!lv) {
-                op_errno = errno;
-                lvm_vg_close (vg);
-                BD_UNLOCK (&priv->lock);
-                goto out;
-        }
-        lvm_vg_close (vg);
-
-        gf_asprintf (&path, "/dev/%s/%s", p_entry->name, loc->name);
-        if (!path) {
-                op_errno = ENOMEM;
-                goto out;
-        }
-        bd_entry_istat (path, &iattr, IA_IFREG);
-        iattr.ia_size = extent;
-        iattr.ia_type = ia_type_from_st_mode (mode);
-        iattr.ia_prot = ia_prot_from_st_mode (mode);
-        lventry = bd_entry_add (p_entry, loc->name, &iattr, IA_IFREG);
-        if (!lventry) {
-                op_errno = EAGAIN;
+        op_errno = bd_create_lv (priv, p_entry, p_entry->name, loc->name, 0,
+                                 mode);
+        if (op_errno)
                 goto out;
-        }
-        BD_UNLOCK (&priv->lock);
 
         BD_ENTRY (priv, lventry, loc->path);
         if (!lventry) {
@@ -1069,6 +1102,11 @@ int bd_create (call_frame_t *frame, xlator_t *this,
         /* Mask O_CREATE since we created LV */
         flags &= ~(O_CREAT | O_EXCL);
 
+        gf_asprintf (&path, "/dev/%s/%s", p_entry->name, loc->name);
+        if (!path) {
+                op_errno = ENOMEM;
+                goto out;
+        }
         _fd = open (path, flags, 0);
         if (_fd == -1) {
                 op_errno = errno;
@@ -2070,6 +2108,166 @@ bd_fxattrop (call_frame_t *frame, xlator_t *this,
         return 0;
 }
 
+int bd_xl_op_create (bd_priv_t *priv, dict_t *input, dict_t *output)
+{
+        char            *vg      = NULL;
+        char            *lv      = NULL;
+        char            *path    = NULL;
+        bd_entry_t      *p_entry = NULL;
+        bd_entry_t      *lventry = NULL;
+        char            *size    = 0;
+        int             ret      = -1;
+        char            *error   = NULL;
+        int             retval   = -1;
+        char            *buff    = NULL;
+        char            *buffp   = NULL;
+        char            *save    = NULL;
+
+        ret = dict_get_str (input, "size", &size);
+        if (ret) {
+                gf_asprintf (&error, "no size specified");
+                goto out;
+        }
+        ret = dict_get_str (input, "path", &path);
+        if (ret) {
+                gf_asprintf (&error, "no path specified");
+                goto out;
+        }
+
+        buff = buffp = gf_strdup (path);
+
+        vg = strtok_r (buff, "/", &save);
+        lv = strtok_r (NULL, "/", &save);
+
+        if (!vg || !lv) {
+                gf_asprintf (&error, "invalid path %s", path);
+                ret = -1;
+                goto out;
+        }
+
+        BD_ENTRY (priv, p_entry, vg);
+        if (!p_entry) {
+                ret = -ENOENT;
+                goto out;
+        }
+        BD_ENTRY (priv, lventry, path);
+        if (lventry) {
+                ret = -EEXIST;
+                gf_asprintf (&error, "%s already exists", lv);
+                BD_PUT_ENTRY (priv, lventry);
+                goto out;
+        }
+
+        ret = bd_create_lv (priv, p_entry, vg, lv, size, 0);
+        if (ret < 0) {
+                gf_asprintf (&error, "bd_create_lv error %d", -ret);
+                goto out;
+        }
+        ret = 0;
+out:
+        if (p_entry)
+                BD_PUT_ENTRY (priv, p_entry);
+
+        if (buffp)
+                GF_FREE (buffp);
+
+        if (error)
+                retval = dict_set_dynstr (output, "error", error);
+        return ret;
+}
+
+int bd_xl_op_delete (bd_priv_t *priv, dict_t *input, dict_t *output)
+{
+        char            *vg      = NULL;
+        char            *path    = NULL;
+        bd_entry_t      *p_entry = NULL;
+        bd_entry_t      *lventry = NULL;
+        int             ret      = -1;
+        char            *error   = NULL;
+        int             retval   = -1;
+        char            *buff    = NULL;
+        char            *buffp   = NULL;
+        char            *save    = NULL;
+        int             op_errno = 0;
+
+        ret = dict_get_str (input, "path", &path);
+        if (ret) {
+                gf_asprintf (&error, "no path specified");
+                goto out;
+        }
+
+        buff = buffp = gf_strdup (path);
+
+        vg = strtok_r (buff, "/", &save);
+        if (!vg) {
+                gf_asprintf (&error, "invalid path %s", path);
+                op_errno = EINVAL;
+                ret = -1;
+                goto out;
+        }
+
+        BD_ENTRY (priv, p_entry, vg);
+        BD_ENTRY (priv, lventry, path);
+        if (!p_entry || !lventry) {
+                op_errno = -ENOENT;
+                gf_asprintf (&error, "%s not found", path);
+                ret = -1;
+                goto out;
+        }
+        ret = bd_delete_lv (priv, p_entry, lventry, path, &op_errno);
+        if (ret < 0) {
+                gf_asprintf (&error, "bd_delete_lv error, error:%d", op_errno);
+                goto out;
+        }
+        ret = 0;
+out:
+        if (p_entry)
+                BD_PUT_ENTRY (priv, p_entry);
+        if (lventry)
+                BD_PUT_ENTRY (priv, lventry);
+        if (buffp)
+                GF_FREE (buffp);
+        if (error)
+                retval = dict_set_dynstr (output, "error", error);
+        return ret;
+}
+
+int32_t
+bd_notify (xlator_t *this, dict_t *input, dict_t *output)
+{
+        int             ret      = -1;
+        int             retval   = -1;
+        int32_t         bdop     = -1;
+        bd_priv_t       *priv    = NULL;
+        char            *error   = NULL;
+
+        priv = this->private;
+        VALIDATE_OR_GOTO (priv, out);
+
+        ret = dict_get_int32 (input, "bd-op", (int32_t *)&bdop);
+        if (ret) {
+                asprintf (&error, "no sub-op specified");
+                goto out;
+        }
+
+        switch (bdop)
+        {
+        case GF_BD_OP_NEW_BD:
+                ret = bd_xl_op_create (priv, input, output);
+                break;
+        case GF_BD_OP_DELETE_BD:
+                ret = bd_xl_op_delete (priv, input, output);
+                break;
+        default:
+                gf_asprintf (&error, "invalid bd-op %d specified", bdop);
+                retval = dict_set_dynstr (output, "error", error);
+                goto out;
+        }
+
+out:
+        return ret;
+}
+
 /**
  * notify - when parent sends PARENT_UP, send CHILD_UP event from here
  */
@@ -2079,6 +2277,16 @@ notify (xlator_t *this,
         void *data,
         ...)
 {
+        va_list ap;
+        int     ret    = -1;
+        void    *data2 = NULL;
+        dict_t  *input = NULL;
+        dict_t  *output = NULL;
+
+        va_start (ap, data);
+        data2 = va_arg (ap, dict_t *);
+        va_end (ap);
+
         switch (event)
         {
         case GF_EVENT_PARENT_UP:
@@ -2087,10 +2295,18 @@ notify (xlator_t *this,
                 default_notify (this, GF_EVENT_CHILD_UP, data);
         }
         break;
+        case GF_EVENT_TRANSLATOR_OP:
+                input = data;
+                output = data2;
+                if (!output)
+                        output = dict_new ();
+                ret = bd_notify (this, input, output);
+                break;
+
         default:
                 break;
         }
-        return 0;
+        return ret;
 }
 
 int32_t