Snapshot/clone: clone of a snapshot that will act as a regular volume

snapshot clone will allow us to take a snpahot of a snapshot.
Newly created clone volume will be a regular volume with read/write
permissions.

CLI command
snapshot clone <clonename> <snapname>

Change-Id: Icadb993fa42fff787a330f8f49452da54e9db7de
BUG: 1199894
Signed-off-by: Mohammed Rafi KC <rkavunga@redhat.com>
Reviewed-on: http://review.gluster.org/9750
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Rajesh Joseph <rjoseph@redhat.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

7 files changed, 1074 insertions, 189 deletions

diff --git a/cli/src/cli-cmd-parser.c b/cli/src/cli-cmd-parser.c
index b2ef1d77104..aa512738784 100644
--- a/cli/src/cli-cmd-parser.c
+++ b/cli/src/cli-cmd-parser.c
@@ -3358,6 +3358,83 @@ out:
         return ret;
 }
 
+/* snapshot clone <clonename> <snapname>
+ * @arg-0, dict     : Request Dictionary to be sent to server side.
+ * @arg-1, words    : Contains individual words of CLI command.
+ * @arg-2, wordcount: Contains number of words present in the CLI command.
+ *
+ * return value : -1 on failure
+ *                 0 on success
+ */
+int
+cli_snap_clone_parse (dict_t *dict, const char **words, int wordcount) {
+        uint64_t        i               =       0;
+        int             ret             =       -1;
+        char            key[PATH_MAX]   =       "";
+        char            *clonename      =       NULL;
+        unsigned int    cmdi            =       2;
+        int             flags           =       0;
+        /* cmdi is command index, here cmdi is "2" (gluster snapshot clone)*/
+
+        GF_ASSERT (words);
+        GF_ASSERT (dict);
+
+        if (wordcount == cmdi + 1) {
+                cli_err ("Invalid Syntax.");
+                gf_log ("cli", GF_LOG_ERROR,
+                        "Invalid number of  words for snap clone command");
+                goto out;
+        }
+
+        if (strlen(words[cmdi]) >= GLUSTERD_MAX_SNAP_NAME) {
+                cli_err ("snapshot clone: failed: clonename cannot exceed "
+                         "255 characters.");
+                gf_log ("cli", GF_LOG_ERROR, "Clone name too long");
+
+                goto out;
+        }
+
+        clonename = (char *) words[cmdi];
+        for (i = 0 ; i < strlen (clonename); i++) {
+                /* Following volume name convention */
+                if (!isalnum (clonename[i]) && (clonename[i] != '_'
+                                           && (clonename[i] != '-'))) {
+                        /* TODO : Is this message enough?? */
+                        cli_err ("Clonename can contain only alphanumeric, "
+                                 "\"-\" and \"_\" characters");
+                        goto out;
+                }
+        }
+
+        ret = dict_set_int32 (dict, "volcount", 1);
+        if (ret) {
+                gf_log ("cli", GF_LOG_ERROR, "Could not save volcount");
+                goto out;
+        }
+
+        ret = dict_set_str (dict, "clonename", (char *)words[cmdi]);
+        if (ret) {
+                gf_log ("cli", GF_LOG_ERROR, "Could not save clone "
+                        "name(%s)", (char *)words[cmdi]);
+                goto out;
+        }
+
+        /* Filling snap name in the dictionary */
+        ret = dict_set_str (dict, "snapname", (char *)words[cmdi+1]);
+        if (ret) {
+                gf_log ("cli", GF_LOG_ERROR, "Could not "
+                        "save snap name(%s)", (char *)words[cmdi+1]);
+                goto out;
+        }
+
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
 /* snapshot create <snapname> <vol-name(s)> [description <description>]
  *                                           [force]
  * @arg-0, dict     : Request Dictionary to be sent to server side.
@@ -4223,16 +4300,16 @@ out:
 }
 
 int
-validate_snapname (const char *snapname, char **opwords) {
+validate_op_name (const char *op, const char *opname, char **opwords) {
         int     ret     =       -1;
         int     i       =       0;
 
-        GF_ASSERT (snapname);
+        GF_ASSERT (opname);
         GF_ASSERT (opwords);
 
         for (i = 0 ; opwords[i] != NULL; i++) {
-                if (strcmp (opwords[i], snapname) == 0) {
-                        cli_out ("\"%s\" cannot be a snapname", snapname);
+                if (strcmp (opwords[i], opname) == 0) {
+                        cli_out ("\"%s\" cannot be a %s", opname, op);
                         goto out;
                 }
         }
@@ -4251,9 +4328,19 @@ cli_cmd_snapshot_parse (const char **words, int wordcount, dict_t **options,
         char               *w         = NULL;
         char               *opwords[] = {"create", "delete", "restore",
                                         "activate", "deactivate", "list",
-                                        "status", "config", "info", NULL};
+                                        "status", "config", "info", "clone",
+                                        NULL};
         char               *invalid_snapnames[] = {"description", "force",
                                                   "volume", "all", NULL};
+        char               *invalid_volnames[]  = {"volume", "type",
+                                                   "subvolumes", "option",
+                                                   "end-volume", "all",
+                                                   "volume_not_in_ring",
+                                                   "description", "force",
+                                                   "snap-max-hard-limit",
+                                                   "snap-max-soft-limit",
+                                                   "auto-delete",
+                                                   "activate-on-create", NULL};
 
         GF_ASSERT (words);
         GF_ASSERT (options);
@@ -4295,6 +4382,8 @@ cli_cmd_snapshot_parse (const char **words, int wordcount, dict_t **options,
                 type = GF_SNAP_OPTION_TYPE_ACTIVATE;
         } else if (!strcmp (w, "deactivate")) {
                 type = GF_SNAP_OPTION_TYPE_DEACTIVATE;
+        } else if (!strcmp(w, "clone")) {
+                type = GF_SNAP_OPTION_TYPE_CLONE;
         }
 
         if (type != GF_SNAP_OPTION_TYPE_CONFIG &&
@@ -4339,7 +4428,8 @@ cli_cmd_snapshot_parse (const char **words, int wordcount, dict_t **options,
                         goto out;
                 }
 
-                ret = validate_snapname (words[2], invalid_snapnames);
+                ret = validate_op_name ("snapname", words[2],
+                                        invalid_snapnames);
                 if (ret) {
                         goto out;
                 }
@@ -4352,6 +4442,35 @@ cli_cmd_snapshot_parse (const char **words, int wordcount, dict_t **options,
                 }
                 break;
 
+        case GF_SNAP_OPTION_TYPE_CLONE:
+                /* Syntax :
+                 * gluster snapshot clone <clonename> <snapname>
+                 */
+                /* In cases where the clonename is not given then
+                 * parsing fails & snapname cannot be "description",
+                 * "force" and "volume", that check is made here
+                 */
+                if (wordcount == 2) {
+                        ret = -1;
+                        gf_log ("cli", GF_LOG_ERROR, "Invalid Syntax");
+                        goto out;
+                }
+
+                ret = validate_op_name ("clonename", words[2],
+                                        invalid_volnames);
+                if (ret) {
+                        goto out;
+                }
+
+                ret = cli_snap_clone_parse (dict, words, wordcount);
+                if (ret) {
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "clone command parsing failed.");
+                        goto out;
+                }
+                break;
+
+
         case GF_SNAP_OPTION_TYPE_INFO:
                 /* Syntax :
                  * gluster snapshot info [(snapname] | [vol <volname>)]
diff --git a/cli/src/cli-cmd-snapshot.c b/cli/src/cli-cmd-snapshot.c
index 7c6a52e0a88..e885a641de2 100644
--- a/cli/src/cli-cmd-snapshot.c
+++ b/cli/src/cli-cmd-snapshot.c
@@ -87,6 +87,10 @@ struct cli_cmd snapshot_cmds[] = {
           cli_cmd_snapshot_cbk,
           "Snapshot Create."
         },
+        { "snapshot clone <clonename> <snapname>",
+          cli_cmd_snapshot_cbk,
+          "Snapshot Clone."
+        },
         { "snapshot restore <snapname>",
           cli_cmd_snapshot_cbk,
           "Snapshot Restore."
diff --git a/cli/src/cli-rpc-ops.c b/cli/src/cli-rpc-ops.c
index 8ea43f824bc..a02761d5e6e 100644
--- a/cli/src/cli-rpc-ops.c
+++ b/cli/src/cli-rpc-ops.c
@@ -8870,6 +8870,7 @@ gf_cli_snapshot_cbk (struct rpc_req *req, struct iovec *iov,
         gf_cli_rsp            rsp                      = {0, };
         dict_t               *dict                     = NULL;
         char                 *snap_name                = NULL;
+        char                 *clone_name               = NULL;
         int32_t               type                     =  0;
         call_frame_t         *frame                    = NULL;
         gf_boolean_t         snap_driven               = _gf_false;
@@ -8961,6 +8962,35 @@ gf_cli_snapshot_cbk (struct rpc_req *req, struct iovec *iov,
                 ret = 0;
                 break;
 
+        case GF_SNAP_OPTION_TYPE_CLONE:
+                if (rsp.op_ret) {
+                        cli_err("snapshot clone: failed: %s",
+                                 rsp.op_errstr ? rsp.op_errstr :
+                                 "Please check log file for details");
+                                 ret = rsp.op_ret;
+                                 goto out;
+                }
+
+                ret = dict_get_str (dict, "clonename", &clone_name);
+                if (ret) {
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "Failed to get clone name");
+                        goto out;
+                }
+
+                ret = dict_get_str (dict, "snapname", &snap_name);
+                if (ret) {
+                        gf_log ("cli", GF_LOG_ERROR,
+                                "Failed to get snapname name");
+                        goto out;
+                }
+
+                cli_out ("snapshot clone: success: Clone %s created "
+                                        "successfully", clone_name);
+
+                ret = 0;
+                break;
+
         case GF_SNAP_OPTION_TYPE_RESTORE:
                 /* TODO: Check if rsp.op_ret needs to be checked here. Or is
                  * it ok to check this in the start of the function where we
diff --git a/rpc/xdr/src/cli1-xdr.x b/rpc/xdr/src/cli1-xdr.x
index f7bfcf469d5..565f2182f6e 100644
--- a/rpc/xdr/src/cli1-xdr.x
+++ b/rpc/xdr/src/cli1-xdr.x
@@ -151,6 +151,7 @@ enum gf1_cli_snapshot {
         GF_SNAP_OPTION_TYPE_LIST,
         GF_SNAP_OPTION_TYPE_STATUS,
         GF_SNAP_OPTION_TYPE_CONFIG,
+        GF_SNAP_OPTION_TYPE_CLONE,
         GF_SNAP_OPTION_TYPE_INFO
 };
 
diff --git a/tests/basic/volume-snapshot-clone.t b/tests/basic/volume-snapshot-clone.t
new file mode 100755
index 00000000000..b142804e51b
--- /dev/null
+++ b/tests/basic/volume-snapshot-clone.t
@@ -0,0 +1,114 @@
+#!/bin/bash
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+. $(dirname $0)/../cluster.rc
+. $(dirname $0)/../snapshot.rc
+
+
+V1="patchy2"
+
+function create_volumes() {
+        $CLI_1 volume create $V0 $H1:$L1 &
+        PID_1=$!
+
+        $CLI_2 volume create $V1 $H2:$L2 $H3:$L3 &
+        PID_2=$!
+
+        wait $PID_1 $PID_2
+}
+
+function create_snapshots() {
+
+        $CLI_1 snapshot create $1 $2 no-timestamp&
+        PID_1=$!
+
+        wait $PID_1
+}
+
+
+
+function delete_snapshot() {
+        $CLI_1 snapshot delete $1 &
+        PID_1=$!
+
+        wait $PID_1
+}
+
+cleanup;
+
+TEST verify_lvm_version;
+#Create cluster with 3 nodes
+TEST launch_cluster 3;
+TEST setup_lvm 3
+
+TEST $CLI_1 peer probe $H2;
+TEST $CLI_1 peer probe $H3;
+EXPECT_WITHIN $PROBE_TIMEOUT 2 peer_count;
+
+create_volumes
+EXPECT 'Created' volinfo_field $V0 'Status';
+EXPECT 'Created' volinfo_field $V1 'Status';
+
+start_volumes 2
+EXPECT 'Started' volinfo_field $V0 'Status';
+EXPECT 'Started' volinfo_field $V1 'Status';
+
+TEST $CLI_1 snapshot config activate-on-create enable
+
+#Snapshot Operations
+create_snapshots ${V0}_snap ${V0};
+create_snapshots ${V1}_snap ${V1};
+
+EXPECT 'Started' snapshot_status ${V0}_snap;
+EXPECT 'Started' snapshot_status ${V1}_snap;
+
+sleep 5
+TEST $CLI_1 snapshot clone ${V0}_clone ${V0}_snap
+TEST $CLI_1 snapshot clone ${V1}_clone ${V1}_snap
+
+EXPECT 'Created' volinfo_field ${V0}_clone 'Status';
+EXPECT 'Created' volinfo_field ${V1}_clone 'Status';
+
+TEST $CLI_1 volume start ${V0}_clone force;
+TEST $CLI_1 volume start ${V1}_clone force;
+
+EXPECT 'Started' volinfo_field ${V0}_clone 'Status';
+EXPECT 'Started' volinfo_field ${V1}_clone 'Status';
+
+
+TEST glusterfs -s $H1 --volfile-id=/${V0}_clone $M0
+TEST glusterfs -s $H2 --volfile-id=/${V1}_clone $M1
+
+TEST touch $M0/file1
+TEST touch $M1/file1
+
+TEST echo "Hello world" $M0/file1
+TEST echo "Hello world" $M1/file1
+
+EXPECT_WITHIN $UMOUNT_TIMEOUT "Y" force_umount $M0
+EXPECT_WITHIN $UMOUNT_TIMEOUT "Y" force_umount $M1
+
+TEST kill_glusterd 2;
+TEST $glusterd_2;
+
+sleep 5
+
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT 'Started' volinfo_field ${V0}_clone 'Status';
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT 'Started' volinfo_field ${V1}_clone 'Status';
+#Clean up
+stop_force_volumes 2
+EXPECT 'Stopped' volinfo_field $V0 'Status';
+EXPECT 'Stopped' volinfo_field $V1 'Status';
+
+TEST delete_snapshot ${V0}_snap
+TEST delete_snapshot ${V1}_snap
+
+TEST ! snapshot_exists 1 ${V0}_snap
+TEST ! snapshot_exists 1 ${V1}_snap
+
+delete_volumes 2
+TEST ! volume_exists $V0
+TEST ! volume_exists $V1
+
+cleanup;
diff --git a/xlators/mgmt/glusterd/src/glusterd-snapshot-utils.c b/xlators/mgmt/glusterd/src/glusterd-snapshot-utils.c
index 97935b6a975..fb22fc1c23e 100644
--- a/xlators/mgmt/glusterd/src/glusterd-snapshot-utils.c
+++ b/xlators/mgmt/glusterd/src/glusterd-snapshot-utils.c
@@ -2172,6 +2172,7 @@ glusterd_snap_use_rsp_dict (dict_t *dst, dict_t *src)
         switch (snap_command) {
         case GF_SNAP_OPTION_TYPE_CREATE:
         case GF_SNAP_OPTION_TYPE_DELETE:
+        case GF_SNAP_OPTION_TYPE_CLONE:
                 ret = glusterd_snap_create_use_rsp_dict (dst, src);
                 if (ret) {
                         gf_log ("", GF_LOG_ERROR, "Unable to use rsp dict");
@@ -2781,6 +2782,21 @@ glusterd_snap_quorum_check (dict_t *dict, gf_boolean_t snap_volume,
                         goto out;
                 }
                 break;
+        case GF_SNAP_OPTION_TYPE_CLONE:
+
+                if (!does_gd_meet_server_quorum (this, peers_list, _gf_true)) {
+                        ret = -1;
+                        snprintf (err_str, sizeof (err_str),
+                                  "glusterds are not in quorum");
+                        gf_log (this->name, GF_LOG_WARNING, "%s",
+                                err_str);
+                        *op_errstr = gf_strdup (err_str);
+                        goto out;
+                }
+
+                gf_log (this->name, GF_LOG_DEBUG, "glusterds are in "
+                        "quorum");
+                break;
         case GF_SNAP_OPTION_TYPE_DELETE:
         case GF_SNAP_OPTION_TYPE_RESTORE:
                 if (!does_gd_meet_server_quorum (this, peers_list, _gf_true)) {
diff --git a/xlators/mgmt/glusterd/src/glusterd-snapshot.c b/xlators/mgmt/glusterd/src/glusterd-snapshot.c
index 93e695c85f5..761115d22ed 100644
--- a/xlators/mgmt/glusterd/src/glusterd-snapshot.c
+++ b/xlators/mgmt/glusterd/src/glusterd-snapshot.c
@@ -1388,7 +1388,7 @@ out:
         return ret;
 }
 int
-glusterd_snap_create_pre_val_use_rsp_dict (dict_t *dst, dict_t *src)
+glusterd_snap_create_clone_pre_val_use_rsp_dict (dict_t *dst, dict_t *src)
 {
         char                  *snap_brick_dir        = NULL;
         char                  *snap_device           = NULL;
@@ -1696,7 +1696,9 @@ glusterd_snap_pre_validate_use_rsp_dict (dict_t *dst, dict_t *src)
 
         switch (snap_command) {
         case GF_SNAP_OPTION_TYPE_CREATE:
-                ret = glusterd_snap_create_pre_val_use_rsp_dict (dst, src);
+        case GF_SNAP_OPTION_TYPE_CLONE:
+                ret = glusterd_snap_create_clone_pre_val_use_rsp_dict (dst,
+                                                                       src);
                 if (ret) {
                         gf_log (this->name, GF_LOG_ERROR, "Unable to use "
                                 "rsp dict");
@@ -1833,21 +1835,302 @@ out:
 }
 
 int
+glusterd_snap_create_clone_common_prevalidate (dict_t *rsp_dict, int flags,
+            char *snapname, char *err_str, char *snap_volname, int64_t volcount,
+            glusterd_volinfo_t *volinfo, gf_loglevel_t *loglevel, int clone)  {
+
+        char                  *device            = NULL;
+        char                   key[PATH_MAX]     = "";
+        int                    ret               = -1;
+        int64_t                i                 = 1;
+        int64_t                brick_order       = 0;
+        int64_t                brick_count       = 0;
+        xlator_t              *this              = NULL;
+        glusterd_conf_t       *conf              = NULL;
+        glusterd_brickinfo_t  *brickinfo         = NULL;
+
+        this = THIS;
+        conf = this->private;
+        GF_ASSERT (conf);
+
+        if (!snapname || !volinfo) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to validate "
+                        "snapname or volume information");
+                ret = -1;
+                goto out;
+        }
+
+        cds_list_for_each_entry (brickinfo, &volinfo->bricks,
+                                 brick_list) {
+                if (uuid_compare (brickinfo->uuid, MY_UUID)) {
+                        brick_order++;
+                        continue;
+                }
+
+                if (!glusterd_is_brick_started (brickinfo)) {
+                        if (!clone && (flags & GF_CLI_FLAG_OP_FORCE)) {
+                                gf_log (this->name, GF_LOG_WARNING,
+                                        "brick %s:%s is not started",
+                                        brickinfo->hostname,
+                                        brickinfo->path);
+                                brick_order++;
+                                brick_count++;
+                                continue;
+                        }
+                        if (!clone) {
+                        snprintf (err_str, PATH_MAX,
+                                  "brick %s:%s is not started. "
+                                  "Please start the stopped brick "
+                                  "and then issue snapshot create "
+                                  "command or use [force] option in "
+                                  "snapshot create to override this "
+                                  "behavior.", brickinfo->hostname,
+                                  brickinfo->path);
+                        } else {
+                        snprintf (err_str, PATH_MAX,
+                                  "brick %s:%s is not started. "
+                                  "Please start the stopped brick "
+                                  "and then issue snapshot clone "
+                                  "command ", brickinfo->hostname,
+                                  brickinfo->path);
+                        }
+                        ret = -1;
+                        goto out;
+                }
+
+
+                device = glusterd_get_brick_mount_device
+                                          (brickinfo->path);
+                if (!device) {
+                        snprintf (err_str, PATH_MAX,
+                                  "getting device name for the brick "
+                                 "%s:%s failed", brickinfo->hostname,
+                                  brickinfo->path);
+                        ret = -1;
+                        goto out;
+                }
+                if (!clone) {
+                     if (!glusterd_is_thinp_brick (device)) {
+                                snprintf (err_str, PATH_MAX,
+                                          "Snapshot is supported only for "
+                                          "thin provisioned LV. Ensure that "
+                                          "all bricks of %s are thinly "
+                                          "provisioned LV.", volinfo->volname);
+                                ret = -1;
+                                goto out;
+                     }
+                }
+
+                device = glusterd_build_snap_device_path (device,
+                                                  snap_volname,
+                                                  brick_count);
+                if (!device) {
+                        snprintf (err_str, PATH_MAX,
+                                  "cannot copy the snapshot device "
+                                "name (volname: %s, snapname: %s)",
+                                volinfo->volname, snapname);
+                        *loglevel = GF_LOG_WARNING;
+                        ret = -1;
+                        goto out;
+                }
+
+                snprintf (key, sizeof(key),
+                        "vol%"PRId64".brick_snapdevice%"PRId64,
+                        i, brick_count);
+                ret = dict_set_dynstr (rsp_dict, key, device);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to set %s", key);
+                        GF_FREE (device);
+                        goto out;
+                }
+                device = NULL;
+
+                ret = glusterd_update_mntopts (brickinfo->path,
+                                               brickinfo);
+                if (ret) {
+                         gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                                "update mount options for %s brick",
+                                brickinfo->path);
+                }
+
+                snprintf (key, sizeof(key), "vol%"PRId64".fstype%"
+                          PRId64, i, brick_count);
+                ret = dict_set_dynstr_with_alloc (rsp_dict, key,
+                                                  brickinfo->fstype);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to set %s", key);
+                        goto out;
+                }
+
+                snprintf (key, sizeof(key), "vol%"PRId64".mnt_opts%"
+                          PRId64, i, brick_count);
+                ret = dict_set_dynstr_with_alloc (rsp_dict, key,
+                                                  brickinfo->mnt_opts);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to set %s", key);
+                        goto out;
+                }
+
+                snprintf (key, sizeof(key), "vol%"PRId64".brickdir%"PRId64, i,
+                          brick_count);
+                ret = dict_set_dynstr_with_alloc (rsp_dict, key,
+                                                  brickinfo->mount_dir);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to set %s", key);
+                        goto out;
+                }
+
+                snprintf (key, sizeof(key) - 1,
+                          "vol%"PRId64".brick%"PRId64".order", i, brick_count);
+                ret = dict_set_int64 (rsp_dict, key, brick_order);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to set %s", key);
+                        goto out;
+                }
+
+                snprintf (key, sizeof (key),
+                          "vol%"PRId64".brick%"PRId64".status", i, brick_order);
+
+                ret = glusterd_add_brick_status_to_dict (rsp_dict,
+                                                         volinfo,
+                                                         brickinfo,
+                                                         key);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "failed to "
+                                "add brick status to dict");
+                        goto out;
+                }
+                brick_count++;
+                brick_order++;
+        }
+        snprintf (key, sizeof(key) - 1, "vol%"PRId64"_brickcount", volcount);
+        ret = dict_set_int64 (rsp_dict, key, brick_count);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set %s",
+                        key);
+                goto out;
+        }
+        ret = 0;
+out:
+        if (device)
+               GF_FREE (device);
+
+       return ret;
+
+}
+
+
+int
+glusterd_snapshot_clone_prevalidate (dict_t *dict, char **op_errstr,
+                                      dict_t *rsp_dict)
+{
+        char                  *clonename         = NULL;
+        char                  *snapname          = NULL;
+        char                   key[PATH_MAX]     = "";
+        glusterd_snap_t       *snap              = NULL;
+        char                   err_str[PATH_MAX] = "";
+        int                    ret               = -1;
+        int64_t                volcount          = 1;
+        glusterd_volinfo_t    *snap_vol          = NULL;
+        xlator_t              *this              = NULL;
+        uuid_t                *snap_volid        = NULL;
+        gf_loglevel_t          loglevel          = GF_LOG_ERROR;
+
+        this = THIS;
+        GF_ASSERT (op_errstr);
+        GF_ASSERT (dict);
+
+        ret = dict_get_str (dict, "clonename", &clonename);
+        if (ret) {
+                snprintf (err_str, sizeof (err_str), "Failed to "
+                          "get the clone name");
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "snapname", &snapname);
+        if (ret) {
+                snprintf (err_str, sizeof (err_str), "Failed to get snapname");
+                goto out;
+        }
+
+        if (glusterd_check_volume_exists(clonename)) {
+                ret = -1;
+                snprintf (err_str, sizeof (err_str), "Volume with name:%s "
+                          "already exists", clonename);
+                goto out;
+        }
+        /* need to find snap volinfo*/
+        snap = glusterd_find_snap_by_name (snapname);
+        if (!snap) {
+                ret = -1;
+                snprintf (err_str, sizeof (err_str), "Failed to find :%s "
+                          "snap", snapname);
+                goto out;
+        }
+
+        /* TODO : As of now there is only one volume in snapshot.
+         * Change this when multiple volume snapshot is introduced
+         */
+        snap_vol = list_entry (snap->volumes.next,
+                                   glusterd_volinfo_t, vol_list);
+        if (!snap_vol) {
+                gf_log ("glusterd", GF_LOG_ERROR, "Failed to get snap "
+                "volinfo %s", snap->snapname);
+                goto out;
+        }
+
+        snprintf (key, sizeof(key) - 1, "vol1_volid");
+        ret = dict_get_bin (dict, key, (void **)&snap_volid);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to fetch snap_volid");
+                goto out;
+        }
+
+
+        /* Adding snap bricks mount paths to the dict */
+        ret = glusterd_snap_create_clone_common_prevalidate (rsp_dict, 0,
+                        snapname, err_str, clonename, 1, snap_vol,
+                        &loglevel, 1);
+        if (ret) {
+                goto out;
+        }
+        ret = dict_set_int64 (rsp_dict, "volcount", volcount);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set volcount");
+                goto out;
+        }
+
+        ret = 0;
+out:
+
+        if (ret && err_str[0] != '\0') {
+                gf_log (this->name, loglevel, "%s", err_str);
+                *op_errstr = gf_strdup (err_str);
+        }
+
+        gf_log (this->name, GF_LOG_TRACE, "Returning %d", ret);
+        return ret;
+}
+
+
+int
 glusterd_snapshot_create_prevalidate (dict_t *dict, char **op_errstr,
                                       dict_t *rsp_dict)
 {
         char                  *volname           = NULL;
         char                  *snapname          = NULL;
-        char                  *device            = NULL;
         char                   key[PATH_MAX]     = "";
         char                   snap_volname[64]  = "";
         char                   err_str[PATH_MAX] = "";
         int                    ret               = -1;
         int64_t                i                 = 0;
         int64_t                volcount          = 0;
-        int64_t                brick_count       = 0;
-        int64_t                brick_order       = 0;
-        glusterd_brickinfo_t  *brickinfo         = NULL;
         glusterd_volinfo_t    *volinfo           = NULL;
         xlator_t              *this              = NULL;
         uuid_t                *snap_volid        = NULL;
@@ -1986,156 +2269,15 @@ glusterd_snapshot_create_prevalidate (dict_t *dict, char **op_errstr,
                    provided by user */
                 GLUSTERD_GET_UUID_NOHYPHEN (snap_volname, *snap_volid);
 
-                brick_count = 0;
-                brick_order = 0;
-                /* Adding snap bricks mount paths to the dict */
-                cds_list_for_each_entry (brickinfo, &volinfo->bricks,
-                                         brick_list) {
-                        if (uuid_compare (brickinfo->uuid, MY_UUID)) {
-                                brick_order++;
-                                continue;
-                        }
-
-                        if (!glusterd_is_brick_started (brickinfo)) {
-                                if(flags & GF_CLI_FLAG_OP_FORCE) {
-                                        gf_log (this->name, GF_LOG_WARNING,
-                                                "brick %s:%s is not started",
-                                                brickinfo->hostname,
-                                                brickinfo->path);
-                                        brick_order++;
-                                        brick_count++;
-                                        continue;
-                                }
-
-                                snprintf (err_str, sizeof (err_str),
-                                          "brick %s:%s is not started. "
-                                          "Please start the stopped brick "
-                                          "and then issue snapshot create "
-                                          "command or use [force] option in "
-                                          "snapshot create to override this "
-                                          "behavior.", brickinfo->hostname,
-                                          brickinfo->path);
-                                ret = -1;
-                                goto out;
-                        }
-
-                        device = glusterd_get_brick_mount_device
-                                                          (brickinfo->path);
-                        if (!device) {
-                                snprintf (err_str, sizeof (err_str),
-                                          "getting device name for the brick "
-                                          "%s:%s failed", brickinfo->hostname,
-                                          brickinfo->path);
-                                ret = -1;
-                                goto out;
-                        }
-
-                        if (!glusterd_is_thinp_brick (device)) {
-                                snprintf (err_str, sizeof (err_str),
-                                          "Snapshot is supported only for "
-                                          "thin provisioned LV. Ensure that "
-                                          "all bricks of %s are thinly "
-                                          "provisioned LV.", volinfo->volname);
-                                ret = -1;
-                                goto out;
-                        }
-
-                        device = glusterd_build_snap_device_path (device,
-                                                                  snap_volname,
-                                                                  brick_count);
-                        if (!device) {
-                                snprintf (err_str, sizeof (err_str),
-                                          "cannot copy the snapshot device "
-                                          "name (volname: %s, snapname: %s)",
-                                          volinfo->volname, snapname);
-                                loglevel = GF_LOG_WARNING;
-                                ret = -1;
-                                goto out;
-                        }
-
-                        snprintf (key, sizeof(key),
-                                  "vol%"PRId64".brick_snapdevice%"PRId64, i,
-                                  brick_count);
-                        ret = dict_set_dynstr (rsp_dict, key, device);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR,
-                                        "Failed to set %s", key);
-                                GF_FREE (device);
-                                goto out;
-                        }
-                        device = NULL;
-
-                        ret = glusterd_update_mntopts (brickinfo->path,
-                                                       brickinfo);
-                        if (ret) {
-                                 gf_log (this->name, GF_LOG_ERROR, "Failed to "
-                                         "update mount options for %s brick",
-                                         brickinfo->path);
-                        }
-
-                        snprintf (key, sizeof(key), "vol%"PRId64".fstype%"
-                                  PRId64, i, brick_count);
-                        ret = dict_set_dynstr_with_alloc (rsp_dict, key,
-                                                          brickinfo->fstype);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR,
-                                        "Failed to set %s", key);
-                                goto out;
-                        }
-
-                        snprintf (key, sizeof(key), "vol%"PRId64".mnt_opts%"
-                                  PRId64, i, brick_count);
-                        ret = dict_set_dynstr_with_alloc (rsp_dict, key,
-                                                          brickinfo->mnt_opts);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR,
-                                        "Failed to set %s", key);
-                                goto out;
-                        }
-
-                        snprintf (key, sizeof(key), "vol%"PRId64".brickdir%"PRId64, i,
-                                  brick_count);
-                        ret = dict_set_dynstr_with_alloc (rsp_dict, key,
-                                                          brickinfo->mount_dir);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR,
-                                        "Failed to set %s", key);
-                                goto out;
-                        }
-
-                        snprintf (key, sizeof(key) - 1, "vol%"PRId64".brick%"PRId64".order",
-                                  i, brick_count);
-                        ret = dict_set_int64 (rsp_dict, key, brick_order);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR,
-                                        "Failed to set %s", key);
-                                goto out;
-                        }
-
-                        snprintf (key, sizeof (key), "vol%"PRId64".brick%"PRId64".status",
-                                  i, brick_order);
-
-                        ret = glusterd_add_brick_status_to_dict (rsp_dict,
-                                                                 volinfo,
-                                                                 brickinfo,
-                                                                 key);
-                        if (ret) {
-                                gf_log (this->name, GF_LOG_ERROR, "failed to "
-                                        "add brick status to dict");
-                                goto out;
-                        }
-
-                        brick_count++;
-                        brick_order++;
-                }
-
-                snprintf (key, sizeof(key) - 1, "vol%"PRId64"_brickcount", i);
-                ret = dict_set_int64 (rsp_dict, key, brick_count);
+                ret = glusterd_snap_create_clone_common_prevalidate (rsp_dict,
+                                flags, snapname, err_str, snap_volname, i,
+                                volinfo, &loglevel, 0);
                 if (ret) {
-                        gf_log (this->name, GF_LOG_ERROR, "Failed to set %s",
-                                key);
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Failed to pre validate");
                         goto out;
                 }
