Snapshot Volume for AWS EBS storage (#4361)

* Snapshot Volume for AWS EBS storage

* address review suggestions

* address review suggestions

* Address review comments and add unit tests

* Add function for profile validation

* fix compilation error

* add unit tests

glide.yaml

@@ -23,6 +23,8 @@ import:
   vcs: git
   version: rm-k8s-dep
   repo: https://github.com/kastenhq/operator-kit.git
+- package: github.com/satori/go.uuid
+  version: v1.2.0  
 - package: github.com/sirupsen/logrus
   version: v1.0.6
 - package: github.com/spf13/cobra

pkg/function/create_volume_from_snapshot.go

@@ -6,9 +6,13 @@ import (
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
 	kanister "github.com/kanisterio/kanister/pkg"
+	"github.com/kanisterio/kanister/pkg/blockstorage"
+	"github.com/kanisterio/kanister/pkg/blockstorage/awsebs"
+	"github.com/kanisterio/kanister/pkg/blockstorage/getter"
 	"github.com/kanisterio/kanister/pkg/kube"
 	"github.com/kanisterio/kanister/pkg/param"
 )
@@ -32,43 +36,63 @@ func (*createVolumeFromSnapshotFunc) Name() string {
 	return "CreateVolumeFromSnapshot"
 }
 
-func createVolumeFromSnapshot(ctx context.Context, cli kubernetes.Interface, namespace, snapshotinfo string, profile *param.Profile) error {
+func createVolumeFromSnapshot(ctx context.Context, cli kubernetes.Interface, namespace, snapshotinfo string, profile *param.Profile, getter getter.Getter) (map[string]blockstorage.Provider, error) {
 	PVCData := []VolumeSnapshotInfo{}
 	err := json.Unmarshal([]byte(snapshotinfo), &PVCData)
 	if err != nil {
-		return errors.Wrapf(err, "Could not decode JSON data")
+		return nil, errors.Wrapf(err, "Could not decode JSON data")
 	}
+	// providerList required for unit testing
+	providerList := make(map[string]blockstorage.Provider)
 	for _, pvcInfo := range PVCData {
-		var storageType string
-		switch pvcInfo.StorageType {
-		// TODO: use constants once blockstorage is moved to kanister repo
-		case "EBS":
-			storageType = "EBS"
-		case "GPD":
-			storageType = "GPD"
-		case "AD":
-			storageType = "AD"
-		case "Cinder":
-			storageType = "Cinder"
-		case "Ceph":
-			storageType = "Ceph"
-		default:
-			return errors.Errorf("Storage type %s not supported!", pvcInfo.StorageType)
+		config := make(map[string]string)
+		switch pvcInfo.Type {
+		case blockstorage.TypeEBS:
+			if err = ValidateProfile(profile); err != nil {
+				return nil, errors.Wrap(err, "Profile validation failed")
+			}
+			config[awsebs.ConfigRegion] = pvcInfo.Region
+			config[awsebs.AccessKeyID] = profile.Credential.KeyPair.ID
+			config[awsebs.SecretAccessKey] = profile.Credential.KeyPair.Secret
 		}
-		log.Infof("snapshotId: %s, StorageType: %s, region: %s", pvcInfo.SnapshotID, storageType, pvcInfo.Region)
-		if err := createPVCFromSnapshot(); err != nil {
-			return errors.Wrapf(err, "Could not create PVC")
+		provider, err := getter.Get(pvcInfo.Type, config)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Could not get storage provider %v", pvcInfo.Type)
+		}
+		_, err = cli.Core().PersistentVolumeClaims(namespace).Get(pvcInfo.PVCName, metav1.GetOptions{})
+		if err == nil {
+			if err = kube.DeletePVC(cli, namespace, pvcInfo.PVCName); err != nil {
+				return nil, err
+			}
+		}
+		snapshot, err := provider.SnapshotGet(ctx, pvcInfo.SnapshotID)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Failed to get Snapshot from Provider")
+		}
+		tags := map[string]string{
+			"pvcname": pvcInfo.PVCName,
+		}
+		snapshot.Volume.VolumeType = pvcInfo.VolumeType
+		snapshot.Volume.Az = pvcInfo.Az
+		snapshot.Volume.Tags = pvcInfo.Tags
+		vol, err := provider.VolumeCreateFromSnapshot(ctx, *snapshot, tags)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Failed to create volume from snapshot, snapID: %s", snapshot.ID)
 		}
-	}
-	return nil
-}
-
-func createPVCFromSnapshot() error {
-	return errors.Wrapf(createPV(), "Could not create PV")
-}
 
-func createPV() error {
-	return nil
+		annotations := map[string]string{}
+		pvc, err := kube.CreatePVC(ctx, cli, namespace, pvcInfo.PVCName, vol.Size, vol.ID, annotations)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Unable to create PVC for volume %v", *vol)
+		}
+		pv, err := kube.CreatePV(ctx, cli, vol, vol.Type, annotations)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Unable to create PV for volume %v", *vol)
+		}
+		log.Infof("Restore/Create volume from snapshot completed for pvc: %s, volume: %s", pvc, pv)
+		providerList[pvcInfo.PVCName] = provider
+	}
+	return providerList, nil
 }
 
 func (kef *createVolumeFromSnapshotFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
@@ -83,7 +107,8 @@ func (kef *createVolumeFromSnapshotFunc) Exec(ctx context.Context, tp param.Temp
 	if err = Arg(args, CreateVolumeFromSnapshotManifestArg, &snapshotinfo); err != nil {
 		return nil, err
 	}
-	return nil, createVolumeFromSnapshot(ctx, cli, namespace, snapshotinfo, tp.Profile)
+	_, err = createVolumeFromSnapshot(ctx, cli, namespace, snapshotinfo, tp.Profile, getter.New())
+	return nil, err
 }
 
 func (*createVolumeFromSnapshotFunc) RequiredArgs() []string {

pkg/function/create_volume_from_snapshot_test.go

@@ -0,0 +1,108 @@
+package function
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	. "gopkg.in/check.v1"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/testing"
+
+	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
+	"github.com/kanisterio/kanister/pkg/blockstorage"
+	"github.com/kanisterio/kanister/pkg/param"
+	"github.com/kanisterio/kanister/pkg/testutil/mockblockstorage"
+)
+
+type CreateVolumeFromSnapshotTestSuite struct{}
+
+var _ = Suite(&CreateVolumeFromSnapshotTestSuite{})
+
+func (s *CreateVolumeFromSnapshotTestSuite) TestCreateVolumeFromSnapshot(c *C) {
+	ctx := context.Background()
+	ns := "ns"
+	mockGetter := mockblockstorage.NewGetter()
+	profile := &param.Profile{
+		Location: crv1alpha1.Location{
+			Type: crv1alpha1.LocationTypeS3Compliant,
+			S3Compliant: &crv1alpha1.S3CompliantLocation{
+				Region: "us-west-2"},
+		},
+		Credential: param.Credential{
+			Type: param.CredentialTypeKeyPair,
+			KeyPair: &param.KeyPair{
+				ID:     "foo",
+				Secret: "bar",
+			},
+		},
+	}
+	cli := fake.NewSimpleClientset()
+	// fake doesn't handle generated names for PVs, so ...
+	var i int
+	pvl := &v1.PersistentVolumeList{}
+	// kube.CreatePV() calls create() and list() which is to be handled for fake client
+	cli.PrependReactor("create", "persistentvolumes",
+		func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+			ca := action.(testing.CreateAction)
+			pv := ca.GetObject().(*v1.PersistentVolume)
+			pvl.Items = append(pvl.Items, *pv)
+			if pv.ObjectMeta.Name == "" && pv.ObjectMeta.GenerateName != "" {
+				pv.ObjectMeta.Name = fmt.Sprintf("%s%d", pv.ObjectMeta.GenerateName, i)
+				i++
+				return true, pv, nil
+			}
+			return false, nil, nil
+		})
+	cli.PrependReactor("list", "persistentvolumes",
+		func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+			return true, pvl, nil
+		})
+	tags := []*blockstorage.KeyValue{
+		{Key: "testkey", Value: "testval"},
+	}
+	volInfo1 := VolumeSnapshotInfo{SnapshotID: "snap-1", Type: blockstorage.TypeEBS, Region: "us-west-2", PVCName: "pvc-1", Az: "us-west-2a", Tags: tags, VolumeType: "ssd"}
+	volInfo2 := VolumeSnapshotInfo{SnapshotID: "snap-2", Type: blockstorage.TypeEBS, Region: "us-west-2", PVCName: "pvc-2", Az: "us-west-2a", Tags: tags, VolumeType: "ssd"}
+	var PVCData1 []VolumeSnapshotInfo
+	PVCData1 = append(PVCData1, volInfo1)
+	PVCData1 = append(PVCData1, volInfo2)
+	info, err := json.Marshal(PVCData1)
+	c.Assert(err, IsNil)
+	snapinfo := string(info)
+	for _, tc := range []struct {
+		snapshotinfo string
+		check        Checker
+	}{
+		{
+			snapshotinfo: snapinfo,
+			check:        IsNil,
+		},
+	} {
+		providerList, err := createVolumeFromSnapshot(ctx, cli, ns, tc.snapshotinfo, profile, mockGetter)
+		c.Assert(providerList, Not(Equals), tc.check)
+		c.Assert(err, tc.check)
+		if err != nil {
+			continue
+		}
+		c.Assert(len(providerList) == 2, Equals, true)
+		provider, ok := providerList["pvc-1"]
+		c.Assert(ok, Equals, true)
+		c.Assert(len(provider.(*mockblockstorage.Provider).SnapIDList) == 1, Equals, true)
+		c.Assert(mockblockstorage.CheckID("snap-1", provider.(*mockblockstorage.Provider).SnapIDList), Equals, true)
+		c.Assert(len(provider.(*mockblockstorage.Provider).VolIDList) == 1, Equals, true)
+
+		provider, ok = providerList["pvc-2"]
+		c.Assert(ok, Equals, true)
+		c.Assert(len(provider.(*mockblockstorage.Provider).SnapIDList) == 1, Equals, true)
+		c.Assert(mockblockstorage.CheckID("snap-2", provider.(*mockblockstorage.Provider).SnapIDList), Equals, true)
+		c.Assert(len(provider.(*mockblockstorage.Provider).VolIDList) == 1, Equals, true)
+
+		_, err = cli.Core().PersistentVolumeClaims(ns).Get("pvc-1", metav1.GetOptions{})
+		c.Assert(err, IsNil)
+		_, err = cli.Core().PersistentVolumeClaims(ns).Get("pvc-2", metav1.GetOptions{})
+		c.Assert(err, IsNil)
+	}
+}