Add clone support for volume snapshot

- Add control on volume snapshot to take clone
- Add control on volume snapshot to delete it

Signed-off-by: Akash Srivastava <akash.srivastava@openebs.io>

probe/kubernetes/client.go

@@ -1,6 +1,7 @@
 package kubernetes
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"strings"
@@ -21,6 +22,7 @@ import (
 	apiextensionsv1beta1 "k8s.io/api/extensions/v1beta1"
 	storagev1 "k8s.io/api/storage/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -50,9 +52,11 @@ type Client interface {
 
 	WatchPods(f func(Event, Pod))
 
+	CloneVolumeSnapshot(namespaceID, volumeSnapshotID, persistentVolumeClaimID, capacity string) error
 	CreateVolumeSnapshot(namespaceID, persistentVolumeClaimID, capacity string) error
 	GetLogs(namespaceID, podID string, containerNames []string) (io.ReadCloser, error)
 	DeletePod(namespaceID, podID string) error
+	DeleteVolumeSnapshot(namespaceID, volumeSnapshotID string) error
 	ScaleUp(resource, namespaceID, id string) error
 	ScaleDown(resource, namespaceID, id string) error
 }
@@ -430,6 +434,68 @@ func (c *client) WalkVolumeSnapshotData(f func(VolumeSnapshotData) error) error
 	return nil
 }
 
