cinder-csi: Adds support for managing backups (kubernetes#2473)

docs/cinder-csi-plugin/using-cinder-csi-plugin.md

@@ -254,6 +254,7 @@ helm install --namespace kube-system --name cinder-csi ./charts/cinder-csi-plugi
 | StorageClass `parameters`  | `availability`          | `nova`          | String. Volume Availability Zone |
 | StorageClass `parameters`  | `type`                  | Empty String    | String. Name/ID of Volume type. Corresponding volume type should exist in cinder     |
 | VolumeSnapshotClass `parameters` | `force-create`    | `false`         | Enable to support creating snapshot for a volume in in-use status |
+| VolumeSnapshotClass `parameters` | `type`            | Empty String    | `snapshot` creates a VolumeSnapshot object linked to a Cinder volume snapshot. `backup` creates a VolumeSnapshot object linked to a cinder volume backup. Defaults to `snapshot` if not defined |
 | Inline Volume `volumeAttributes`   | `capacity`              | `1Gi`       | volume size for creating inline volumes| 
 | Inline Volume `VolumeAttributes`   | `type`              | Empty String  | Name/ID of Volume type. Corresponding volume type should exist in cinder |
 

pkg/csi/cinder/controllerserver.go

@@ -21,6 +21,7 @@ import (
 	"strconv"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
@@ -112,15 +113,29 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	content := req.GetVolumeContentSource()
 	var snapshotID string
 	var sourcevolID string
+	var sourcebackupID string
 
 	if content != nil && content.GetSnapshot() != nil {
 		snapshotID = content.GetSnapshot().GetSnapshotId()
-		_, err := cloud.GetSnapshotByID(snapshotID)
+		snapshotError, err := GetSnapshot(cloud, snapshotID)
 		if err != nil {
-			if cpoerrors.IsNotFound(err) {
-				return nil, status.Errorf(codes.NotFound, "VolumeContentSource Snapshot %s not found", snapshotID)
-			}
-			return nil, status.Errorf(codes.Internal, "Failed to retrieve the snapshot %s: %v", snapshotID, err)
+			return nil, err
+		}
+		// Look for a Backup with the same ID as the snapshot.
+		BackupError, err := GetBackup(cloud, snapshotID)
+		if err != nil {
+			return nil, err
+		}
+
+		// If Backup and snapshot not found
+		if snapshotError != nil && BackupError != nil {
+			return nil, status.Errorf(codes.NotFound, "VolumeContentSource Snapshot or Backup %s not found", snapshotID)
+		}
+
+		// Case: Backup found, Snapshot not found. Create Volume from Backup.
+		if BackupError == nil && snapshotError != nil {
+			sourcebackupID = snapshotID
+			snapshotID = ""
 		}
 	}
 
@@ -135,7 +150,11 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		}
 	}
 
-	vol, err := cloud.CreateVolume(volName, volSizeGB, volType, volAvailability, snapshotID, sourcevolID, &properties)
+	vol, err := cloud.CreateVolume(volName, volSizeGB, volType, volAvailability, snapshotID, sourcevolID, sourcebackupID, &properties)
+	// When creating a volume from a backup, the response does not include the backupID.
+	if sourcebackupID != "" {
+		vol.BackupID = &sourcebackupID
+	}
 
 	if err != nil {
 		klog.Errorf("Failed to CreateVolume: %v", err)
@@ -148,6 +167,38 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	return getCreateVolumeResponse(vol, ignoreVolumeAZ, req.GetAccessibilityRequirements()), nil
 }
 
+func GetSnapshot(cloud openstack.IOpenStack, snapshotID string) (error, error) {
+	snap, errsnap := cloud.GetSnapshotByID(snapshotID)
+	// Case: Snapshot exists but is not yet available
+	if errsnap == nil && snap.Status != "available" {
+		return errsnap, status.Errorf(codes.Internal, "Snapshot is not yet available. Status: %s", snap.Status)
+	}
+	// Case: General error while getting the Snapshot
+	if errsnap != nil && !cpoerrors.IsNotFound(errsnap) {
+		return errsnap, status.Errorf(codes.Internal, "Failed to retrieve the snapshot %s: %v", snapshotID, errsnap)
+	}
+
+	return errsnap, nil
+}
+
+func GetBackup(cloud openstack.IOpenStack, snapshotID string) (error, error) {
+	// Backup exists but is not yet available.
+	back, errback := cloud.GetBackupByID(snapshotID)
+	if errback == nil && back.Status != "available" {
+		return errback, status.Errorf(codes.Internal, "Backup is not currently available. Status: %s", back.Status)
+	}
+	// Case: Error while getting Backup.
+	// 		 Do not return error because some openstack deployments do not support backups
+	if errback != nil {
+		if cpoerrors.IsNotFound(errback) {
+			klog.Errorf("VolumeContentSource Backup %s not found", snapshotID)
+		} else {
+			klog.Errorf("Failed to retrieve the backup %s: %v", snapshotID, errback)
+		}
+	}
+	return errback, nil
+}
+
 func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
 	klog.V(4).Infof("DeleteVolume: called with args %+v", protosanitizer.StripSecrets(*req))
 
