Merge pull request #123 from xing-yang/datasource2

Restore Volume from Snapshot

Author: k8s-ci-robot

Gopkg.lock

@@ -32,6 +32,10 @@
   branch = "master"
   name = "github.com/golang/glog"
 
+[[constraint]]
+  branch = "master"
+  name = "github.com/kubernetes-csi/external-snapshotter"
+
 [[constraint]]
   branch = "master"
   name = "github.com/kubernetes-incubator/external-storage"
@@ -42,12 +46,12 @@
 
 [[constraint]]
   name = "k8s.io/apimachinery"
-  version = "kubernetes-1.10.0-beta.1"
+  version = "kubernetes-1.12.0-alpha.0"
 
 [[constraint]]
   name = "k8s.io/client-go"
-  version = "v6.0.0"
+  version = "kubernetes-1.12.0-alpha.0"
 
 [[constraint]]
   name = "k8s.io/api"
-  version = "kubernetes-1.10.0-beta.1"
+  branch = "master"

Gopkg.toml

@@ -27,6 +27,7 @@ import (
 	"github.com/golang/glog"
 
 	ctrl "github.com/kubernetes-csi/external-provisioner/pkg/controller"
+	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
 
 	"google.golang.org/grpc"
@@ -86,6 +87,11 @@ func init() {
 	if err != nil {
 		glog.Fatalf("Failed to create client: %v", err)
 	}
+	// snapclientset.NewForConfig creates a new Clientset for VolumesnapshotV1alpha1Client
+	snapClient, err := snapclientset.NewForConfig(config)
+	if err != nil {
+		glog.Fatalf("Failed to create snapshot client: %v", err)
+	}
 
 	// The controller needs to know what the server version is because out-of-tree
 	// provisioners aren't officially supported until 1.5
@@ -112,7 +118,7 @@ func init() {
 	}
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
-	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient)
+	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient)
 	provisionController = controller.NewProvisionController(
 		clientset,
 		*provisioner,

cmd/csi-provisioner/csi-provisioner.go

@@ -32,6 +32,9 @@ import (
 
 	"github.com/kubernetes-incubator/external-storage/lib/controller"
 
+	snapapi "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1"
+	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
+
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -69,13 +72,17 @@ const (
 	backoffSteps    = 10
 
 	defaultFSType = "ext4"
+
+	snapshotKind     = "VolumeSnapshot"
+	snapshotAPIGroup = snapapi.GroupName // "snapshot.storage.k8s.io"
 )
 
 // CSIProvisioner struct
 type csiProvisioner struct {
 	client               kubernetes.Interface
 	csiClient            csi.ControllerClient
 	grpcClient           *grpc.ClientConn
+	snapshotClient       snapclientset.Interface
 	timeout              time.Duration
 	identity             string
 	volumeNamePrefix     string
@@ -213,20 +220,49 @@ func supportsControllerCreateVolume(conn *grpc.ClientConn, timeout time.Duration
 	return false, nil
 }
 
+func supportsControllerCreateSnapshot(conn *grpc.ClientConn, timeout time.Duration) (bool, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	client := csi.NewControllerClient(conn)
+	req := csi.ControllerGetCapabilitiesRequest{}
+
+	rsp, err := client.ControllerGetCapabilities(ctx, &req)
+	if err != nil {
+		return false, err
+	}
+	caps := rsp.GetCapabilities()
+	for _, cap := range caps {
+		if cap == nil {
+			continue
+		}
+		rpc := cap.GetRpc()
+		if rpc == nil {
+			continue
+		}
+		if rpc.GetType() == csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 // NewCSIProvisioner creates new CSI provisioner
 func NewCSIProvisioner(client kubernetes.Interface,
 	csiEndpoint string,
 	connectionTimeout time.Duration,
 	identity string,
 	volumeNamePrefix string,
 	volumeNameUUIDLength int,
-	grpcClient *grpc.ClientConn) controller.Provisioner {
+	grpcClient *grpc.ClientConn,
+	snapshotClient snapclientset.Interface) controller.Provisioner {
 
 	csiClient := csi.NewControllerClient(grpcClient)
 	provisioner := &csiProvisioner{
 		client:               client,
 		grpcClient:           grpcClient,
 		csiClient:            csiClient,
+		snapshotClient:       snapshotClient,
 		timeout:              connectionTimeout,
 		identity:             identity,
 		volumeNamePrefix:     volumeNamePrefix,
@@ -238,7 +274,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 // This function get called before any attepmt to communicate with the driver.
 // Before initiating Create/Delete API calls provisioner checks if Capabilities:
 // PluginControllerService,  ControllerCreateVolume sre supported and gets the  driver name.
-func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration) (string, error) {
+func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration, needSnapshotSupport bool) (string, error) {
 	ok, err := supportsPluginControllerService(grpcClient, timeout)
 	if err != nil {
 		glog.Errorf("failed to get support info :%v", err)
@@ -257,6 +293,24 @@ func checkDriverState(grpcClient *grpc.ClientConn, timeout time.Duration) (strin
 		glog.Error("no create/delete volume support detected")
 		return "", fmt.Errorf("no create/delete volume support detected")
 	}
+
+	// If PVC.Spec.DataSource is not nil, it indicates the request is to create volume
+	// from snapshot and therefore we should check for snapshot support;
+	// otherwise we don't need to check for snapshot support.
+	if needSnapshotSupport {
+		// Check whether plugin supports create snapshot
+		// If not, create volume from snapshot cannot proceed
+		ok, err = supportsControllerCreateSnapshot(grpcClient, timeout)
+		if err != nil {
+			glog.Errorf("failed to get support info :%v", err)
+			return "", err
+		}
+		if !ok {
+			glog.Error("no create/delete snapshot support detected. Cannot create volume from shapshot")
+			return "", fmt.Errorf("no create/delete snapshot support detected. Cannot create volume from shapshot")
+		}
+	}
+
 	driverName, err := getDriverName(grpcClient, timeout)
 	if err != nil {
 		glog.Errorf("failed to get driver info :%v", err)
@@ -289,7 +343,21 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 		return nil, fmt.Errorf("claim Selector is not supported")
 	}
 
-	driverName, err := checkDriverState(p.grpcClient, p.timeout)
+	var needSnapshotSupport bool = false
+	if options.PVC.Spec.DataSource != nil {
+		// PVC.Spec.DataSource.Name is the name of the VolumeSnapshot API object
+		if options.PVC.Spec.DataSource.Name == "" {
+			return nil, fmt.Errorf("the PVC source not found for PVC %s", options.PVC.Name)
+		}
+		if options.PVC.Spec.DataSource.Kind != snapshotKind {
+			return nil, fmt.Errorf("the PVC source is not the right type. Expected %s, Got %s", snapshotKind, options.PVC.Spec.DataSource.Kind)
+		}
+		if options.PVC.Spec.DataSource.APIGroup != snapshotAPIGroup {
+			return nil, fmt.Errorf("the PVC source does not belong to the right APIGroup. Expected %s, Got %s", snapshotAPIGroup, options.PVC.Spec.DataSource.APIGroup)
+		}
+		needSnapshotSupport = true
+	}
+	driverName, err := checkDriverState(p.grpcClient, p.timeout, needSnapshotSupport)
 	if err != nil {
 		return nil, err
 	}
@@ -317,6 +385,17 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 			RequiredBytes: int64(volSizeBytes),
 		},
 	}
+
+	if needSnapshotSupport {
+		volumeContentSource, err := p.getVolumeContentSource(options)
+		if err != nil {
+			return nil, fmt.Errorf("error getting snapshot handle for snapshot %s: %v", options.PVC.Spec.DataSource.Name, err)
+		}
+		req.VolumeContentSource = volumeContentSource
+	}
+
+	glog.V(5).Infof("CreateVolumeRequest %+v", req)
+
 	rep := &csi.CreateVolumeResponse{}
 
 	// Resolve provision secret credentials.
@@ -433,13 +512,65 @@ func (p *csiProvisioner) Provision(options controller.VolumeOptions) (*v1.Persis
 	return pv, nil
 }
 
+func (p *csiProvisioner) getVolumeContentSource(options controller.VolumeOptions) (*csi.VolumeContentSource, error) {
+	snapshotObj, err := p.snapshotClient.VolumesnapshotV1alpha1().VolumeSnapshots(options.PVC.Namespace).Get(options.PVC.Spec.DataSource.Name, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("error getting snapshot %s from api server: %v", options.PVC.Spec.DataSource.Name, err)
+	}
+	if snapshotObj.Status.Ready == false {
+		return nil, fmt.Errorf("snapshot %s is not Ready", options.PVC.Spec.DataSource.Name)
+	}
+
+	glog.V(5).Infof("VolumeSnapshot %+v", snapshotObj)
+
+	snapContentObj, err := p.snapshotClient.VolumesnapshotV1alpha1().VolumeSnapshotContents().Get(snapshotObj.Spec.SnapshotContentName, metav1.GetOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("error getting snapshot:snapshotcontent %s:%s from api server: %v", snapshotObj.Name, snapshotObj.Spec.SnapshotContentName, err)
+	}
+	glog.V(5).Infof("VolumeSnapshotContent %+v", snapContentObj)
+
+	if snapContentObj.Spec.VolumeSnapshotSource.CSI == nil {
+		return nil, fmt.Errorf("error getting snapshot source from snapshot:snapshotcontent %s:%s", snapshotObj.Name, snapshotObj.Spec.SnapshotContentName)
+	}
+
+	snapshotSource := csi.VolumeContentSource_Snapshot{
+		Snapshot: &csi.VolumeContentSource_SnapshotSource{
+			Id: snapContentObj.Spec.VolumeSnapshotSource.CSI.SnapshotHandle,
+		},
+	}
+	glog.V(5).Infof("VolumeContentSource_Snapshot %+v", snapshotSource)
+
+	if snapshotObj.Status.RestoreSize != nil {
+		capacity, exists := options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
+		if !exists {
+			return nil, fmt.Errorf("error getting capacity for PVC %s when creating snapshot %s", options.PVC.Name, snapshotObj.Name)
+		}
+		volSizeBytes := capacity.Value()
+		glog.V(5).Infof("Requested volume size is %d and snapshot size is %d for the source snapshot %s", int64(volSizeBytes), int64(snapshotObj.Status.RestoreSize.Value()), snapshotObj.Name)
+		// When restoring volume from a snapshot, the volume size should
+		// be equal to or larger than its snapshot size.
+		if int64(volSizeBytes) < int64(snapshotObj.Status.RestoreSize.Value()) {
+			return nil, fmt.Errorf("requested volume size %d is less than the size %d for the source snapshot %s", int64(volSizeBytes), int64(snapshotObj.Status.RestoreSize.Value()), snapshotObj.Name)
+		}
+		if int64(volSizeBytes) > int64(snapshotObj.Status.RestoreSize.Value()) {
+			glog.Warningf("requested volume size %d is greater than the size %d for the source snapshot %s. Volume plugin needs to handle volume expansion.", int64(volSizeBytes), int64(snapshotObj.Status.RestoreSize.Value()), snapshotObj.Name)
+		}
+	}
+
+	volumeContentSource := &csi.VolumeContentSource{
+		Type: &snapshotSource,
+	}
+
+	return volumeContentSource, nil
+}
+
 func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {
 	if volume == nil || volume.Spec.CSI == nil {
 		return fmt.Errorf("invalid CSI PV")
 	}
 	volumeId := p.volumeHandleToId(volume.Spec.CSI.VolumeHandle)
 
-	_, err := checkDriverState(p.grpcClient, p.timeout)
+	_, err := checkDriverState(p.grpcClient, p.timeout, false)
 	if err != nil {
 		return err
 	}