+
         }
 
         ret = dict_set_int64 (rsp_dict, "volcount", volcount);
@@ -2146,9 +2288,6 @@ glusterd_snapshot_create_prevalidate (dict_t *dict, char **op_errstr,
 
         ret = 0;
 out:
-        if (device)
-                GF_FREE (device);
-
         if (ret && err_str[0] != '\0') {
                 gf_log (this->name, loglevel, "%s", err_str);
                 *op_errstr = gf_strdup (err_str);
@@ -3678,6 +3817,145 @@ out:
         return ret;
 }
 
+/* This is a snapshot clone handler function. This function will be
+ * executed in the originator node. This function is responsible for
+ * calling mgmt_v3 framework to do the actual snap clone on all the bricks
+ *
+ * @param req           RPC request object
+ * @param op            gluster operation
+ * @param dict          dictionary containing snapshot restore request
+ * @param err_str       In case of an err this string should be populated
+ * @param len           length of err_str buffer
+ *
+ * @return              Negative value on Failure and 0 in success
+ */
+int
+glusterd_handle_snapshot_clone (rpcsvc_request_t *req, glusterd_op_t op,
+                               dict_t *dict, char *err_str, size_t len)
+{
+        int           ret                              = -1;
+        char         *clonename                        = NULL;
+        char         *snapname                         = NULL;
+        int64_t       volcount                         = 0;
+        xlator_t     *this                             = NULL;
+        char          key[PATH_MAX]                    = "";
+        char         *username                         = NULL;
+        char         *password                         = NULL;
+        char         *volname                          = NULL;
+        uuid_t       *uuid_ptr                         = NULL;
+        uuid_t        tmp_uuid                         = {0};
+        int           i                                = 0;
+        char          snap_volname[GD_VOLUME_NAME_MAX] = {0, };
+
+        this = THIS;
+        GF_ASSERT (this);
+        GF_ASSERT (req);
+        GF_ASSERT (dict);
+        GF_ASSERT (err_str);
+
+        ret = dict_get_str (dict, "clonename", &clonename);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to "
+                        "get the clone name");
+                goto out;
+        }
+        /*We need to take a volume lock on clone name*/
+        volname = gf_strdup (clonename);
+        snprintf (key, sizeof(key), "volname1");
+        ret = dict_set_dynstr (dict, key, volname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set clone "
+                        "name for volume locking");
+                GF_FREE (volname);
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "snapname", &snapname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "failed to get the snapname");
+                goto out;
+        }
+
+        uuid_ptr = GF_CALLOC (1, sizeof(uuid_t), gf_common_mt_uuid_t);
+        if (!uuid_ptr) {
+                gf_log (this->name, GF_LOG_ERROR, "Out Of Memory");
+                ret = -1;
+                goto out;
+        }
+
+        uuid_generate (*uuid_ptr);
+        ret = dict_set_bin (dict, "clone-id", uuid_ptr, sizeof(uuid_t));
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to set clone-id");
+                GF_FREE (uuid_ptr);
+                goto out;
+        }
+        uuid_ptr = NULL;
+
+        ret = dict_get_str (dict, "snapname", &snapname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Failed to get snapname name");
+                goto out;
+        }
+
+        uuid_generate (tmp_uuid);
+        username = gf_strdup (uuid_utoa (tmp_uuid));
+        snprintf (key, sizeof(key), "volume1_username");
+        ret = dict_set_dynstr (dict, key, username);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set clone "
+                        "username for volume %s", clonename);
+                GF_FREE (username);
+                goto out;
+        }
+
+        uuid_generate (tmp_uuid);
+        password = gf_strdup (uuid_utoa (tmp_uuid));
+        snprintf (key, sizeof(key), "volume1_password");
+        ret = dict_set_dynstr (dict, key, password);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to set clone "
+                        "password for volume %s", clonename);
+                GF_FREE (password);
+                goto out;
+        }
+
+        uuid_ptr = GF_CALLOC (1, sizeof(uuid_t), gf_common_mt_uuid_t);
+        if (!uuid_ptr) {
+                gf_log (this->name, GF_LOG_ERROR, "Out Of Memory");
+                ret = -1;
+                goto out;
+        }
+
+        snprintf (key, sizeof(key) - 1, "vol1_volid");
+        uuid_generate (*uuid_ptr);
+        ret = dict_set_bin (dict, key, uuid_ptr, sizeof(uuid_t));
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to set clone_volid");
+                GF_FREE (uuid_ptr);
+                goto out;
+        }
+        snprintf (key, sizeof (key), "clone-volname%d", i);
+        ret = dict_set_dynstr_with_alloc (dict, key, snap_volname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to set snap volname");
+                GF_FREE (uuid_ptr);
+                goto out;
+        }
+
+        ret = glusterd_mgmt_v3_initiate_snap_phases (req, op, dict);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to initiate snap "
+                                "phases");
+        }
+
+out:
+        return ret;
+}
+
 
 /* This is a snapshot restore handler function. This function will be
  * executed in the originator node. This function is responsible for
@@ -3730,7 +4008,7 @@ glusterd_handle_snapshot_restore (rpcsvc_request_t *req, glusterd_op_t op,
                 goto out;
         }
 
-        cds_list_for_each_entry (snap_volinfo, &snap->volumes, vol_list) {
+        list_for_each_entry (snap_volinfo, &snap->volumes, vol_list) {
                 i++;
                 snprintf (key, sizeof (key), "volname%d", i);
                 buf = gf_strdup (snap_volinfo->parent_volname);
@@ -4112,7 +4390,8 @@ static int32_t
 glusterd_add_brick_to_snap_volume (dict_t *dict, dict_t *rsp_dict,
                                     glusterd_volinfo_t  *snap_vol,
                                     glusterd_brickinfo_t *original_brickinfo,
-                                    int64_t volcount, int32_t brick_count)
+                                    int64_t volcount, int32_t brick_count,
+                                    int clone)
 {
         char                    key[PATH_MAX]                   = "";
         char                   *value                           = NULL;
@@ -4255,9 +4534,14 @@ glusterd_add_brick_to_snap_volume (dict_t *dict, dict_t *rsp_dict,
         uuid_copy (snap_brickinfo->uuid, original_brickinfo->uuid);
         /* AFR changelog names are based on brick_id and hence the snap
          * volume's bricks must retain the same ID */