+func (c *client) CloneVolumeSnapshot(namespaceID, volumeSnapshotID, persistentVolumeClaimID, capacity string) error {
+	var scName string
+	var claimSize string
+	UID := strings.Split(uuid.New(), "-")
+	scProvisionerName := "volumesnapshot.external-storage.k8s.io/snapshot-promoter"
+	scList, err := c.client.StorageV1().StorageClasses().List(metav1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	// Retrieve the first snapshot-promoter storage class
+	for _, sc := range scList.Items {
+		if sc.Provisioner == scProvisionerName {
+			scName = sc.Name
+			break
+		}
+	}
+	if scName == "" {
+		return errors.New("snapshot-promoter storage class is not present")
+	}
+	volumeSnapshot, _ := c.snapshotClient.VolumesnapshotV1().VolumeSnapshots(namespaceID).Get(volumeSnapshotID, metav1.GetOptions{})
+	if volumeSnapshot.Spec.PersistentVolumeClaimName != "" {
+		persistentVolumeClaim, err := c.client.CoreV1().PersistentVolumeClaims(namespaceID).Get(volumeSnapshot.Spec.PersistentVolumeClaimName, metav1.GetOptions{})
+		if err == nil {
+			storage := persistentVolumeClaim.Spec.Resources.Requests[apiv1.ResourceStorage]
+			if storage.String() != "" {
+				claimSize = storage.String()
+			}
+		}
+	}
+	// Set default volume size to the one stored in volume snapshot annotation,
+	// if unable to get PVC size.
+	if claimSize == "" {
+		claimSize = capacity
+	}
+
+	persistentVolumeClaim := &apiv1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "clone-" + persistentVolumeClaimID + "-" + UID[1],
+			Namespace: namespaceID,
+			Annotations: map[string]string{
+				"snapshot.alpha.kubernetes.io/snapshot": volumeSnapshotID,
+			},
+		},
+		Spec: apiv1.PersistentVolumeClaimSpec{
+			StorageClassName: &scName,
+			AccessModes: []apiv1.PersistentVolumeAccessMode{
+				apiv1.ReadWriteOnce,
+			},
+			Resources: apiv1.ResourceRequirements{
+				Requests: apiv1.ResourceList{
+					apiv1.ResourceName(apiv1.ResourceStorage): resource.MustParse(claimSize),
+				},
+			},
+		},
+	}
+	_, err = c.client.CoreV1().PersistentVolumeClaims(namespaceID).Create(persistentVolumeClaim)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (c *client) CreateVolumeSnapshot(namespaceID, persistentVolumeClaimID, capacity string) error {
 	UID := strings.Split(uuid.New(), "-")
 	volumeSnapshot := &snapshotv1.VolumeSnapshot{
@@ -479,6 +545,10 @@ func (c *client) DeletePod(namespaceID, podID string) error {
 	return c.client.CoreV1().Pods(namespaceID).Delete(podID, &metav1.DeleteOptions{})
 }
 
+func (c *client) DeleteVolumeSnapshot(namespaceID, volumeSnapshotID string) error {
+	return c.snapshotClient.VolumesnapshotV1().VolumeSnapshots(namespaceID).Delete(volumeSnapshotID, &metav1.DeleteOptions{})
+}
+
 func (c *client) ScaleUp(resource, namespaceID, id string) error {
 	return c.modifyScale(resource, namespaceID, id, func(scale *apiextensionsv1beta1.Scale) {
 		scale.Spec.Replicas++

probe/kubernetes/controls.go

@@ -11,9 +11,11 @@ import (
 
 // Control IDs used by the kubernetes integration.
 const (
+	CloneVolumeSnapshot  = report.KubernetesCloneVolumeSnapshot
 	CreateVolumeSnapshot = report.KubernetesCreateVolumeSnapshot
 	GetLogs              = report.KubernetesGetLogs
 	DeletePod            = report.KubernetesDeletePod
+	DeleteVolumeSnapshot = report.KubernetesDeleteVolumeSnapshot
 	ScaleUp              = report.KubernetesScaleUp
 	ScaleDown            = report.KubernetesScaleDown
 )
@@ -44,6 +46,14 @@ func (r *Reporter) GetLogs(req xfer.Request, namespaceID, podID string, containe
 	}
 }
 
+func (r *Reporter) cloneVolumeSnapshot(req xfer.Request, namespaceID, volumeSnapshotID, persistentVolumeClaimID, capacity string) xfer.Response {
+	err := r.client.CloneVolumeSnapshot(namespaceID, volumeSnapshotID, persistentVolumeClaimID, capacity)
+	if err != nil {
+		return xfer.ResponseError(err)
+	}
+	return xfer.Response{}
+}
+
 func (r *Reporter) createVolumeSnapshot(req xfer.Request, namespaceID, persistentVolumeClaimID, capacity string) xfer.Response {
 	err := r.client.CreateVolumeSnapshot(namespaceID, persistentVolumeClaimID, capacity)
 	if err != nil {
@@ -61,6 +71,15 @@ func (r *Reporter) deletePod(req xfer.Request, namespaceID, podID string, _ []st
 	}
 }
 
+func (r *Reporter) deleteVolumeSnapshot(req xfer.Request, namespaceID, volumeSnapshotID, _, _ string) xfer.Response {
+	if err := r.client.DeleteVolumeSnapshot(namespaceID, volumeSnapshotID); err != nil {
+		return xfer.ResponseError(err)
+	}
+	return xfer.Response{
+		RemovedNode: req.NodeID,
+	}
+}
+
 // CapturePod is exported for testing
 func (r *Reporter) CapturePod(f func(xfer.Request, string, string, []string) xfer.Response) func(xfer.Request) xfer.Response {
 	return func(req xfer.Request) xfer.Response {
@@ -126,6 +145,28 @@ func (r *Reporter) CapturePersistentVolumeClaim(f func(xfer.Request, string, str
 	}
 }
 
+// CaptureVolumeSnapshot will return name, pvc name, namespace and capacity of volume snapshot
+func (r *Reporter) CaptureVolumeSnapshot(f func(xfer.Request, string, string, string, string) xfer.Response) func(xfer.Request) xfer.Response {
+	return func(req xfer.Request) xfer.Response {
+		uid, ok := report.ParseVolumeSnapshotNodeID(req.NodeID)
+		if !ok {
+			return xfer.ResponseErrorf("Invalid ID: %s", req.NodeID)
+		}
+		// find volume snapshot by UID
+		var volumeSnapshot VolumeSnapshot
+		r.client.WalkVolumeSnapshots(func(p VolumeSnapshot) error {
+			if p.UID() == uid {
+				volumeSnapshot = p
+			}
+			return nil
+		})
+		if volumeSnapshot == nil {
+			return xfer.ResponseErrorf("Volume snapshot not found: %s", uid)
+		}
+		return f(req, volumeSnapshot.Namespace(), volumeSnapshot.Name(), volumeSnapshot.GetVolumeName(), volumeSnapshot.GetCapacity())
+	}
+}
+
 // ScaleUp is the control to scale up a deployment
 func (r *Reporter) ScaleUp(req xfer.Request, namespace, id string) xfer.Response {
 	return xfer.ResponseError(r.client.ScaleUp(report.Deployment, namespace, id))
@@ -138,9 +179,11 @@ func (r *Reporter) ScaleDown(req xfer.Request, namespace, id string) xfer.Respon
 
 func (r *Reporter) registerControls() {
 	controls := map[string]xfer.ControlHandlerFunc{
+		CloneVolumeSnapshot:  r.CaptureVolumeSnapshot(r.cloneVolumeSnapshot),
 		CreateVolumeSnapshot: r.CapturePersistentVolumeClaim(r.createVolumeSnapshot),
 		GetLogs:              r.CapturePod(r.GetLogs),
 		DeletePod:            r.CapturePod(r.deletePod),
+		DeleteVolumeSnapshot: r.CaptureVolumeSnapshot(r.deleteVolumeSnapshot),
 		ScaleUp:              r.CaptureDeployment(r.ScaleUp),
 		ScaleDown:            r.CaptureDeployment(r.ScaleDown),
 	}
@@ -149,9 +192,11 @@ func (r *Reporter) registerControls() {
 
 func (r *Reporter) deregisterControls() {
 	controls := []string{
+		CloneVolumeSnapshot,
 		CreateVolumeSnapshot,
 		GetLogs,
 		DeletePod,
+		DeleteVolumeSnapshot,
 		ScaleUp,
 		ScaleDown,
 	}

probe/kubernetes/reporter.go

@@ -497,6 +497,18 @@ func (r *Reporter) volumeSnapshotTopology() (report.Topology, []VolumeSnapshot,
 	result := report.MakeTopology().
 		WithMetadataTemplates(VolumeSnapshotMetadataTemplates).
 		WithTableTemplates(TableTemplates)
+	result.Controls.AddControl(report.Control{
+		ID:    CloneVolumeSnapshot,
+		Human: "Clone snapshot",
+		Icon:  "fa-clone",
+		Rank:  0,
+	})
+	result.Controls.AddControl(report.Control{
+		ID:    DeleteVolumeSnapshot,
+		Human: "Delete",
+		Icon:  "fa-trash-o",
+		Rank:  1,
+	})
 	err := r.client.WalkVolumeSnapshots(func(p VolumeSnapshot) error {
 		result.AddNode(p.GetNode(r.probeID))
 		volumeSnapshots = append(volumeSnapshots, p)

probe/kubernetes/reporter_test.go

@@ -187,9 +187,15 @@ func (c *mockClient) ScaleUp(resource, namespaceID, id string) error {
 func (c *mockClient) ScaleDown(resource, namespaceID, id string) error {
 	return nil
 }
+func (c *mockClient) CloneVolumeSnapshot(namespaceID, VolumeSnapshotID, persistentVolumeClaimID, capacity string) error {
+	return nil
+}
 func (c *mockClient) CreateVolumeSnapshot(namespaceID, persistentVolumeClaimID, capacity string) error {
 	return nil
 }
+func (c *mockClient) DeleteVolumeSnapshot(namespaceID, VolumeSnapshotID string) error {
+	return nil
+}
 
 type mockPipeClient map[string]xfer.Pipe
 

probe/kubernetes/volumesnapshot.go

@@ -10,10 +10,17 @@ const (
 	SnapshotPVName = "SnapshotMetadata-PVName"
 )
 
+// Capacity is the annotation key which provides the storage size
+const (
+	Capacity = "capacity"
+)
+
 // VolumeSnapshot represent kubernetes VolumeSnapshot interface
 type VolumeSnapshot interface {
 	Meta
 	GetNode(probeID string) report.Node
+	GetVolumeName() string
+	GetCapacity() string
 }
 
 // volumeSnapshot represents kubernetes volume snapshots
@@ -27,13 +34,27 @@ func NewVolumeSnapshot(p *snapshotv1.VolumeSnapshot) VolumeSnapshot {
 	return &volumeSnapshot{VolumeSnapshot: p, Meta: meta{p.ObjectMeta}}
 }
 
+// GetVolumeName returns the PVC name for volume snapshot
+func (p *volumeSnapshot) GetVolumeName() string {
+	return p.Spec.PersistentVolumeClaimName
+}
+
+// GetCapacity returns the capacity of the source PVC stored in annotation
+func (p *volumeSnapshot) GetCapacity() string {
+	capacity := p.GetAnnotations()[Capacity]
+	if capacity != "" {
+		return capacity
+	}
+	return ""
+}
+
 // GetNode returns VolumeSnapshot as Node
 func (p *volumeSnapshot) GetNode(probeID string) report.Node {
 	return p.MetaNode(report.MakeVolumeSnapshotNodeID(p.UID())).WithLatests(map[string]string{
 		report.ControlProbeID: probeID,
 		NodeType:              "Volume Snapshot",
-		VolumeClaim:           p.Spec.PersistentVolumeClaimName,
+		VolumeClaim:           p.GetVolumeName(),
 		SnapshotData:          p.Spec.SnapshotDataName,
 		VolumeName:            p.GetLabels()[SnapshotPVName],
-	})
+	}).WithLatestActiveControls(CloneVolumeSnapshot, DeleteVolumeSnapshot)
 }

report/map_keys.go

@@ -85,6 +85,8 @@ const (
 	KubernetesSnapshotData         = "kuberneets_snapshot_data"
 	KubernetesCreateVolumeSnapshot = "kubernetes_create_volume_snapshot"
 	KubernetesVolumeCapacity       = "kubernetes_volume_capacity"
+	KubernetesCloneVolumeSnapshot  = "kubernetes_clone_volume_snapshot"
+	KubernetesDeleteVolumeSnapshot = "kubernetes_delete_volume_snapshot"
 	// probe/awsecs
 	ECSCluster             = "ecs_cluster"
 	ECSCreatedAt           = "ecs_created_at"