Add disk image snapshot support

pkg/common/parameters.go

@@ -30,8 +30,11 @@ const (
 
 	// Parameters for VolumeSnapshotClass
 	ParameterKeyStorageLocations = "storage-locations"
-
-	replicationTypeNone = "none"
+	ParameterKeySnapshotType     = "snapshot-type"
+	ParameterKeyImageFamily      = "image-family"
+	DiskSnapshotType             = "snapshots"
+	DiskImageType                = "images"
+	replicationTypeNone          = "none"
 
 	// Keys for PV and PVC parameters as reported by external-provisioner
 	ParameterKeyPVCName      = "csi.storage.k8s.io/pvc/name"
@@ -67,6 +70,8 @@ type DiskParameters struct {
 // SnapshotParameters contains normalized and defaulted parameters for snapshots
 type SnapshotParameters struct {
 	StorageLocations []string
+	SnapshotType     string
+	ImageFamily      string
 }
 
 // ExtractAndDefaultParameters will take the relevant parameters from a map and
@@ -131,6 +136,7 @@ func ExtractAndDefaultParameters(parameters map[string]string, driverName string
 func ExtractAndDefaultSnapshotParameters(parameters map[string]string) (SnapshotParameters, error) {
 	p := SnapshotParameters{
 		StorageLocations: []string{},
+		SnapshotType:     DiskSnapshotType,
 	}
 	for k, v := range parameters {
 		switch strings.ToLower(k) {
@@ -140,6 +146,14 @@ func ExtractAndDefaultSnapshotParameters(parameters map[string]string) (Snapshot
 				return p, err
 			}
 			p.StorageLocations = normalizedStorageLocations
+		case ParameterKeySnapshotType:
+			err := ValidateSnapshotType(v)
+			if err != nil {
+				return p, err
+			}
+			p.SnapshotType = v
+		case ParameterKeyImageFamily:
+			p.ImageFamily = v
 		default:
 			return p, fmt.Errorf("parameters contains invalid option %q", k)
 		}

pkg/common/parameters_test.go

@@ -165,37 +165,50 @@ func TestExtractAndDefaultParameters(t *testing.T) {
 	}
 }
 
-// Currently the only parameter is storage-locations, which is already tested
-// in utils_test/TestSnapshotStorageLocations. Here we just test the case where
-// no parameter is set in the snapshot class.
+// Currently the storage-locations parameter is tested in utils_test/TestSnapshotStorageLocations.
+// Here we just test other parameters.
 func TestSnapshotParameters(t *testing.T) {
 	tests := []struct {
 		desc                    string
 		parameters              map[string]string
 		expectedSnapshotParames SnapshotParameters
+		expectError             bool
 	}{
 		{
 			desc:       "valid parameter",
-			parameters: map[string]string{ParameterKeyStorageLocations: "ASIA "},
+			parameters: map[string]string{ParameterKeyStorageLocations: "ASIA ", ParameterKeySnapshotType: "images", ParameterKeyImageFamily: "test-family"},
 			expectedSnapshotParames: SnapshotParameters{
 				StorageLocations: []string{"asia"},
+				SnapshotType:     DiskImageType,
+				ImageFamily:      "test-family",
 			},
+			expectError: false,
 		},
 		{
 			desc:       "nil parameter",
 			parameters: nil,
 			expectedSnapshotParames: SnapshotParameters{
 				StorageLocations: []string{},
+				SnapshotType:     DiskSnapshotType,
 			},
+			expectError: false,
+		},
+		{
+			desc:        "invalid snapshot type",
+			parameters:  map[string]string{ParameterKeySnapshotType: "invalid-type"},
+			expectError: true,
 		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.desc, func(t *testing.T) {
 			p, err := ExtractAndDefaultSnapshotParameters(tc.parameters)
-			if err != nil {
-				t.Errorf("Got error processing snapshot parameters: %v; expect no error", err)
+			if err != nil && !tc.expectError {
+				t.Errorf("Got error %v; expect no error", err)
+			}
+			if err == nil && tc.expectError {
+				t.Error("Got no error; expect an error")
 			}
-			if !reflect.DeepEqual(p, tc.expectedSnapshotParames) {
+			if err == nil && !reflect.DeepEqual(p, tc.expectedSnapshotParames) {
 				t.Errorf("Got ExtractAndDefaultSnapshotParameters(%+v) = %+v; expect %+v", tc.parameters, p, tc.expectedSnapshotParames)
 			}
 		})

pkg/common/utils.go

@@ -114,15 +114,15 @@ func GenerateUnderspecifiedVolumeID(diskName string, isZonal bool) string {
 	return fmt.Sprintf(volIDRegionalFmt, UnspecifiedValue, UnspecifiedValue, diskName)
 }
 
-func SnapshotIDToProjectKey(id string) (string, string, error) {
+func SnapshotIDToProjectKey(id string) (string, string, string, error) {
 	splitId := strings.Split(id, "/")
 	if len(splitId) != snapshotTotalElements {
-		return "", "", fmt.Errorf("failed to get id components. Expected projects/{project}/global/snapshot/{name}. Got: %s", id)
+		return "", "", "", fmt.Errorf("failed to get id components. Expected projects/{project}/global/{snapshots|images}/{name}. Got: %s", id)
 	}
 	if splitId[snapshotTopologyKey] == "global" {
-		return splitId[snapshotProjectKey], splitId[snapshotTotalElements-1], nil
+		return splitId[snapshotProjectKey], splitId[snapshotTotalElements-2], splitId[snapshotTotalElements-1], nil
 	} else {
-		return "", "", fmt.Errorf("could not get id components, expected global, got: %v", splitId[snapshotTopologyKey])
+		return "", "", "", fmt.Errorf("could not get id components, expected global, got: %v", splitId[snapshotTopologyKey])
 	}
 }
 
@@ -238,3 +238,13 @@ func ProcessStorageLocations(storageLocations string) ([]string, error) {
 	}
 	return []string{normalizedLoc}, nil
 }
+
+// ValidateSnapshotType validates the type
+func ValidateSnapshotType(snapshotType string) error {
+	switch snapshotType {
+	case DiskSnapshotType, DiskImageType:
+		return nil
+	default:
+		return fmt.Errorf("invalid snapshot type %s", snapshotType)
+	}
+}

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

@@ -17,7 +17,6 @@ package gcecloudprovider
 import (
 	"context"
 	"fmt"
-	"strconv"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
@@ -37,6 +36,7 @@ const (
 	Timestamp                 = "2018-09-05T15:17:08.270-07:00"
 	BasePath                  = "https://www.googleapis.com/compute/v1/projects/"
 	snapshotURITemplateGlobal = "%s/global/snapshots/%s" //{gce.projectID}/global/snapshots/{snapshot.Name}"
+	imageURITemplateGlobal    = "%s/global/images/%s"    //{gce.projectID}/global/images/{image.Name}"
 )
 
 type FakeCloudProvider struct {
@@ -47,6 +47,7 @@ type FakeCloudProvider struct {
 	pageTokens map[string]sets.String
 	instances  map[string]*computev1.Instance
 	snapshots  map[string]*computev1.Snapshot
+	images     map[string]*computev1.Image
 
 	// marker to set disk status during InsertDisk operation.
 	mockDiskStatus string
@@ -61,6 +62,7 @@ func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*Fa
 		disks:      map[string]*CloudDisk{},
 		instances:  map[string]*computev1.Instance{},
 		snapshots:  map[string]*computev1.Snapshot{},
+		images:     map[string]*computev1.Image{},
 		pageTokens: map[string]sets.String{},
 		// A newly created disk is marked READY by default.
 		mockDiskStatus: "READY",
@@ -122,7 +124,7 @@ func (cloud *FakeCloudProvider) ListDisks(ctx context.Context) ([]*computev1.Dis
 	return d, "", nil
 }
 
-func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string, maxEntries int64, pageToken string) ([]*computev1.Snapshot, string, error) {
+func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string) ([]*computev1.Snapshot, string, error) {
 	var sourceDisk string
 	snapshots := []*computev1.Snapshot{}
 	if len(filter) > 0 {
@@ -141,45 +143,7 @@ func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string
 		snapshots = append(snapshots, snapshot)
 	}
 
-	var (
-		ulenSnapshots = len(snapshots)
-		startingToken int
-	)
-
-	if len(pageToken) > 0 {
-		i, err := strconv.ParseUint(pageToken, 10, 32)
-		if err != nil {
-			return nil, "", invalidError()
-		}
-		startingToken = int(i)
-	}
-
-	if startingToken > ulenSnapshots {
-		return nil, "", invalidError()
-	}
-
-	// Discern the number of remaining entries.
-	rem := ulenSnapshots - startingToken
-
-	// If maxEntries is 0 or greater than the number of remaining entries then
-	// set maxEntries to the number of remaining entries.
-	max := int(maxEntries)
-	if max == 0 || max > rem {
-		max = rem
-	}
-
-	results := []*computev1.Snapshot{}
-	j := startingToken
-	for i := 0; i < max; i++ {
-		results = append(results, snapshots[j])
-		j++
-	}
-
-	var nextToken string
-	if j < ulenSnapshots {
-		nextToken = fmt.Sprintf("%d", j)
-	}
-	return results, nextToken, nil
+	return snapshots, "", nil
 }
 
 // Disk Methods
@@ -238,6 +202,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
 		Type:             cloud.GetDiskTypeURI(project, volKey, params.DiskType),
 		SourceSnapshotId: snapshotID,
 		SourceDiskId:     volumeContentSourceVolumeID,
+		SourceImageId:    snapshotID,
 		Status:           cloud.mockDiskStatus,
 		Labels:           params.Labels,
 	}
@@ -399,6 +364,71 @@ func (cloud *FakeCloudProvider) DeleteSnapshot(ctx context.Context, project, sna
 	return nil
 }
 
+func (cloud *FakeCloudProvider) ListImages(ctx context.Context, filter string) ([]*computev1.Image, string, error) {
+	var sourceDisk string
+	images := []*computev1.Image{}
+	if len(filter) > 0 {
+		filterSplits := strings.Fields(filter)
+		if len(filterSplits) != 3 || filterSplits[0] != "sourceDisk" {
+			return nil, "", invalidError()
+		}
+		sourceDisk = filterSplits[2]
+	}
+	for _, image := range cloud.images {
+		if len(sourceDisk) > 0 {
+			if image.SourceDisk == sourceDisk {
+				continue
+			}
+		}
+		images = append(images, image)
+	}
+
+	return images, "", nil
+}
+
+func (cloud *FakeCloudProvider) GetImage(ctx context.Context, project, imageName string) (*computev1.Image, error) {
+	image, ok := cloud.images[imageName]
+	if !ok {
+		return nil, notFoundError()
+	}
+	image.Status = "READY"
+	return image, nil
+}
+
+func (cloud *FakeCloudProvider) CreateImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*computev1.Image, error) {
+	if image, ok := cloud.images[imageName]; ok {
+		return image, nil
+	}
+
+	imageToCreate := &computev1.Image{
+		CreationTimestamp: Timestamp,
+		DiskSizeGb:        int64(DiskSizeGb),
+		Family:            snapshotParams.ImageFamily,
+		Name:              imageName,
+		SelfLink:          cloud.getGlobalImageURI(project, imageName),
+		SourceType:        "RAW",
+		Status:            "PENDING",
+		StorageLocations:  snapshotParams.StorageLocations,
+	}
+
+	switch volKey.Type() {
+	case meta.Zonal:
+		imageToCreate.SourceDisk = cloud.getZonalDiskSourceURI(project, volKey.Name, volKey.Zone)
+	case meta.Regional:
+		imageToCreate.SourceDisk = cloud.getRegionalDiskSourceURI(project, volKey.Name, volKey.Region)
+	default:
+		return nil, fmt.Errorf("could not create image, disk key was neither zonal nor regional, instead got: %v", volKey.String())
+	}
+
+	cloud.images[imageName] = imageToCreate
+	return imageToCreate, nil
+}
+
+func (cloud *FakeCloudProvider) DeleteImage(ctx context.Context, project, imageName string) error {
+	delete(cloud.images, imageName)
+	return nil
+}
+
 func (cloud *FakeCloudProvider) ValidateExistingSnapshot(resp *computev1.Snapshot, volKey *meta.Key) error {
 	if resp == nil {
 		return fmt.Errorf("disk does not exist")
@@ -447,6 +477,13 @@ func (cloud *FakeCloudProvider) getGlobalSnapshotURI(project, snapshotName strin
 		snapshotName)
 }
 
+func (cloud *FakeCloudProvider) getGlobalImageURI(project, imageName string) string {
+	return BasePath + fmt.Sprintf(
+		imageURITemplateGlobal,
+		project,
+		imageName)
+}
+
 func (cloud *FakeCloudProvider) UpdateDiskStatus(s string) {
 	cloud.mockDiskStatus = s
 }
@@ -467,6 +504,13 @@ func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, proj
 	return cloud.FakeCloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, snapshotParams)
 }
 
+func (cloud *FakeBlockingCloudProvider) CreateImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*computev1.Image, error) {
+	executeCreateSnapshot := make(chan struct{})
+	cloud.ReadyToExecute <- executeCreateSnapshot
+	<-executeCreateSnapshot
+	return cloud.FakeCloudProvider.CreateImage(ctx, project, volKey, imageName, snapshotParams)
+}
+
 func notFoundError() *googleapi.Error {
 	return &googleapi.Error{
 		Errors: []googleapi.ErrorItem{