-        strcpy (snap_brickinfo->brick_id, original_brickinfo->brick_id);
         cds_list_add_tail (&snap_brickinfo->brick_list, &snap_vol->bricks);
 
+        if (clone) {
+                GLUSTERD_ASSIGN_BRICKID_TO_BRICKINFO (snap_brickinfo, snap_vol,
+                                                      brick_count);
+        } else
+                strcpy (snap_brickinfo->brick_id, original_brickinfo->brick_id);
+
 out:
         if (ret && snap_brickinfo)
                 GF_FREE (snap_brickinfo);
@@ -4414,7 +4698,8 @@ out:
 
 glusterd_volinfo_t *
 glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
-                      dict_t *dict, dict_t *rsp_dict, int64_t volcount)
+                      dict_t *dict, dict_t *rsp_dict, int64_t volcount,
+                      int clone)
 {
         char                    key[PATH_MAX]                   = "";
         char                   *username                        = NULL;
@@ -4427,6 +4712,7 @@ glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
         int32_t                 brick_count                     = 0;
         xlator_t               *this                            = NULL;
         int64_t                 brick_order                     = 0;
+        char *clonename         = NULL;
 
         this = THIS;
         GF_ASSERT (this);
@@ -4445,7 +4731,6 @@ glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
                         "snap %s", key, snap->snapname);
                 goto out;
         }
