Allow users to specify "storage-locations" for snapshots.

Use a comma separated string, for example "us-central1,us-west1"
as the "snapshot-locations" field of snapshot class parameters.
See the PR for detailed examples.

Author: TeweiLuo

pkg/common/parameters.go

@@ -22,11 +22,15 @@ import (
 )
 
 const (
+	// Parameters for StorageClass
 	ParameterKeyType                 = "type"
 	ParameterKeyReplicationType      = "replication-type"
 	ParameterKeyDiskEncryptionKmsKey = "disk-encryption-kms-key"
 	ParameterKeyLabels               = "labels"
 
+	// Parameters for VolumeSnapshotClass
+	ParameterStorageLocations = "storage-locations"
+
 	replicationTypeNone = "none"
 
 	// Keys for PV and PVC parameters as reported by external-provisioner

pkg/gce-cloud-provider/compute/fake-gce.go

@@ -387,7 +387,7 @@ func (cloud *FakeCloudProvider) GetSnapshot(ctx context.Context, project, snapsh
 	return snapshot, nil
 }
 
-func (cloud *FakeCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error) {
+func (cloud *FakeCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error) {
 	if snapshot, ok := cloud.snapshots[snapshotName]; ok {
 		return snapshot, nil
 	}
@@ -398,6 +398,7 @@ func (cloud *FakeCloudProvider) CreateSnapshot(ctx context.Context, project stri
 		CreationTimestamp: Timestamp,
 		Status:            "UPLOADING",
 		SelfLink:          cloud.getGlobalSnapshotURI(project, snapshotName),
+		StorageLocations:  strings.Split(storageLocations, ","),
 	}
 	switch volKey.Type() {
 	case meta.Zonal:
@@ -493,11 +494,11 @@ type FakeBlockingCloudProvider struct {
 // Upon starting a CreateSnapshot, it passes a chan 'executeCreateSnapshot' into readyToExecute, then blocks on executeCreateSnapshot.
 // The test calling this function can block on readyToExecute to ensure that the operation has started and
 // allowed the CreateSnapshot to continue by passing a struct into executeCreateSnapshot.
-func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error) {
+func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error) {
 	executeCreateSnapshot := make(chan struct{})
 	cloud.ReadyToExecute <- executeCreateSnapshot
 	<-executeCreateSnapshot
-	return cloud.FakeCloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName)
+	return cloud.FakeCloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, storageLocations)
 }
 
 func notFoundError() *googleapi.Error {

pkg/gce-cloud-provider/compute/gce-compute.go

@@ -73,7 +73,7 @@ type GCECompute interface {
 	ListZones(ctx context.Context, region string) ([]string, error)
 	ListSnapshots(ctx context.Context, filter string, maxEntries int64, pageToken string) ([]*computev1.Snapshot, string, error)
 	GetSnapshot(ctx context.Context, project, snapshotName string) (*computev1.Snapshot, error)
-	CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error)
+	CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error)
 	DeleteSnapshot(ctx context.Context, project, snapshotName string) error
 }
 
@@ -787,13 +787,13 @@ func (cloud *CloudProvider) DeleteSnapshot(ctx context.Context, project, snapsho
 	return nil
 }
 
-func (cloud *CloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error) {
+func (cloud *CloudProvider) CreateSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error) {
 	klog.V(5).Infof("Creating snapshot %s for volume %v", snapshotName, volKey)
 	switch volKey.Type() {
 	case meta.Zonal:
-		return cloud.createZonalDiskSnapshot(ctx, project, volKey, snapshotName)
+		return cloud.createZonalDiskSnapshot(ctx, project, volKey, snapshotName, storageLocations)
 	case meta.Regional:
-		return cloud.createRegionalDiskSnapshot(ctx, project, volKey, snapshotName)
+		return cloud.createRegionalDiskSnapshot(ctx, project, volKey, snapshotName, storageLocations)
 	default:
 		return nil, fmt.Errorf("could not create snapshot, key was neither zonal nor regional, instead got: %v", volKey.String())
 	}
@@ -864,9 +864,10 @@ func (cloud *CloudProvider) resizeRegionalDisk(ctx context.Context, project stri
 	return requestGb, nil
 }
 
-func (cloud *CloudProvider) createZonalDiskSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error) {
+func (cloud *CloudProvider) createZonalDiskSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error) {
 	snapshotToCreate := &computev1.Snapshot{
-		Name: snapshotName,
+		Name:             snapshotName,
+		StorageLocations: strings.Split(storageLocations, ","),
 	}
 
 	_, err := cloud.service.Disks.CreateSnapshot(project, volKey.Zone, volKey.Name, snapshotToCreate).Context(ctx).Do()
@@ -878,9 +879,10 @@ func (cloud *CloudProvider) createZonalDiskSnapshot(ctx context.Context, project
 	return cloud.waitForSnapshotCreation(ctx, project, snapshotName)
 }
 
-func (cloud *CloudProvider) createRegionalDiskSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string) (*computev1.Snapshot, error) {
+func (cloud *CloudProvider) createRegionalDiskSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, storageLocations string) (*computev1.Snapshot, error) {
 	snapshotToCreate := &computev1.Snapshot{
-		Name: snapshotName,
+		Name:             snapshotName,
+		StorageLocations: strings.Split(storageLocations, ","),
 	}
 
 	_, err := cloud.service.RegionDisks.CreateSnapshot(project, volKey.Region, volKey.Name, snapshotToCreate).Context(ctx).Do()

pkg/gce-pd-csi-driver/controller.go

@@ -609,7 +609,11 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown get snapshot error: %v", err))
 		}
 		// If we could not find the snapshot, we create a new one
-		snapshot, err = gceCS.CloudProvider.CreateSnapshot(ctx, project, volKey, req.Name)
+		parameters := make(map[string]string)
+		if req.GetParameters() != nil {
+			parameters = req.GetParameters()
+		}
+		snapshot, err = gceCS.CloudProvider.CreateSnapshot(ctx, project, volKey, req.Name, parameters[common.ParameterStorageLocations])
 		if err != nil {
 			if gce.IsGCEError(err, "notFound") {
 				return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volKey.String(), err))

pkg/gce-pd-csi-driver/controller_test.go

@@ -848,7 +848,11 @@ func TestCreateVolumeWithVolumeSource(t *testing.T) {
 		}
 
 		if tc.snapshotOnCloud {
-			gceDriver.cs.CloudProvider.CreateSnapshot(context.Background(), tc.project, tc.volKey, name)
+			parameters := make(map[string]string)
+			if req.GetParameters() != nil {
+				parameters = req.GetParameters()
+			}
+			gceDriver.cs.CloudProvider.CreateSnapshot(context.Background(), tc.project, tc.volKey, name, parameters[common.ParameterStorageLocations])
 		}
 		resp, err := gceDriver.cs.CreateVolume(context.Background(), req)
 		//check response