@@ -328,6 +379,12 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 
 	name := req.Name
 	volumeID := req.GetSourceVolumeId()
+	snapshotType, snapshotTypeDeclared := req.Parameters[openstack.SnapshotType]
+
+	// Set snapshot type to 'snapshot' by default
+	if !snapshotTypeDeclared {
+		snapshotType = "snapshot"
+	}
 
 	if name == "" {
 		return nil, status.Error(codes.InvalidArgument, "Snapshot name must be provided in CreateSnapshot request")
@@ -337,6 +394,11 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		return nil, status.Error(codes.InvalidArgument, "VolumeID must be provided in CreateSnapshot request")
 	}
 
+	// Verify snapshot type has a valid value
+	if snapshotType != "snapshot" && snapshotType != "backup" && snapshotType != "" {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot type must be 'backup', 'snapshot' or not defined")
+	}
+
 	// Verify a snapshot with the provided name doesn't already exist for this tenant
 	var snap *snapshots.Snapshot
 	filters := map[string]string{}
@@ -347,6 +409,14 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		return nil, status.Error(codes.Internal, "Failed to get snapshots")
 	}
 
+	// Verify a backup with the provided name doesn't already exist for this tenant
+	var backup *backups.Backup
+	backups, err := cs.Cloud.ListBackups(filters)
+	if err != nil {
+		klog.Errorf("Failed to query for existing Backup during CreateBackups: %v", err)
+		return nil, status.Error(codes.Internal, "Failed to get backups")
+	}
+
 	if len(snapshots) == 1 {
 		snap = &snapshots[0]
 
@@ -360,6 +430,18 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		klog.Errorf("found multiple existing snapshots with selected name (%s) during create", name)
 		return nil, status.Error(codes.Internal, "Multiple snapshots reported by Cinder with same name")
 
+	} else if len(backups) == 1 && snapshotType == "backup" {
+		backup = &backups[0]
+
+		if backup.VolumeID != volumeID {
+			return nil, status.Error(codes.AlreadyExists, "Backup with given name already exists, with different source volume ID")
+		}
+		klog.V(3).Infof("Found existing backup %s from volume with ID: %s", name, volumeID)
+
+	} else if len(backups) > 1 && snapshotType == "backup" {
+		klog.Errorf("found multiple existing backups with selected name (%s) during create", name)
+		return nil, status.Error(codes.Internal, "Multiple backups reported by Cinder with same name")
+
 	} else {
 		// Add cluster ID to the snapshot metadata
 		properties := map[string]string{cinderCSIClusterIDKey: cs.Driver.cluster}
@@ -394,6 +476,61 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		klog.Errorf("Failed to WaitSnapshotReady: %v", err)
 		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateSnapshot failed with error %v", err))
 	}