-
         snprintf (key, sizeof(key), "volume%"PRId64"_password", volcount);
         ret = dict_get_str (dict, key, &password);
         if (ret) {
@@ -4475,34 +4760,54 @@ glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
 
         /* uuid is used as lvm snapshot name.
            This will avoid restrictions on snapshot names provided by user */
-        GLUSTERD_GET_UUID_NOHYPHEN (snap_vol->volname, *snap_volid);
         uuid_copy (snap_vol->volume_id, *snap_volid);
         snap_vol->is_snap_volume = _gf_true;
-        strcpy (snap_vol->parent_volname, origin_vol->volname);
         snap_vol->snapshot = snap;
 
-        glusterd_auth_set_username (snap_vol, username);
-        glusterd_auth_set_password (snap_vol, password);
-
+        if (clone) {
+                snap_vol->is_snap_volume = _gf_false;
+                ret = dict_get_str (dict, "clonename", &clonename);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to get %s "
+                                "for snap %s", key, snap->snapname);
+                        goto out;
+                }
+                cds_list_add_tail (&snap_vol->vol_list, &snap->volumes);
+                strcpy(snap_vol->volname, clonename);
+        } else {
+                GLUSTERD_GET_UUID_NOHYPHEN (snap_vol->volname, *snap_volid);
+                strcpy (snap_vol->parent_volname, origin_vol->volname);
+                ret = glusterd_list_add_snapvol (origin_vol, snap_vol);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "could not add the "
+                                "snap volume %s to the list",
+                                snap_vol->volname);
+                        goto out;
+                }
         /* TODO : Sync before taking a snapshot */
         /* Copy the status and config files of geo-replication before
          * taking a snapshot. During restore operation these files needs
          * to be copied back in /var/lib/glusterd/georeplication/
          */
