From 86058ec53c4bf9113291f4d13fdf61146a754db3 Mon Sep 17 00:00:00 2001
From: Pawan <pawan@mayadata.io>
Date: Tue, 10 Nov 2020 12:21:28 +0530
Subject: [PATCH] feat(clone): add support for creating the Clone from volume
 as datasource

Signed-off-by: Pawan <pawan@mayadata.io>
---
 changelogs/unreleased/234-pawanpraka1 |  1 +
 pkg/driver/controller.go              | 71 +++++++++++++++++++++++++--
 2 files changed, 69 insertions(+), 3 deletions(-)
 create mode 100644 changelogs/unreleased/234-pawanpraka1

diff --git a/changelogs/unreleased/234-pawanpraka1 b/changelogs/unreleased/234-pawanpraka1
new file mode 100644
index 000000000..a8bc44a28
--- /dev/null
+++ b/changelogs/unreleased/234-pawanpraka1
@@ -0,0 +1 @@
+add support for creating the Clone from volume as datasource
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 69c815551..0e306ac02 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -165,8 +165,70 @@ func CreateZFSVolume(req *csi.CreateVolumeRequest) (string, error) {
 	return selected, nil
 }
 
-// CreateZFSClone create a clone of zfs volume
-func CreateZFSClone(req *csi.CreateVolumeRequest, snapshot string) (string, error) {
+// CreateVolClone creates the clone from a volume
+func CreateVolClone(req *csi.CreateVolumeRequest, srcVol string) (string, error) {
+	volName := req.GetName()
+	parameters := req.GetParameters()
+	// lower case keys, cf CreateZFSVolume()
+	pool := helpers.GetInsensitiveParameter(&parameters, "poolname")
+	size := getRoundedCapacity(req.GetCapacityRange().RequiredBytes)
+	volsize := strconv.FormatInt(int64(size), 10)
+
+	vol, err := zfs.GetZFSVolume(srcVol)
+	if err != nil {
+		return "", status.Error(codes.NotFound, err.Error())
+	}
+
+	if vol.Spec.PoolName != pool {
+		return "", status.Errorf(codes.Internal,
+			"vol clone to a different pool src pool %s dst pool %s",
+			vol.Spec.PoolName, pool)
+	}
+
+	if vol.Spec.Capacity != volsize {
+		return "", status.Error(codes.Internal, "vol clone volume size is not matching")
+	}
+
+	selected := vol.Spec.OwnerNodeID
+
+	// create the snpashot
+
+	labels := map[string]string{zfs.ZFSVolKey: vol.Name}
+
+	snapObj, err := snapbuilder.NewBuilder().
+		WithName(volName).
+		WithLabels(labels).Build()
+
+	snapObj.Spec = vol.Spec
+	snapObj.Status.State = zfs.ZFSStatusPending
+
+	err = zfs.ProvisionSnapshot(snapObj)
+	if err != nil {
+		return "", status.Errorf(codes.Internal,
+			"clone: not able to provision the snapshot %s", err.Error())
+	}
+
+	// create the clone from the above snapshot
+
+	volObj, err := volbuilder.NewBuilder().
+		WithName(volName).
+		WithVolumeStatus(zfs.ZFSStatusPending).
+		Build()
+
+	volObj.Spec = vol.Spec
+	volObj.Spec.SnapName = volName
+
+	err = zfs.ProvisionVolume(volObj)
+	if err != nil {
+		return "", status.Errorf(codes.Internal,
+			"clone: not able to provision the volume %s", err.Error())
+	}
+
+	return selected, nil
+}
+
+// CreateSnapClone creates the clone from a snapshot
+func CreateSnapClone(req *csi.CreateVolumeRequest, snapshot string) (string, error) {
 
 	volName := req.GetName()
 	parameters := req.GetParameters()
@@ -243,7 +305,10 @@ func (cs *controller) CreateVolume(
 	if contentSource != nil && contentSource.GetSnapshot() != nil {
 		snapshotID := contentSource.GetSnapshot().GetSnapshotId()
 
-		selected, err = CreateZFSClone(req, snapshotID)
+		selected, err = CreateSnapClone(req, snapshotID)
+	} else if contentSource != nil && contentSource.GetVolume() != nil {
+		srcVol := contentSource.GetVolume().GetVolumeId()
+		selected, err = CreateVolClone(req, srcVol)
 	} else {
 		selected, err = CreateZFSVolume(req)
 	}