+	// If snapshotType is 'backup', create a backup from the snapshot and delete the snapshot.
+	if snapshotType == "backup" {
+		// Add cluster ID to the snapshot metadata
+		properties := map[string]string{cinderCSIClusterIDKey: cs.Driver.cluster}
+
+		// see https://github.com/kubernetes-csi/external-snapshotter/pull/375/
+		// Also, we don't want to tag every param but we still want to send the
+		// 'force-create' flag to openstack layer so that we will honor the
+		// force create functions
+		for _, mKey := range []string{"csi.storage.k8s.io/volumesnapshot/name", "csi.storage.k8s.io/volumesnapshot/namespace", "csi.storage.k8s.io/volumesnapshotcontent/name", openstack.SnapshotForceCreate, openstack.SnapshotType} {
+			if v, ok := req.Parameters[mKey]; ok {
+				properties[mKey] = v
+			}
+		}
+		backup, err = cs.Cloud.CreateBackup(name, volumeID, snap.ID, &properties)
+		if err != nil {
+			klog.Errorf("Failed to Create backup: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateBackup failed with error %v", err))
+		}
+		klog.V(4).Infof("Backup created: %+v", backup)
+
+		ctime = timestamppb.New(backup.CreatedAt)
+		if err := ctime.CheckValid(); err != nil {
+			klog.Errorf("Error to convert time to timestamp: %v", err)
+		}
+
+		err = cs.Cloud.WaitBackupReady(backup.ID)
+		if err != nil {
+			klog.Errorf("Failed to WaitBackupReady: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateBackup failed with error %v", err))
+		}
+
+		err = cs.Cloud.DeleteSnapshot(snap.ID)
+		if err != nil {
+			klog.Errorf("Failed to DeleteSnapshot: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("DeleteSnapshot failed with error %v", err))
+		}
+
+		// Necessary to get all the backup information, including size.
+		backup, err = cs.Cloud.GetBackupByID(backup.ID)
+		if err != nil {
+			klog.Errorf("Failed to GetBackupByID after backup creation: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("GetBackupByID failed with error %v", err))
+		}
+
+		return &csi.CreateSnapshotResponse{
+			Snapshot: &csi.Snapshot{
+				SnapshotId:     backup.ID,
+				SizeBytes:      int64(backup.Size * 1024 * 1024 * 1024),
+				SourceVolumeId: backup.VolumeID,
+				CreationTime:   ctime,
+				ReadyToUse:     true,
+			},
+		}, nil
+	}
 
 	return &csi.CreateSnapshotResponse{
 		Snapshot: &csi.Snapshot{
@@ -415,8 +552,23 @@ func (cs *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 		return nil, status.Error(codes.InvalidArgument, "Snapshot ID must be provided in DeleteSnapshot request")
 	}
 
+	// If volumeSnapshot object was linked to a cinder backup, delete the backup.
+	back, err := cs.Cloud.GetBackupByID(id)
+	if err == nil && back != nil {
+		err = cs.Cloud.DeleteBackup(id)
+		if err != nil {
+			if cpoerrors.IsNotFound(err) {
+				klog.V(3).Infof("Backup %s is already deleted.", id)
+				return &csi.DeleteSnapshotResponse{}, nil
+			}
+			klog.Errorf("Failed to Delete backup: %v", err)
+			return nil, status.Error(codes.Internal, fmt.Sprintf("DeleteBackup failed with error %v", err))
+		}
+		return &csi.DeleteSnapshotResponse{}, nil
+	}
+
 	// Delegate the check to openstack itself
-	err := cs.Cloud.DeleteSnapshot(id)
+	err = cs.Cloud.DeleteSnapshot(id)
 	if err != nil {
 		if cpoerrors.IsNotFound(err) {
 			klog.V(3).Infof("Snapshot %s is already deleted.", id)
@@ -677,6 +829,16 @@ func getCreateVolumeResponse(vol *volumes.Volume, ignoreVolumeAZ bool, accessibl
 		}
 	}
 
+	if vol.BackupID != nil && *vol.BackupID != "" {
+		volsrc = &csi.VolumeContentSource{
+			Type: &csi.VolumeContentSource_Snapshot{
+				Snapshot: &csi.VolumeContentSource_SnapshotSource{
+					SnapshotId: *vol.BackupID,
+				},
+			},
+		}
+	}
+
 	var accessibleTopology []*csi.Topology
 	// If ignore-volume-az is true , dont set the accessible topology to volume az,
 	// use from preferred topologies instead.

pkg/csi/cinder/controllerserver_test.go

@@ -45,8 +45,8 @@ func TestCreateVolume(t *testing.T) {
 
 	// mock OpenStack
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", &properties).Return(&FakeVol, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, sourcebackupID string, tags *map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", "", &properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 	// Init assert
@@ -92,9 +92,9 @@ func TestCreateVolumeWithParam(t *testing.T) {
 
 	// mock OpenStack
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, sourcebackupID string, tags *map[string]string) (string, string, int, error)
 	// Vol type and availability comes from CreateVolumeRequest.Parameters
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), "dummyVolType", "cinder", "", "", &properties).Return(&FakeVol, nil)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), "dummyVolType", "cinder", "", "", "", &properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 	// Init assert
@@ -149,8 +149,8 @@ func TestCreateVolumeWithExtraMetadata(t *testing.T) {
 		"csi.storage.k8s.io/pvc/name":      FakePVCName,
 		"csi.storage.k8s.io/pvc/namespace": FakePVCNamespace,
 	}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", &properties).Return(&FakeVol, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, sourcebackupID string, tags *map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, FakeAvailability, "", "", "", &properties).Return(&FakeVol, nil)
 
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
@@ -190,8 +190,8 @@ func TestCreateVolumeWithExtraMetadata(t *testing.T) {
 func TestCreateVolumeFromSnapshot(t *testing.T) {
 
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", FakeSnapshotID, "", &properties).Return(&FakeVolFromSnapshot, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, sourcebackupID string, tags *map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", FakeSnapshotID, "", "", &properties).Return(&FakeVolFromSnapshot, nil)
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
 	// Init assert
@@ -238,8 +238,8 @@ func TestCreateVolumeFromSnapshot(t *testing.T) {
 func TestCreateVolumeFromSourceVolume(t *testing.T) {
 
 	properties := map[string]string{"cinder.csi.openstack.org/cluster": FakeCluster}
-	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, tags *map[string]string) (string, string, int, error)
-	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", "", FakeVolID, &properties).Return(&FakeVolFromSourceVolume, nil)
+	// CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, sourcebackupID string, tags *map[string]string) (string, string, int, error)
+	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", "", FakeVolID, "", &properties).Return(&FakeVolFromSourceVolume, nil)
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 
 	// Init assert
@@ -465,6 +465,7 @@ func TestCreateSnapshot(t *testing.T) {
 	osmock.On("CreateSnapshot", FakeSnapshotName, FakeVolID, &map[string]string{cinderCSIClusterIDKey: "cluster"}).Return(&FakeSnapshotRes, nil)
 	osmock.On("ListSnapshots", map[string]string{"Name": FakeSnapshotName}).Return(FakeSnapshotListEmpty, "", nil)
 	osmock.On("WaitSnapshotReady", FakeSnapshotID).Return(nil)
+	osmock.On("ListBackups", map[string]string{"Name": FakeSnapshotName}).Return(FakeBackupListEmpty, nil)
 
 	// Init assert
 	assert := assert.New(t)
@@ -535,6 +536,7 @@ func TestDeleteSnapshot(t *testing.T) {
 
 	// DeleteSnapshot(volumeID string) error
 	osmock.On("DeleteSnapshot", FakeSnapshotID).Return(nil)
+	osmock.On("DeleteBackup", FakeSnapshotID).Return(nil)
 
 	// Init assert
 	assert := assert.New(t)

pkg/csi/cinder/fake.go

@@ -17,6 +17,7 @@ limitations under the License.
 package cinder
 
 import (
+	"github.com/gophercloud/gophercloud/openstack/blockstorage/extensions/backups"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/snapshots"
 	"github.com/gophercloud/gophercloud/openstack/blockstorage/v3/volumes"
 	"golang.org/x/net/context"
@@ -102,6 +103,7 @@ var FakeVolListMultiple = []volumes.Volume{FakeVol1, FakeVol3}
 var FakeVolList = []volumes.Volume{FakeVol1}
 var FakeVolListEmpty = []volumes.Volume{}
 var FakeSnapshotListEmpty = []snapshots.Snapshot{}
+var FakeBackupListEmpty = []backups.Backup{}
 
 var FakeInstanceID = "321a8b81-3660-43e5-bab8-6470b65ee4e8"
 

pkg/csi/cinder/nodeserver.go

@@ -150,7 +150,7 @@ func nodePublishEphemeral(req *csi.NodePublishVolumeRequest, ns *nodeServer) (*c
 		volumeType = ""
 	}
 
-	evol, err := ns.Cloud.CreateVolume(volName, size, volumeType, volAvailability, "", "", &properties)
+	evol, err := ns.Cloud.CreateVolume(volName, size, volumeType, volAvailability, "", "", "", &properties)
 
 	if err != nil {
 		klog.V(3).Infof("Failed to Create Ephemeral Volume: %v", err)

pkg/csi/cinder/nodeserver_test.go

@@ -129,7 +129,7 @@ func TestNodePublishVolumeEphermeral(t *testing.T) {
 	fvolName := fmt.Sprintf("ephemeral-%s", FakeVolID)
 	tState := []string{"available"}
 
-	omock.On("CreateVolume", fvolName, 2, "test", "nova", "", "", &properties).Return(&FakeVol, nil)
+	omock.On("CreateVolume", fvolName, 2, "test", "nova", "", "", "", &properties).Return(&FakeVol, nil)
 
 	omock.On("AttachVolume", FakeNodeID, FakeVolID).Return(FakeVolID, nil)
 	omock.On("WaitDiskAttached", FakeNodeID, FakeVolID).Return(nil)