-        ret = glusterd_copy_geo_rep_files (origin_vol, snap_vol, rsp_dict);
-        if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "Failed to copy geo-rep "
-                        "config and status files for volume %s",
-                        origin_vol->volname);
-                goto out;
+                ret = glusterd_copy_geo_rep_files (origin_vol, snap_vol,
+                                                   rsp_dict);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to copy "
+                                "geo-rep config and status files for volume %s",
+                                origin_vol->volname);
+                        goto out;
+                }
         }
+        glusterd_auth_set_username (snap_vol, username);
+        glusterd_auth_set_password (snap_vol, password);
 
         /* Adding snap brickinfos to the snap volinfo */
         brick_count = 0;
         cds_list_for_each_entry (brickinfo, &origin_vol->bricks, brick_list) {
                 ret = glusterd_add_brick_to_snap_volume (dict, rsp_dict,
                                                          snap_vol, brickinfo,
-                                                         volcount, brick_count);
+                                                         volcount, brick_count,
+                                                         clone);
                 if (ret) {
                         gf_log (this->name, GF_LOG_ERROR,
                                 "Failed to add the snap brick for "
@@ -4510,7 +4815,6 @@ glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
                                 brickinfo->hostname, brickinfo->path);
                         goto out;
                 }
-
                 brick_count++;
         }
 
@@ -4566,13 +4870,6 @@ glusterd_do_snap_vol (glusterd_volinfo_t *origin_vol, glusterd_snap_t *snap,
                 goto out;
         }
 
-        ret = glusterd_list_add_snapvol (origin_vol, snap_vol);
-        if (ret) {
-                gf_log (this->name, GF_LOG_ERROR, "could not add the snap "
-                        "volume %s to the list", snap_vol->volname);
-                goto out;
-        }
-
 out:
         if (ret) {
                 if (snap_vol)
@@ -5523,7 +5820,7 @@ glusterd_schedule_brick_snapshot (dict_t *dict, dict_t *rsp_dict,
                 brickcount = 0;
                 brickorder = 0;
                 cds_list_for_each_entry (brickinfo, &snap_vol->bricks,
-                                         brick_list) {
+                                                            brick_list) {
                         snprintf (key, sizeof(key) - 1,
                                   "snap-vol%d.brick%d.order", volcount,
                                   brickcount);
@@ -5609,6 +5906,187 @@ out:
         return ret;
 }
 
+glusterd_snap_t*
+glusterd_create_snap_object_for_clone (dict_t *dict, dict_t *rsp_dict)
+{
+        char              *snapname    = NULL;
+        uuid_t            *snap_id     = NULL;
+        glusterd_snap_t   *snap        = NULL;
+        xlator_t          *this        = NULL;
+        glusterd_conf_t   *priv        = NULL;
+        int                ret         = -1;
+        int64_t            time_stamp  = 0;
+
+        this = THIS;
+        priv = this->private;
+
+        GF_ASSERT (dict);
+        GF_ASSERT (rsp_dict);
+
+        /* Fetch snapname, description, id and time from dict */
+        ret = dict_get_str (dict, "clonename", &snapname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch clonename");
+                goto out;
+        }
+
+        ret = dict_get_bin (dict, "clone-id", (void **)&snap_id);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch clone_id");
+                goto out;
+        }
+
+        snap = glusterd_new_snap_object ();
+        if (!snap) {
+                gf_log (this->name, GF_LOG_ERROR, "Could not create "
+                        "the snap object for snap %s", snapname);
+                goto out;
+        }
+
+        strcpy (snap->snapname, snapname);
+        uuid_copy (snap->snap_id, *snap_id);
+
+        ret = 0;
+
+out:
+        if (ret) {
+                if (snap)
+                        glusterd_snap_remove (rsp_dict, snap,
+                                              _gf_true, _gf_true);
+                snap = NULL;
+        }
+
+        return snap;
+}
+
+
+int32_t
+glusterd_snapshot_clone_commit (dict_t *dict, char **op_errstr,
+                                 dict_t *rsp_dict)
+{
+        int                     ret                     = -1;
+        int64_t                 i                       = 0;
+        int64_t                 volcount                = 0;
+        int32_t                 snap_activate           = 0;
+        char                    *snapname               = NULL;
+        char                    *volname                = NULL;
+        char                    *tmp_name               = NULL;
+        char                    key[PATH_MAX]           = "";
+        xlator_t                *this                   = NULL;
+        glusterd_snap_t         *snap_parent            = NULL;
+        glusterd_snap_t         *snap                   = NULL;
+        glusterd_volinfo_t      *origin_vol             = NULL;
+        glusterd_volinfo_t      *snap_vol               = NULL;
+        glusterd_brickinfo_t    *brickinfo              = NULL;
+        glusterd_conf_t         *priv                   = NULL;
+
+        this = THIS;
+        GF_ASSERT(this);
+        GF_ASSERT(dict);
+        GF_ASSERT(op_errstr);
+        GF_ASSERT(rsp_dict);
+        priv = this->private;
+        GF_ASSERT(priv);
+
+
+        ret = dict_get_str (dict, "clonename", &snapname);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch clonename");
+                goto out;
+        }
+        tmp_name = gf_strdup (snapname);
+        if (!tmp_name) {
+                gf_log (this->name, GF_LOG_ERROR, "Out of memory");
+                ret = -1;
+                goto out;
+        }
+
+        ret = dict_set_dynstr (rsp_dict, "clonename", tmp_name);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Unable to set clonename in rsp_dict");
+                GF_FREE (tmp_name);
+                goto out;
+        }
+        tmp_name = NULL;
+
+
+       ret = dict_get_str (dict, "snapname", &volname);
+       if (ret) {
+               gf_log (this->name, GF_LOG_ERROR,
+                       "failed to get snap name");
+               goto out;
+       }
+       snap_parent = glusterd_find_snap_by_name (volname);
+       /* TODO : As of now there is only one volume in snapshot.
+        * Change this when multiple volume snapshot is introduced
+        */
+       origin_vol = cds_list_entry (snap_parent->volumes.next,
+                                  glusterd_volinfo_t, vol_list);
+       if (!origin_vol) {
+               gf_log ("glusterd", GF_LOG_ERROR, "Failed to get snap "
+               "volinfo %s", snap_parent->snapname);
+               goto out;
+       }
+       snap = glusterd_create_snap_object_for_clone (dict, rsp_dict);
+       if (!snap) {
+               gf_log (this->name, GF_LOG_ERROR, "creating the"
+                       "snap object %s failed", snapname);
+               ret = -1;
+               goto out;
+       }
+
+       snap_vol = glusterd_do_snap_vol (origin_vol, snap, dict,
+                                       rsp_dict, 1, 1);
+       if (!snap_vol) {
+              ret = -1;
+              gf_log (this->name, GF_LOG_WARNING, "taking the "
+                      "snapshot of the volume %s failed", volname);
+              goto out;
+       }
+       volcount = 1;
+       ret = dict_set_int64 (rsp_dict, "volcount", volcount);
+       if (ret) {
+               gf_log (this->name, GF_LOG_ERROR, "Failed to set volcount");
+               goto out;
+       }
+
+       ret = glusterd_schedule_brick_snapshot (dict, rsp_dict, snap);
+       if (ret) {
+               gf_log (this->name, GF_LOG_ERROR, "Failed to take backend "
+                       "snapshot %s", snap->snapname);
+               goto out;
+       }
+
+       cds_list_del_init (&snap_vol->vol_list);
+       ret = dict_set_dynstr_with_alloc (rsp_dict, "snapuuid",
+                                         uuid_utoa (snap->snap_id));
+       if (ret) {
+               gf_log (this->name, GF_LOG_ERROR, "Failed to set snap "
+                       "uuid in response dictionary for %s snapshot",
+                       snap->snapname);
+               goto out;
+       }
+       ret = glusterd_snapdsvc_init (snap_vol);
+       glusterd_list_add_order (&snap_vol->vol_list, &priv->volumes,
+                       glusterd_compare_volume_name);
+
+       ret = 0;
+
+
+out:
+       if (ret) {
+               if (snap)
+                       glusterd_snap_remove (rsp_dict, snap,
+                                             _gf_true, _gf_true);
+               snap = NULL;
+       }
+
+       gf_log (this->name, GF_LOG_TRACE, "Returning %d", ret);
+       return ret;
+}
+
+
 int32_t
 glusterd_snapshot_create_commit (dict_t *dict, char **op_errstr,
                                  dict_t *rsp_dict)
@@ -5701,7 +6179,7 @@ glusterd_snapshot_create_commit (dict_t *dict, char **op_errstr,
                 }
 
                 snap_vol = glusterd_do_snap_vol (origin_vol, snap, dict,
-                                                 rsp_dict, i);
+                                                 rsp_dict, i, 0);
                 if (!snap_vol) {
                         ret = -1;
                         gf_log (this->name, GF_LOG_WARNING, "taking the "
@@ -6786,6 +7264,86 @@ out:
 }
 
 int32_t
+glusterd_snapshot_clone_postvalidate (dict_t *dict, int32_t op_ret,
+                                       char **op_errstr, dict_t *rsp_dict)
+{
+        xlator_t              *this           = NULL;
+        glusterd_conf_t       *priv           = NULL;
+        int                    ret            = -1;
+        int32_t                cleanup        = 0;
+        glusterd_snap_t       *snap           = NULL;
+        glusterd_volinfo_t    *snap_vol       = NULL;
+        char                  *clonename      = NULL;
+        char                  *auto_delete    = NULL;
+
+        this = THIS;
+
+        GF_ASSERT (this);
+        GF_ASSERT (dict);
+        GF_ASSERT (rsp_dict);
+
+        priv = this->private;
+        GF_ASSERT (priv);
+
+        if (op_ret) {
+                ret = dict_get_int32 (dict, "cleanup", &cleanup);
+                if (!ret && cleanup) {
+                        ret = glusterd_do_snap_cleanup (dict, op_errstr,
+                                                        rsp_dict);
+                        if (ret) {
+                                gf_log (this->name, GF_LOG_WARNING, "cleanup "
+                                        "operation failed");
+                                goto out;
+                        }
+                }
+                /* Irrespective of status of cleanup its better
+                 * to return from this function. As the functions
+                 * following this block is not required to be
+                 * executed in case of failure scenario.
+                 */
+                ret = 0;
+                goto out;
+        }
+
+        ret = dict_get_str (dict, "clonename", &clonename);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Unable to fetch "
+                        "clonename");
+                goto out;
+        }
+
+        ret = glusterd_volinfo_find (clonename, &snap_vol);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "unable to find clone "
+                        "%s volinfo", clonename);
+                goto out;
+        }
+
+        ret = glusterd_snapshot_update_snaps_post_validate (dict,
+                                                            op_errstr,
+                                                            rsp_dict);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                        "create snapshot");
+                goto out;
+        }
+        snap = snap_vol->snapshot;
+        ret = glusterd_snapobject_delete (snap);
+        if (ret) {
+                gf_log (this->name, GF_LOG_WARNING, "Failed to delete "
+                        "snap object %s", snap->snapname);
+                goto out;
+        }
+        snap_vol->snapshot = NULL;
+
+        ret = 0;
+
+out:
+        return ret;
+}
+
+
+int32_t
 glusterd_snapshot_create_postvalidate (dict_t *dict, int32_t op_ret,
                                        char **op_errstr, dict_t *rsp_dict)
 {
@@ -6902,6 +7460,7 @@ glusterd_snapshot (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
         }
 
         switch (snap_command) {
+
         case (GF_SNAP_OPTION_TYPE_CREATE):
                 ret = glusterd_snapshot_create_commit (dict, op_errstr,
                                                        rsp_dict);
@@ -6912,6 +7471,16 @@ glusterd_snapshot (dict_t *dict, char **op_errstr, dict_t *rsp_dict)
                 }
                 break;
 
+       case (GF_SNAP_OPTION_TYPE_CLONE):
+                ret = glusterd_snapshot_clone_commit (dict, op_errstr,
+                                                       rsp_dict);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_ERROR, "Failed to "
+                                "clone snapshot");
+                        goto out;
+                }
+                break;
+
         case GF_SNAP_OPTION_TYPE_CONFIG:
                 ret = glusterd_snapshot_config_commit (dict, op_errstr,
                                                        rsp_dict);
@@ -7138,6 +7707,17 @@ glusterd_snapshot_prevalidate (dict_t *dict, char **op_errstr,
                 }
                 break;
 
+        case (GF_SNAP_OPTION_TYPE_CLONE):
+                ret = glusterd_snapshot_clone_prevalidate (dict, op_errstr,
+                                                            rsp_dict);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_WARNING, "Snapshot clone "
+                                "pre-validation failed");
+                        goto out;
+                }
+                break;
+
+
         case (GF_SNAP_OPTION_TYPE_CONFIG):
                 ret = glusterd_snapshot_config_prevalidate (dict, op_errstr);
                 if (ret) {
@@ -7531,6 +8111,17 @@ glusterd_snapshot_postvalidate (dict_t *dict, int32_t op_ret, char **op_errstr,
                 }
                 glusterd_fetchsnap_notify (this);
                 break;
+         case GF_SNAP_OPTION_TYPE_CLONE:
+                ret = glusterd_snapshot_clone_postvalidate (dict, op_ret,
+                                                             op_errstr,
+                                                             rsp_dict);
+                if (ret) {
+                        gf_log (this->name, GF_LOG_WARNING, "Snapshot create "
+                                "post-validation failed");
+                        goto out;
+                }
+                glusterd_fetchsnap_notify (this);
+                break;
         case GF_SNAP_OPTION_TYPE_DELETE:
                 if (op_ret) {
                         gf_log (this->name, GF_LOG_DEBUG,
@@ -7727,6 +8318,16 @@ glusterd_handle_snapshot_fn (rpcsvc_request_t *req)
                                         "failed: %s", err_str);
                 }
                 break;
+
+        case GF_SNAP_OPTION_TYPE_CLONE:
+                ret = glusterd_handle_snapshot_clone (req, cli_op, dict,
+                                                err_str, sizeof (err_str));
+                if (ret) {
+                        gf_log (this->name, GF_LOG_WARNING, "Snapshot clone "
+                                        "failed: %s", err_str);
+                }
+                break;
+
         case GF_SNAP_OPTION_TYPE_RESTORE:
                 ret = glusterd_handle_snapshot_restore (req, cli_op, dict,
                                                 err_str, sizeof (err_str));