Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #926 (2022-03-10 08:42:37)

Add disk image snapshot support
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**

/kind feature

**What this PR does / why we need it**:
fixes
**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/issues/924

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->

```release-note
Add parameters to VolumeSnapshotClass for disk image config.
```


linux-foundation-easycla[bot]:  <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/20677499/136977103/926/#/?version=2"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-not-signed.svg" alt="CLA Not Signed" align="left" height="28" width="328"></a><br/><br /><ul><li>:x: login: luohao / name: Hao Luo  (8113c53045582feee89ce1fddba8c9da7117b3c2). This user is authorized, but they must confirm their affiliation with their company. Start the authorization process <a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/20677499/136977103/926/#/?version=2' target='_blank'> by clicking here</a>, click "Corporate",select the appropriate company from the list, then confirm your affiliation on the page that appears. For further assistance with EasyCLA, <a href='https://jira.linuxfoundation.org/servicedesk/customer/portal/4' target='_blank'>please submit a support request ticket</a>.</li></ul>
k8s-ci-robot: Welcome @luohao! <br><br>It looks like this is your first PR to <a href='https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver'>kubernetes-sigs/gcp-compute-persistent-disk-csi-driver</a> 🎉. Please refer to our [pull request process documentation](https://git.k8s.io/community/contributors/guide/pull-requests.md) to help your PR have a smooth ride to approval. <br><br>You will be prompted by a bot to use commands during the review process. Do not be afraid to follow the prompts! It is okay to experiment. [Here is the bot commands documentation](https://go.k8s.io/bot-commands). <br><br>You can also check if kubernetes-sigs/gcp-compute-persistent-disk-csi-driver has [its own contribution guidelines](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/tree/master/CONTRIBUTING.md). <br><br>You may want to refer to our [testing guide](https://git.k8s.io/community/contributors/devel/sig-testing/testing.md) if you run into trouble with your tests not passing. <br><br>If you are having difficulty getting your pull request seen, please follow the [recommended escalation practices](https://github.com/kubernetes/community/blob/master/contributors/guide/pull-requests.md#why-is-my-pull-request-not-getting-reviewed). Also, for tips and tricks in the contribution process you may want to read the [Kubernetes contributor cheat sheet](https://git.k8s.io/community/contributors/guide/contributor-cheatsheet/README.md). We want to make sure your contribution gets all the attention it needs! <br><br>Thank you, and welcome to Kubernetes. :smiley:
k8s-ci-robot: Hi @luohao. Thanks for your PR.

I'm waiting for a [kubernetes-sigs](https://github.com/orgs/kubernetes-sigs/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-csi-driver).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/pull/926#" title="Author self-approved">luohao</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **saikat-royc** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @saikat-royc` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-compute-persistent-disk-csi-driver).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes-sigs/gcp-compute-persistent-disk-csi-driver/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["saikat-royc"]} -->
mattcary: /ok-to-test
mattcary: /assign @amacaskill 

Alexis, can you look at this? You're most familiar with our volume cloning code.

Note that Han & I agreed offline about removing the paging for listing snapshots (& disk images). We ended up doing a similar thing in the filestore driver. The consumer of ListSnapshots just concatenates all the pages together anyway, so it makes it much easier to combine the disk & snapshot lists if we don't try to maintain the paging too.
amacaskill: Looks good to me! I just had two small nits to fix. /approve

Patch:
diff --git a/deploy/kubernetes/images/prow-gke-release-staging-rc-master/image.yaml b/deploy/kubernetes/images/prow-gke-release-staging-rc-master/image.yaml
index bbb5ba52..a60bd5bb 100644
--- a/deploy/kubernetes/images/prow-gke-release-staging-rc-master/image.yaml
+++ b/deploy/kubernetes/images/prow-gke-release-staging-rc-master/image.yaml
@@ -47,7 +47,7 @@ metadata:
   name: imagetag-csi-gce-driver-prow-rc
 imageTag:
   name: k8s.gcr.io/cloud-provider-gcp/gcp-compute-persistent-disk-csi-driver
-  newName: k8s.gcr.io/cloud-provider-gcp/gcp-compute-persistent-disk-csi-driver
+  newName: gcr.io/k8s-staging-cloud-provider-gcp/gcp-compute-persistent-disk-csi-driver
   newTag: "v1.4.0"
 ---
 
diff --git a/docs/kubernetes/user-guides/snapshots.md b/docs/kubernetes/user-guides/snapshots.md
index bebe2780..0791a163 100644
--- a/docs/kubernetes/user-guides/snapshots.md
+++ b/docs/kubernetes/user-guides/snapshots.md
@@ -149,8 +149,6 @@ elsewhere in GCP.
   1. Create a `VolumeSnapshot` resource which will be bound to a pre-provisioned
      `VolumeSnapshotContents`. Note this is called `restored-snapshot`; this
      name can be changed, but do it consistently across the other resources.
-     `VolumeSnapshotContentName` field must reference to the
-     VolumeSnapshotContent's name for the bidirectional binding to be valid.
      ```console
      kubectl apply -f - <<EOF
        apiVersion: snapshot.storage.k8s.io/v1beta1
@@ -160,24 +158,22 @@ elsewhere in GCP.
        spec:
          volumeSnapshotClassName: csi-gce-pd-snapshot-class
          source:
-           volumeSnapshotContentName: restored-snapshot-content
+           volumeSnapshotContentName: snapcontent-migrated
      EOF
      ```
   1. Create a `VolumeSnapshotContents` pointing to your existing PD
-     snapshot from the first step. Both `volumeSnapshotRef.name` and
-     `volumeSnapshotRef.namespace` must point to the previously created
-     VolumeSnapshot for the bidirectional binding to be valid.
+     snapshot from the first step.
      ```console
      kubectl apply -f - <<EOF
        apiVersion: snapshot.storage.k8s.io/v1beta1
-       kind: VolumeSnapshotContent
+       kind: VolumeSnapshotContents
        metadata:
          name: restored-snapshot-content
        spec:
          deletionPolicy: Retain
          driver: pd.csi.storage.gke.io
          source:
-           snapshotHandle: projects/$PROJECT_ID/global/snapshots/$SNAPSHOT_NAME
+           snapshotHandle: projects/$PROJECT_ID/global/snapshots/$SNAPSHOT_ID
          volumeSnapshotRef:
            kind: VolumeSnapshot
            name: restored-snapshot
diff --git a/go.mod b/go.mod
index 482f229e..f04744f5 100644
--- a/go.mod
+++ b/go.mod
@@ -30,6 +30,7 @@ require (
 )
 
 require (
+	github.com/GoogleCloudPlatform/guest-configs v0.0.0-20211028171426-a0dacef88470 // indirect
 	github.com/Microsoft/go-winio v0.4.16 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver v3.5.1+incompatible // indirect
diff --git a/go.sum b/go.sum
index 6ec1f4b3..3169992e 100644
--- a/go.sum
+++ b/go.sum
@@ -64,6 +64,8 @@ github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZ
 github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
+github.com/GoogleCloudPlatform/guest-configs v0.0.0-20211028171426-a0dacef88470 h1:A+CGaqTTg3TSo8k4v90qMInYSZ52kJrnnHUlsCgHRVM=
+github.com/GoogleCloudPlatform/guest-configs v0.0.0-20211028171426-a0dacef88470/go.mod h1:PYz+owYRovKCp6mKRumuKgk77Ac9B5UNmlMHRlW0B1w=
 github.com/GoogleCloudPlatform/k8s-cloud-provider v0.0.0-20200415212048-7901bc822317 h1:JhyuWIqYrstW7KHMjk/fTqU0xtMpBOHuiTA2FVc7L4E=
 github.com/GoogleCloudPlatform/k8s-cloud-provider v0.0.0-20200415212048-7901bc822317/go.mod h1:DF8FZRxMHMGv/vP2lQP6h+dYzzjpuRn24VeRiYn3qjQ=
 github.com/GoogleCloudPlatform/testgrid v0.0.1-alpha.3/go.mod h1:f96W2HYy3tiBNV5zbbRc+NczwYHgG1PHXMQfoEWv680=
diff --git a/pkg/common/parameters.go b/pkg/common/parameters.go
index 8ef4e3ed..ea36f627 100644
--- a/pkg/common/parameters.go
+++ b/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)
 		}
diff --git a/pkg/common/utils.go b/pkg/common/utils.go
index 6b71feb7..d5842ea3 100644
--- a/pkg/common/utils.go
+++ b/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)
+	}
+}
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index 6a7adeff..dd995316 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/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{
diff --git a/pkg/gce-cloud-provider/compute/gce-compute.go b/pkg/gce-cloud-provider/compute/gce-compute.go
index f8949950..8d1818c5 100644
--- a/pkg/gce-cloud-provider/compute/gce-compute.go
+++ b/pkg/gce-cloud-provider/compute/gce-compute.go
@@ -35,6 +35,7 @@ import (
 const (
 	operationStatusDone            = "DONE"
 	waitForSnapshotCreationTimeOut = 2 * time.Minute
+	waitForImageCreationTimeOut    = 5 * time.Minute
 	diskKind                       = "compute#disk"
 	cryptoKeyVerDelimiter          = "/cryptoKeyVersions"
 )
@@ -71,10 +72,14 @@ type GCECompute interface {
 	GetInstanceOrError(ctx context.Context, instanceZone, instanceName string) (*computev1.Instance, error)
 	// Zone Methods
 	ListZones(ctx context.Context, region string) ([]string, error)
-	ListSnapshots(ctx context.Context, filter string, maxEntries int64, pageToken string) ([]*computev1.Snapshot, string, error)
+	ListSnapshots(ctx context.Context, filter 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, snapshotParams common.SnapshotParameters) (*computev1.Snapshot, error)
 	DeleteSnapshot(ctx context.Context, project, snapshotName string) error
+	ListImages(ctx context.Context, filter string) ([]*computev1.Image, string, error)
+	GetImage(ctx context.Context, project, imageName string) (*computev1.Image, error)
+	CreateImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*computev1.Image, error)
+	DeleteImage(ctx context.Context, project, imageName string) error
 }
 
 // GetDefaultProject returns the project that was used to instantiate this GCE client.
@@ -189,18 +194,19 @@ func (cloud *CloudProvider) ListZones(ctx context.Context, region string) ([]str
 
 }
 
-func (cloud *CloudProvider) ListSnapshots(ctx context.Context, filter string, maxEntries int64, pageToken string) ([]*computev1.Snapshot, string, error) {
-	klog.V(5).Infof("Listing snapshots with filter: %s, max entries: %v, page token: %s", filter, maxEntries, pageToken)
-	snapshots := []*computev1.Snapshot{}
-	snapshotList, err := cloud.service.Snapshots.List(cloud.project).Filter(filter).MaxResults(maxEntries).PageToken(pageToken).Do()
-	if err != nil {
-		return snapshots, "", err
-	}
-	for _, snapshot := range snapshotList.Items {
-		snapshots = append(snapshots, snapshot)
+func (cloud *CloudProvider) ListSnapshots(ctx context.Context, filter string) ([]*computev1.Snapshot, string, error) {
+	klog.V(5).Infof("Listing snapshots with filter: %s", filter)
+	items := []*computev1.Snapshot{}
+	lCall := cloud.service.Snapshots.List(cloud.project).Filter(filter)
+	nextPageToken := "pageToken"
+	for nextPageToken != "" {
+		snapshotList, err := lCall.Do()
+		if err != nil {
+			return nil, "", err
+		}
+		items = append(items, snapshotList.Items...)
 	}
-	return snapshots, snapshotList.NextPageToken, nil
-
+	return items, "", nil
 }
 
 func (cloud *CloudProvider) GetDisk(ctx context.Context, project string, key *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error) {
@@ -398,11 +404,23 @@ func (cloud *CloudProvider) insertRegionalDisk(
 		Labels:      params.Labels,
 	}
 	if snapshotID != "" {
-		diskToCreate.SourceSnapshot = snapshotID
+		_, snapshotType, _, err := common.SnapshotIDToProjectKey(snapshotID)
+		if err != nil {
+			return err
+		}
+		switch snapshotType {
+		case common.DiskSnapshotType:
+			diskToCreate.SourceSnapshot = snapshotID
+		case common.DiskImageType:
+			diskToCreate.SourceImage = snapshotID
+		default:
+			return fmt.Errorf("invalid snapshot type in snapshot ID: %s", snapshotType)
+		}
 	}
 	if volumeContentSourceVolumeID != "" {
 		diskToCreate.SourceDisk = volumeContentSourceVolumeID
 	}
+
 	if len(replicaZones) != 0 {
 		diskToCreate.ReplicaZones = replicaZones
 	}
@@ -499,7 +517,18 @@ func (cloud *CloudProvider) insertZonalDisk(
 	}
 
 	if snapshotID != "" {
-		diskToCreate.SourceSnapshot = snapshotID
+		_, snapshotType, _, err := common.SnapshotIDToProjectKey(snapshotID)
+		if err != nil {
+			return err
+		}
+		switch snapshotType {
+		case common.DiskSnapshotType:
+			diskToCreate.SourceSnapshot = snapshotID
+		case common.DiskImageType:
+			diskToCreate.SourceImage = snapshotID
+		default:
+			return fmt.Errorf("invalid snapshot type in snapshot ID: %s", snapshotType)
+		}
 	}
 	if volumeContentSourceVolumeID != "" {
 		diskToCreate.SourceDisk = volumeContentSourceVolumeID
@@ -829,6 +858,94 @@ func (cloud *CloudProvider) CreateSnapshot(ctx context.Context, project string,
 	}
 }
 
+func (cloud *CloudProvider) CreateImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*computev1.Image, error) {
+	klog.V(5).Infof("Creating image %s for source %v", imageName, volKey)
+	diskID, err := common.KeyToVolumeID(volKey, project)
+	if err != nil {
+		return nil, err
+	}
+	image := &computev1.Image{
+		SourceDisk:       diskID,
+		Family:           snapshotParams.ImageFamily,
+		Name:             imageName,
+		StorageLocations: snapshotParams.StorageLocations,
+	}
+
+	_, err = cloud.service.Images.Insert(project, image).Context(ctx).Do()
+	if err != nil {
+		return nil, err
+	}
+
+	return cloud.waitForImageCreation(ctx, project, imageName)
+}
+
+func (cloud *CloudProvider) waitForImageCreation(ctx context.Context, project, imageName string) (*computev1.Image, error) {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	timer := time.NewTimer(waitForImageCreationTimeOut)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			klog.V(6).Infof("Checking GCE Image %s.", imageName)
+			image, err := cloud.GetImage(ctx, project, imageName)
+			if err != nil {
+				klog.Warningf("Error in getting image %s, %v", imageName, err)
+			} else if image != nil {
+				if image.Status != "PENDING" {
+					klog.V(6).Infof("Image %s status is %s", imageName, image.Status)
+					return image, nil
+				} else {
+					klog.V(6).Infof("Image %s is still pending", imageName)
+				}
+			}
+		case <-timer.C:
+			return nil, fmt.Errorf("timeout waiting for image %s to be created", imageName)
+		}
+	}
+}
+
+func (cloud *CloudProvider) GetImage(ctx context.Context, project, imageName string) (*computev1.Image, error) {
+	klog.V(5).Infof("Getting image %v", imageName)
+	image, err := cloud.service.Images.Get(project, imageName).Context(ctx).Do()
+	if err != nil {
+		return nil, err
+	}
+	return image, nil
+}
+
+func (cloud *CloudProvider) ListImages(ctx context.Context, filter string) ([]*computev1.Image, string, error) {
+	klog.V(5).Infof("Listing images with filter: %s", filter)
+	var items []*computev1.Image
+	lCall := cloud.service.Images.List(cloud.project).Context(ctx).Filter(filter)
+	nextPageToken := "pageToken"
+	for nextPageToken != "" {
+		imageList, err := lCall.Do()
+		if err != nil {
+			return nil, "", err
+		}
+		items = append(items, imageList.Items...)
+	}
+	return items, "", nil
+}
+
+func (cloud *CloudProvider) DeleteImage(ctx context.Context, project, imageName string) error {
+	klog.V(5).Infof("Deleting image %v", imageName)
+	op, err := cloud.service.Images.Delete(cloud.project, imageName).Context(ctx).Do()
+	if err != nil {
+		if IsGCEError(err, "notFound") {
+			return nil
+		}
+		return err
+	}
+	err = cloud.waitForGlobalOp(ctx, project, op.Name)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // ResizeDisk takes in the requested disk size in bytes and returns the resized
 // size in Gi
 // TODO(#461) The whole driver could benefit from standardized usage of the
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 2ffa96e8..705ac41e 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -25,6 +25,7 @@ import (
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/golang/protobuf/ptypes"
+	"github.com/golang/protobuf/ptypes/timestamp"
 	compute "google.golang.org/api/compute/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -45,6 +46,9 @@ type GCEControllerServer struct {
 	disks []*compute.Disk
 	seen  map[string]int
 
+	snapshots      []*csi.ListSnapshotsResponse_Entry
+	snapshotTokens map[string]int
+
 	// A map storing all volumes with ongoing operations so that additional
 	// operations for that same volume (as defined by Volume Key) return an
 	// Aborted error
@@ -807,19 +811,46 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateSnapshot unknown get disk error: %v", err))
 	}
 
-	// Check if snapshot already exists
+	snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(req.GetParameters())
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid snapshot parameters: %v", err))
+	}
+
+	var snapshot *csi.Snapshot
+	switch snapshotParams.SnapshotType {
+	case common.DiskSnapshotType:
+		snapshot, err = gceCS.createPDSnapshot(ctx, project, volKey, req.Name, snapshotParams)
+		if err != nil {
+			return nil, err
+		}
+	case common.DiskImageType:
+		snapshot, err = gceCS.createImage(ctx, project, volKey, req.Name, snapshotParams)
+		if err != nil {
+			return nil, err
+		}
+	default:
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid snapshot type: %s", snapshotParams.SnapshotType))
+	}
+
+	klog.V(4).Infof("CreateSnapshot succeeded for snapshot %s on volume %s", snapshot.SnapshotId, volumeID)
+	return &csi.CreateSnapshotResponse{Snapshot: snapshot}, nil
+}
+
+func (gceCS *GCEControllerServer) createPDSnapshot(ctx context.Context, project string, volKey *meta.Key, snapshotName string, snapshotParams common.SnapshotParameters) (*csi.Snapshot, error) {
+	volumeID, err := common.KeyToVolumeID(volKey, project)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid volume key: %v", volKey))
+	}
+
+	// Check if PD snapshot already exists
 	var snapshot *compute.Snapshot
-	snapshot, err = gceCS.CloudProvider.GetSnapshot(ctx, project, req.Name)
+	snapshot, err = gceCS.CloudProvider.GetSnapshot(ctx, project, snapshotName)
 	if err != nil {
 		if !gce.IsGCEError(err, "notFound") {
 			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
-		snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(req.GetParameters())
-		if err != nil {
-			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid snapshot parameters: %v", err))
-		}
-		snapshot, err = gceCS.CloudProvider.CreateSnapshot(ctx, project, volKey, req.Name, snapshotParams)
+		snapshot, err = gceCS.CloudProvider.CreateSnapshot(ctx, project, volKey, snapshotName, snapshotParams)
 		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))
@@ -832,32 +863,122 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 	if err != nil {
 		return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Error in creating snapshot: %v", err))
 	}
-	t, err := time.Parse(time.RFC3339, snapshot.CreationTimestamp)
+
+	timestamp, err := parseTimestamp(snapshot.CreationTimestamp)
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to covert creation timestamp: %v", err))
 	}
 
-	tp, err := ptypes.TimestampProto(t)
+	ready, err := isCSISnapshotReady(snapshot.Status)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("Snapshot had error checking ready status: %v", err))
+	}
+
+	return &csi.Snapshot{
+		SizeBytes:      common.GbToBytes(snapshot.DiskSizeGb),
+		SnapshotId:     cleanSelfLink(snapshot.SelfLink),
+		SourceVolumeId: volumeID,
+		CreationTime:   timestamp,
+		ReadyToUse:     ready,
+	}, nil
+}
+
+func (gceCS *GCEControllerServer) createImage(ctx context.Context, project string, volKey *meta.Key, imageName string, snapshotParams common.SnapshotParameters) (*csi.Snapshot, error) {
+	volumeID, err := common.KeyToVolumeID(volKey, project)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("Invalid volume key: %v", volKey))
+	}
+
+	// Check if image already exists
+	var image *compute.Image
+	image, err = gceCS.CloudProvider.GetImage(ctx, project, imageName)
+	if err != nil {
+		if !gce.IsGCEError(err, "notFound") {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown get image error: %v", err))
+		}
+		// create a new image
+		image, err = gceCS.CloudProvider.CreateImage(ctx, project, volKey, imageName, snapshotParams)
+		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))
+			}
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown create image error: %v", err))
+		}
+	}
+
+	err = gceCS.validateExistingImage(image, volKey)
+	if err != nil {
+		return nil, status.Errorf(codes.AlreadyExists, fmt.Sprintf("Error in creating snapshot: %v", err))
+	}
+
+	timestamp, err := parseTimestamp(image.CreationTimestamp)
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Failed to covert creation timestamp: %v", err))
 	}
 
-	ready, err := isCSISnapshotReady(snapshot.Status)
+	ready, err := isImageReady(image.Status)
 	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("Snapshot had error checking ready status: %v", err))
+		return nil, status.Errorf(codes.Internal, fmt.Sprintf("Failed to check image status: %v", err))
 	}
 
-	createResp := &csi.CreateSnapshotResponse{
-		Snapshot: &csi.Snapshot{
-			SizeBytes:      common.GbToBytes(snapshot.DiskSizeGb),
-			SnapshotId:     cleanSelfLink(snapshot.SelfLink),
-			SourceVolumeId: volumeID,
-			CreationTime:   tp,
-			ReadyToUse:     ready,
-		},
+	return &csi.Snapshot{
+		SizeBytes:      common.GbToBytes(image.DiskSizeGb),
+		SnapshotId:     cleanSelfLink(image.SelfLink),
+		SourceVolumeId: volumeID,
+		CreationTime:   timestamp,
+		ReadyToUse:     ready,
+	}, nil
+}
+
+func (gceCS *GCEControllerServer) validateExistingImage(image *compute.Image, volKey *meta.Key) error {
+	if image == nil {
+		return fmt.Errorf("disk does not exist")
+	}
+
+	_, sourceKey, err := common.VolumeIDToKey(cleanSelfLink(image.SourceDisk))
+	if err != nil {
+		return fmt.Errorf("fail to get source disk key %s, %v", image.SourceDisk, err)
+	}
+
+	if sourceKey.String() != volKey.String() {
+		return fmt.Errorf("image already exists with same name but with a different disk source %s, expected disk source %s", sourceKey.String(), volKey.String())
+	}
+
+	klog.V(5).Infof("Compatible image %s exists with source disk %s.", image.Name, image.SourceDisk)
+	return nil
+}
+
+func parseTimestamp(creationTimestamp string) (*timestamp.Timestamp, error) {
+	t, err := time.Parse(time.RFC3339, creationTimestamp)
+	if err != nil {
+		return nil, err
+	}
+
+	timestamp, err := ptypes.TimestampProto(t)
+	if err != nil {
+		return nil, err
+	}
+	return timestamp, nil
+}
+func isImageReady(status string) (bool, error) {
+	// Possible status:
+	//   "DELETING"
+	//   "FAILED"
+	//   "PENDING"
+	//   "READY"
+	switch status {
+	case "DELETING":
+		klog.V(4).Infof("image status is DELETING")
+		return true, nil
+	case "FAILED":
+		return false, fmt.Errorf("image status is FAILED")
+	case "PENDING":
+		return false, nil
+	case "READY":
+		return true, nil
+	default:
+		return false, fmt.Errorf("unknown image status %s", status)
 	}
-	klog.V(4).Infof("CreateSnapshot succeeded for snapshot %s on volume %s", cleanSelfLink(snapshot.SelfLink), volumeID)
-	return createResp, nil
 }
 
 func (gceCS *GCEControllerServer) validateExistingSnapshot(snapshot *compute.Snapshot, volKey *meta.Key) error {
@@ -899,7 +1020,7 @@ func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.D
 		return nil, status.Error(codes.InvalidArgument, "DeleteSnapshot Snapshot ID must be provided")
 	}
 
-	project, key, err := common.SnapshotIDToProjectKey(snapshotID)
+	project, snapshotType, key, err := common.SnapshotIDToProjectKey(snapshotID)
 	if err != nil {
 		// Cannot get snapshot ID from the passing request
 		// This is a success according to the spec
@@ -907,9 +1028,19 @@ func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.D
 		return &csi.DeleteSnapshotResponse{}, nil
 	}
 
-	err = gceCS.CloudProvider.DeleteSnapshot(ctx, project, key)
-	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown Delete snapshot error: %v", err))
+	switch snapshotType {
+	case common.DiskSnapshotType:
+		err = gceCS.CloudProvider.DeleteSnapshot(ctx, project, key)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("unknown Delete snapshot error: %v", err))
+		}
+	case common.DiskImageType:
+		err = gceCS.CloudProvider.DeleteImage(ctx, project, key)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("unknown Delete image error: %v", err))
+		}
+	default:
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("unknown snapshot type %s", snapshotType))
 	}
 
 	return &csi.DeleteSnapshotResponse{}, nil
@@ -922,7 +1053,48 @@ func (gceCS *GCEControllerServer) ListSnapshots(ctx context.Context, req *csi.Li
 	}
 
 	// case 2: no SnapshotId is set, so we return all the snapshots that satify the reqeust.
-	return gceCS.getSnapshots(ctx, req)
+	var maxEntries int = int(req.MaxEntries)
+	if maxEntries < 0 {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf(
+			"ListSnapshots got max entries request %v. GCE only supports values >0", maxEntries))
+	}
+
+	var offset int
+	var length int
+	var ok bool
+	var nextToken string
+	if req.StartingToken == "" {
+		snapshotList, err := gceCS.getSnapshots(ctx, req)
+		if err != nil {
+			if gce.IsGCEInvalidError(err) {
+				return nil, status.Error(codes.Aborted, fmt.Sprintf("ListSnapshots error with invalid request: %v", err))
+			}
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list snapshots error: %v", err))
+		}
+		gceCS.snapshots = snapshotList
+		gceCS.snapshotTokens = map[string]int{}
+	} else {
+		offset, ok = gceCS.snapshotTokens[req.StartingToken]
+		if !ok {
+			return nil, status.Error(codes.Aborted, fmt.Sprintf("ListSnapshots error with invalid startingToken: %s", req.StartingToken))
+		}
+	}
+
+	if maxEntries == 0 {
+		maxEntries = len(gceCS.snapshots)
+	}
+	if maxEntries < len(gceCS.snapshots)-offset {
+		length = maxEntries
+		nextToken = string(uuid.NewUUID())
+		gceCS.snapshotTokens[nextToken] = length + offset
+	} else {
+		length = len(gceCS.snapshots) - offset
+	}
+
+	return &csi.ListSnapshotsResponse{
+		Entries:   gceCS.snapshots[offset : offset+length],
+		NextToken: nextToken,
+	}, nil
 }
 
 func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
@@ -961,59 +1133,90 @@ func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, re
 	}, nil
 }
 
-func (gceCS *GCEControllerServer) getSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
+func (gceCS *GCEControllerServer) getSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) ([]*csi.ListSnapshotsResponse_Entry, error) {
 	var snapshots []*compute.Snapshot
-	var nextToken string
+	var images []*compute.Image
+	var filter string
 	var err error
 	if len(req.GetSourceVolumeId()) != 0 {
-		snapshots, nextToken, err = gceCS.CloudProvider.ListSnapshots(ctx, fmt.Sprintf("sourceDisk eq .*%s$", req.SourceVolumeId), int64(req.MaxEntries), req.StartingToken)
-	} else {
-		snapshots, nextToken, err = gceCS.CloudProvider.ListSnapshots(ctx, "", int64(req.MaxEntries), req.StartingToken)
+		filter = fmt.Sprintf("sourceDisk eq .*%s$", req.SourceVolumeId)
 	}
+	snapshots, _, err = gceCS.CloudProvider.ListSnapshots(ctx, filter)
 	if err != nil {
 		if gce.IsGCEError(err, "invalid") {
 			return nil, status.Error(codes.Aborted, fmt.Sprintf("Invalid error: %v", err))
 		}
 		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list snapshot error: %v", err))
 	}
+
+	images, _, err = gceCS.CloudProvider.ListImages(ctx, filter)
+	if err != nil {
+		if gce.IsGCEError(err, "invalid") {
+			return nil, status.Error(codes.Aborted, fmt.Sprintf("Invalid error: %v", err))
+		}
+		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list image error: %v", err))
+	}
+
 	entries := []*csi.ListSnapshotsResponse_Entry{}
 
 	for _, snapshot := range snapshots {
-		entry, err := generateSnapshotEntry(snapshot)
+		entry, err := generateDiskSnapshotEntry(snapshot)
 		if err != nil {
 			return nil, fmt.Errorf("failed to generate snapshot entry: %v", err)
 		}
 		entries = append(entries, entry)
 	}
-	listSnapshotResp := &csi.ListSnapshotsResponse{
-		Entries:   entries,
-		NextToken: nextToken,
+
+	for _, image := range images {
+		entry, err := generateDiskImageEntry(image)
+		if err != nil {
+			return nil, fmt.Errorf("failed to generate image entry: %v", err)
+		}
+		entries = append(entries, entry)
 	}
-	return listSnapshotResp, nil
+
+	return entries, nil
 }
 
 func (gceCS *GCEControllerServer) getSnapshotByID(ctx context.Context, snapshotID string) (*csi.ListSnapshotsResponse, error) {
-	project, key, err := common.SnapshotIDToProjectKey(snapshotID)
+	project, snapshotType, key, err := common.SnapshotIDToProjectKey(snapshotID)
 	if err != nil {
 		// Cannot get snapshot ID from the passing request
 		klog.Warningf("invalid snapshot id format %s", snapshotID)
 		return &csi.ListSnapshotsResponse{}, nil
 	}
 
-	snapshot, err := gceCS.CloudProvider.GetSnapshot(ctx, project, key)
-	if err != nil {
-		if gce.IsGCEError(err, "notFound") {
-			// return empty list if no snapshot is found
-			return &csi.ListSnapshotsResponse{}, nil
+	var entries []*csi.ListSnapshotsResponse_Entry
+	switch snapshotType {
+	case common.DiskSnapshotType:
+		snapshot, err := gceCS.CloudProvider.GetSnapshot(ctx, project, key)
+		if err != nil {
+			if gce.IsGCEError(err, "notFound") {
+				// return empty list if no snapshot is found
+				return &csi.ListSnapshotsResponse{}, nil
+			}
+			return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list snapshot error: %v", err))
 		}
-		return nil, status.Error(codes.Internal, fmt.Sprintf("Unknown list snapshot error: %v", err))
-	}
-	e, err := generateSnapshotEntry(snapshot)
-	if err != nil {
-		return nil, fmt.Errorf("failed to generate snapshot entry: %v", err)
+		e, err := generateDiskSnapshotEntry(snapshot)
+		if err != nil {
+			return nil, fmt.Errorf("failed to generate snapshot entry: %v", err)
+		}
+		entries = []*csi.ListSnapshotsResponse_Entry{e}
+	case common.DiskImageType:
+		image, err := gceCS.CloudProvider.GetImage(ctx, project, key)
+		if err != nil {
+			if gce.IsGCEError(err, "notFound") {
+				// return empty list if no snapshot is found
+				return &csi.ListSnapshotsResponse{}, nil
+			}
+		}
+		e, err := generateImageEntry(image)
+		if err != nil {
+			return nil, fmt.Errorf("failed to generate image entry: %v", err)
+		}
+		entries = []*csi.ListSnapshotsResponse_Entry{e}
 	}
 
-	entries := []*csi.ListSnapshotsResponse_Entry{e}
 	//entries[0] = entry
 	listSnapshotResp := &csi.ListSnapshotsResponse{
 		Entries: entries,
@@ -1021,7 +1224,7 @@ func (gceCS *GCEControllerServer) getSnapshotByID(ctx context.Context, snapshotI
 	return listSnapshotResp, nil
 }
 
-func generateSnapshotEntry(snapshot *compute.Snapshot) (*csi.ListSnapshotsResponse_Entry, error) {
+func generateDiskSnapshotEntry(snapshot *compute.Snapshot) (*csi.ListSnapshotsResponse_Entry, error) {
 	t, _ := time.Parse(time.RFC3339, snapshot.CreationTimestamp)
 
 	tp, err := ptypes.TimestampProto(t)
@@ -1046,6 +1249,49 @@ func generateSnapshotEntry(snapshot *compute.Snapshot) (*csi.ListSnapshotsRespon
 	return entry, nil
 }
 
+func generateDiskImageEntry(image *compute.Image) (*csi.ListSnapshotsResponse_Entry, error) {
+	t, _ := time.Parse(time.RFC3339, image.CreationTimestamp)
+
+	tp, err := ptypes.TimestampProto(t)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to covert creation timestamp: %v", err)
+	}
+
+	ready, _ := isImageReady(image.Status)
+
+	entry := &csi.ListSnapshotsResponse_Entry{
+		Snapshot: &csi.Snapshot{
+			SizeBytes:      common.GbToBytes(image.DiskSizeGb),
+			SnapshotId:     cleanSelfLink(image.SelfLink),
+			SourceVolumeId: cleanSelfLink(image.SourceDisk),
+			CreationTime:   tp,
+			ReadyToUse:     ready,
+		},
+	}
+	return entry, nil
+}
+
+func generateImageEntry(image *compute.Image) (*csi.ListSnapshotsResponse_Entry, error) {
+	timestamp, err := parseTimestamp(image.CreationTimestamp)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to covert creation timestamp: %v", err)
+	}
+
+	// ignore the error intentionally here since we are just listing images
+	ready, _ := isImageReady(image.Status)
+
+	entry := &csi.ListSnapshotsResponse_Entry{
+		Snapshot: &csi.Snapshot{
+			SizeBytes:      common.GbToBytes(image.DiskSizeGb),
+			SnapshotId:     cleanSelfLink(image.SelfLink),
+			SourceVolumeId: cleanSelfLink(image.SourceDisk),
+			CreationTime:   timestamp,
+			ReadyToUse:     ready,
+		},
+	}
+	return entry, nil
+}
+
 func getRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
 	var capBytes int64
 	// Default case where nothing is set
diff --git a/release-tools/KUBERNETES_CSI_OWNERS_ALIASES b/release-tools/KUBERNETES_CSI_OWNERS_ALIASES
index 2f3e2dfc..8ea922ed 100644
--- a/release-tools/KUBERNETES_CSI_OWNERS_ALIASES
+++ b/release-tools/KUBERNETES_CSI_OWNERS_ALIASES
@@ -31,7 +31,15 @@ aliases:
 
 # This documents who previously contributed to Kubernetes-CSI
 # as approver.
-emeritus_approvers:
+emeritus_approver:
 - lpabon
 - sbezverk
 - vladimirvivien
+
+# This documents who previously contributed to Kubernetes-CSI
+# as reviewer.
+emeritus_reviewer:
+- lpabon
+- saad-ali
+- sbezverk
+- vladimirvivien
diff --git a/release-tools/SIDECAR_RELEASE_PROCESS.md b/release-tools/SIDECAR_RELEASE_PROCESS.md
index 8977fbe6..e4b30e89 100644
--- a/release-tools/SIDECAR_RELEASE_PROCESS.md
+++ b/release-tools/SIDECAR_RELEASE_PROCESS.md
@@ -46,13 +46,10 @@ naming convention `<hostpath-deployment-version>-on-<kubernetes-version>`.
 ## Release Process
 1. Identify all issues and ongoing PRs that should go into the release, and
   drive them to resolution.
-1. Download the latest version of the
-   [K8s release notes generator](https://github.com/kubernetes/release/tree/HEAD/cmd/release-notes)
-1. Create a
-   [Github personal access token](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/creating-a-personal-access-token)
-   with `repo:public_repo` access
+1. Download v2.8+ [K8s release notes
+  generator](https://github.com/kubernetes/release/tree/HEAD/cmd/release-notes)
 1. Generate release notes for the release. Replace arguments with the relevant
-   information.
+  information.
     * Clean up old cached information (also needed if you are generating release
       notes for multiple repos)
       ```bash
@@ -60,24 +57,15 @@ naming convention `<hostpath-deployment-version>-on-<kubernetes-version>`.
       ```
     * For new minor releases on master:
         ```bash
-        GITHUB_TOKEN=<token> release-notes \
-          --discover=mergebase-to-latest \
-          --org=kubernetes-csi \
-          --repo=external-provisioner \
-          --required-author="" \
-          --markdown-links \
-          --output out.md
+        GITHUB_TOKEN=<token> release-notes --discover=mergebase-to-latest
+        --github-org=kubernetes-csi --github-repo=external-provisioner
+        --required-author="" --output out.md
         ```
     * For new patch releases on a release branch:
         ```bash
-        GITHUB_TOKEN=<token> release-notes \
-          --discover=patch-to-latest \
-          --branch=release-1.1 \
-          --org=kubernetes-csi \
-          --repo=external-provisioner \
-          --required-author="" \
-          --markdown-links \
-          --output out.md
+        GITHUB_TOKEN=<token> release-notes --discover=patch-to-latest --branch=release-1.1
+        --github-org=kubernetes-csi --github-repo=external-provisioner
+        --required-author="" --output out.md
         ```
 1. Compare the generated output to the new commits for the release to check if
    any notable change missed a release note.
@@ -112,29 +100,6 @@ naming convention `<hostpath-deployment-version>-on-<kubernetes-version>`.
    and [k/k
    in-tree](https://github.com/kubernetes/kubernetes/tree/HEAD/test/e2e/testing-manifests/storage-csi/hostpath/hostpath)
 
-### Troubleshooting
-
-#### Image build jobs
-
-The following jobs are triggered after tagging to produce the corresponding
-image(s):
-https://k8s-testgrid.appspot.com/sig-storage-image-build
-
-Clicking on a failed build job opens that job in https://prow.k8s.io. Next to
-the job title is a rerun icon (circle with arrow). Clicking it opens a popup
-with a "rerun" button that maintainers with enough permissions can use. If in
-doubt, ask someone on #sig-release to rerun the job.
-
-Another way to rerun a job is to search for it in https://prow.k8s.io and click
-the rerun icon in the resulting job list:
-https://prow.k8s.io/?job=canary-csi-test-push-images
-
-#### Verify images
-
-Canary and staged images can be viewed at https://console.cloud.google.com/gcr/images/k8s-staging-sig-storage
-
-Promoted images can be viewed at https://console.cloud.google.com/gcr/images/k8s-artifacts-prod/us/sig-storage
-
 ## Adding support for a new Kubernetes release
 
 1. Add the new release to `k8s_versions` in
diff --git a/release-tools/build.make b/release-tools/build.make
index ff428890..6caf64fd 100644
--- a/release-tools/build.make
+++ b/release-tools/build.make
@@ -70,7 +70,7 @@ endif
 # Specific packages can be excluded from each of the tests below by setting the *_FILTER_CMD variables
 # to something like "| grep -v 'github.com/kubernetes-csi/project/pkg/foobar'". See usage below.
 
-# BUILD_PLATFORMS contains a set of tuples [os arch buildx_platform suffix base_image addon_image]
+# BUILD_PLATFORMS contains a set of tuples [os arch suffix base_image addon_image]
 # separated by semicolon. An empty variable or empty entry (= just a
 # semicolon) builds for the default platform of the current Go
 # toolchain.
@@ -85,12 +85,12 @@ FULL_LDFLAGS = $(LDFLAGS) $(IMPORTPATH_LDFLAGS) $(EXT_LDFLAGS)
 # defined by BUILD_PLATFORMS.
 $(CMDS:%=build-%): build-%: check-go-version-go
 	mkdir -p bin
-	# os_arch_seen captures all of the $$os-$$arch-$$buildx_platform seen for the current binary
-	# that we want to build, if we've seen an $$os-$$arch-$$buildx_platform before it means that
+	# os_arch_seen captures all of the $$os-$$arch seen for the current binary
+	# that we want to build, if we've seen an $$os-$$arch before it means that
 	# we don't need to build it again, this is done to avoid building
 	# the windows binary multiple times (see the default value of $$BUILD_PLATFORMS)
-	export os_arch_seen="" && echo '$(BUILD_PLATFORMS)' | tr ';' '\n' | while read -r os arch buildx_platform suffix base_image addon_image; do \
-		os_arch_seen_pre=$${os_arch_seen%%$$os-$$arch-$$buildx_platform*}; \
+	export os_arch_seen="" && echo '$(BUILD_PLATFORMS)' | tr ';' '\n' | while read -r os arch suffix base_image addon_image; do \
+		os_arch_seen_pre=$${os_arch_seen%%$$os-$$arch*}; \
 		if ! [ $${#os_arch_seen_pre} = $${#os_arch_seen} ]; then \
 			continue; \
 		fi; \
@@ -98,7 +98,7 @@ $(CMDS:%=build-%): build-%: check-go-version-go
 			echo "Building $* for GOOS=$$os GOARCH=$$arch failed, see error(s) above."; \
 			exit 1; \
 		fi; \
-		os_arch_seen+=";$$os-$$arch-$$buildx_platform"; \
+		os_arch_seen+=";$$os-$$arch"; \
 	done
 
 $(CMDS:%=container-%): container-%: build-%
@@ -154,17 +154,16 @@ $(CMDS:%=push-multiarch-%): push-multiarch-%: check-pull-base-ref build-%
 	dockerfile_windows=$$(if [ -e ./$(CMDS_DIR)/$*/Dockerfile.Windows ]; then echo ./$(CMDS_DIR)/$*/Dockerfile.Windows; else echo Dockerfile.Windows; fi); \
 	if [ '$(BUILD_PLATFORMS)' ]; then build_platforms='$(BUILD_PLATFORMS)'; else build_platforms="linux amd64"; fi; \
 	if ! [ -f "$$dockerfile_windows" ]; then \
-		build_platforms="$$(echo "$$build_platforms" | sed -e 's/windows *[^ ]* *[^ ]* *.exe *[^ ]* *[^ ]*//g' -e 's/; *;/;/g' -e 's/;[ ]*$$//')"; \
+		build_platforms="$$(echo "$$build_platforms" | sed -e 's/windows *[^ ]* *.exe *[^ ]* *[^ ]*//g' -e 's/; *;/;/g' -e 's/;[ ]*$$//')"; \
 	fi; \
 	pushMultiArch () { \
 		tag=$$1; \
-		echo "$$build_platforms" | tr ';' '\n' | while read -r os arch buildx_platform suffix base_image addon_image; do \
+		echo "$$build_platforms" | tr ';' '\n' | while read -r os arch suffix base_image addon_image; do \
 			escaped_base_image=$${base_image/:/-}; \
-			escaped_buildx_platform=$${buildx_platform//\//-}; \
 			if ! [ -z $$escaped_base_image ]; then escaped_base_image+="-"; fi; \
 			docker buildx build --push \
-				--tag $(IMAGE_NAME):$$escaped_buildx_platform-$$os-$$escaped_base_image$$tag \
-				--platform=$$os/$$buildx_platform \
+				--tag $(IMAGE_NAME):$$arch-$$os-$$escaped_base_image$$tag \
+				--platform=$$os/$$arch \
 				--file $$(eval echo \$${dockerfile_$$os}) \
 				--build-arg binary=./bin/$*$$suffix \
 				--build-arg ARCH=$$arch \
@@ -173,14 +172,13 @@ $(CMDS:%=push-multiarch-%): push-multiarch-%: check-pull-base-ref build-%
 				--label revision=$(REV) \
 				.; \
 		done; \
-		images=$$(echo "$$build_platforms" | tr ';' '\n' | while read -r os arch buildx_platform suffix base_image addon_image; do \
+		images=$$(echo "$$build_platforms" | tr ';' '\n' | while read -r os arch suffix base_image addon_image; do \
 			escaped_base_image=$${base_image/:/-}; \
-			escaped_buildx_platform=$${buildx_platform//\//-}; \
 			if ! [ -z $$escaped_base_image ]; then escaped_base_image+="-"; fi; \
-			echo $(IMAGE_NAME):$$escaped_buildx_platform-$$os-$$escaped_base_image$$tag; \
+			echo $(IMAGE_NAME):$$arch-$$os-$$escaped_base_image$$tag; \
 		done); \
 		docker manifest create --amend $(IMAGE_NAME):$$tag $$images; \
-		echo "$$build_platforms" | tr ';' '\n' | while read -r os arch buildx_platform suffix base_image addon_image; do \
+		echo "$$build_platforms" | tr ';' '\n' | while read -r os arch suffix base_image addon_image; do \
 			if [ $$os = "windows" ]; then \
 				escaped_base_image=$${base_image/:/-}; \
 				if ! [ -z $$escaped_base_image ]; then escaped_base_image+="-"; fi; \
diff --git a/release-tools/cloudbuild.yaml b/release-tools/cloudbuild.yaml
index 4bcffad3..f36f265f 100644
--- a/release-tools/cloudbuild.yaml
+++ b/release-tools/cloudbuild.yaml
@@ -16,8 +16,9 @@
 # To promote release images, see https://github.com/kubernetes/k8s.io/tree/HEAD/k8s.gcr.io/images/k8s-staging-sig-storage.
 
 # This must be specified in seconds. If omitted, defaults to 600s (10 mins).
-# Building three images in external-snapshotter takes more than an hour.
-timeout: 7200s
+# Building three images in external-snapshotter takes roughly half an hour,
+# sometimes more.
+timeout: 3600s
 # This prevents errors if you don't use both _GIT_TAG and _PULL_BASE_REF,
 # or any new substitutions added in the future.
 options:
diff --git a/release-tools/prow.sh b/release-tools/prow.sh
index 8053b205..ac8d8b0c 100755
--- a/release-tools/prow.sh
+++ b/release-tools/prow.sh
@@ -78,7 +78,7 @@ version_to_git () {
 # the list of windows versions was matched from:
 # - https://hub.docker.com/_/microsoft-windows-nanoserver
 # - https://hub.docker.com/_/microsoft-windows-servercore
-configvar CSI_PROW_BUILD_PLATFORMS "linux amd64 amd64; linux ppc64le ppc64le -ppc64le; linux s390x s390x -s390x; linux arm arm -arm; linux arm64 arm64 -arm64; linux arm arm/v7 -armv7; windows amd64 amd64 .exe nanoserver:1809 servercore:ltsc2019; windows amd64 amd64 .exe nanoserver:20H2 servercore:20H2; windows amd64 amd64 .exe nanoserver:ltsc2022 servercore:ltsc2022" "Go target platforms (= GOOS + GOARCH) and file suffix of the resulting binaries"
+configvar CSI_PROW_BUILD_PLATFORMS "linux amd64; linux ppc64le -ppc64le; linux s390x -s390x; linux arm -arm; linux arm64 -arm64; windows amd64 .exe nanoserver:1809 servercore:ltsc2019; windows amd64 .exe nanoserver:1909 servercore:1909; windows amd64 .exe nanoserver:2004 servercore:2004; windows amd64 .exe nanoserver:20H2 servercore:20H2; windows amd64 .exe nanoserver:ltsc2022 servercore:ltsc2022" "Go target platforms (= GOOS + GOARCH) and file suffix of the resulting binaries"
 
 # If we have a vendor directory, then use it. We must be careful to only
 # use this for "make" invocations inside the project's repo itself because
@@ -149,8 +149,7 @@ configvar CSI_PROW_KIND_VERSION "$(kind_version_default)" "kind"
 
 # kind images to use. Must match the kind version.
 # The release notes of each kind release list the supported images.
-configvar CSI_PROW_KIND_IMAGES "kindest/node:v1.23.0@sha256:49824ab1727c04e56a21a5d8372a402fcd32ea51ac96a2706a12af38934f81ac
-kindest/node:v1.22.0@sha256:b8bda84bb3a190e6e028b1760d277454a72267a5454b57db34437c34a588d047
+configvar CSI_PROW_KIND_IMAGES "kindest/node:v1.22.0@sha256:b8bda84bb3a190e6e028b1760d277454a72267a5454b57db34437c34a588d047
 kindest/node:v1.21.1@sha256:69860bda5563ac81e3c0057d654b5253219618a22ec3a346306239bba8cfa1a6
 kindest/node:v1.20.7@sha256:cbeaf907fc78ac97ce7b625e4bf0de16e3ea725daf6b04f930bd14c67c671ff9
 kindest/node:v1.19.11@sha256:07db187ae84b4b7de440a73886f008cf903fcf5764ba8106a9fd5243d6f32729
@@ -234,7 +233,7 @@ configvar CSI_PROW_E2E_IMPORT_PATH "k8s.io/kubernetes" "E2E package"
 # of the cluster. The alternative would have been to (cross-)compile csi-sanity
 # and install it inside the cluster, which is not necessarily easier.
 configvar CSI_PROW_SANITY_REPO https://github.com/kubernetes-csi/csi-test "csi-test repo"
-configvar CSI_PROW_SANITY_VERSION v4.3.0 "csi-test version"
+configvar CSI_PROW_SANITY_VERSION v4.2.0 "csi-test version"
 configvar CSI_PROW_SANITY_PACKAGE_PATH github.com/kubernetes-csi/csi-test "csi-test package"
 configvar CSI_PROW_SANITY_SERVICE "hostpath-service" "Kubernetes TCP service name that exposes csi.sock"
 configvar CSI_PROW_SANITY_POD "csi-hostpathplugin-0" "Kubernetes pod with CSI driver"
@@ -737,7 +736,7 @@ install_csi_driver () {
     fi
 }
 
-# Installs all necessary snapshotter CRDs
+# Installs all nessesary snapshotter CRDs
 install_snapshot_crds() {
   # Wait until volumesnapshot CRDs are in place.
   CRD_BASE_DIR="https://raw.githubusercontent.com/kubernetes-csi/external-snapshotter/${CSI_SNAPSHOTTER_VERSION}/client/config/crd"
diff --git a/vendor/modules.txt b/vendor/modules.txt
index e0253e18..c35f4730 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -3,6 +3,8 @@
 cloud.google.com/go/compute/metadata
 cloud.google.com/go/iam
 cloud.google.com/go/kms/apiv1
+# github.com/GoogleCloudPlatform/guest-configs v0.0.0-20211028171426-a0dacef88470
+## explicit
 # github.com/GoogleCloudPlatform/k8s-cloud-provider v0.0.0-20200415212048-7901bc822317
 ## explicit; go 1.13
 github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta


Test patch:
diff --git a/pkg/common/parameters_test.go b/pkg/common/parameters_test.go
index d0fb8f8e..5f7eff78 100644
--- a/pkg/common/parameters_test.go
+++ b/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)
 			}
 		})
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index 14f20361..5ac0215f 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -64,6 +64,7 @@ var (
 	region, _      = common.GetRegionFromZones([]string{zone})
 	testRegionalID = fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, region, name)
 	testSnapshotID = fmt.Sprintf("projects/%s/global/snapshots/%s", project, name)
+	testImageID    = fmt.Sprintf("projects/%s/global/images/%s", project, name)
 	testNodeID     = fmt.Sprintf("projects/%s/zones/%s/instances/%s", project, zone, node)
 )
 
@@ -99,6 +100,24 @@ func TestCreateSnapshotArguments(t *testing.T) {
 				ReadyToUse:     false,
 			},
 		},
+		{
+			name: "success disk image of zonal disk",
+			req: &csi.CreateSnapshotRequest{
+				Name:           name,
+				SourceVolumeId: testVolumeID,
+				Parameters:     map[string]string{common.ParameterKeyStorageLocations: " US-WEST2", common.ParameterKeySnapshotType: "images"},
+			},
+			seedDisks: []*gce.CloudDisk{
+				createZonalCloudDisk(name),
+			},
+			expSnapshot: &csi.Snapshot{
+				SnapshotId:     testImageID,
+				SourceVolumeId: testVolumeID,
+				CreationTime:   tp,
+				SizeBytes:      common.GbToBytes(gce.DiskSizeGb),
+				ReadyToUse:     false,
+			},
+		},
 		{
 			name: "fail no name",
 			req: &csi.CreateSnapshotRequest{
@@ -199,11 +218,17 @@ func TestDeleteSnapshot(t *testing.T) {
 		expErrCode codes.Code
 	}{
 		{
-			name: "valid",
+			name: "valid snapshot delete",
 			req: &csi.DeleteSnapshotRequest{
 				SnapshotId: testSnapshotID,
 			},
 		},
+		{
+			name: "valid image delete",
+			req: &csi.DeleteSnapshotRequest{
+				SnapshotId: testImageID,
+			},
+		},
 		{
 			name: "invalid id",
 			req: &csi.DeleteSnapshotRequest{
@@ -246,31 +271,62 @@ func TestDeleteSnapshot(t *testing.T) {
 func TestListSnapshotsArguments(t *testing.T) {
 	// Define test cases
 	testCases := []struct {
-		name         string
-		req          *csi.ListSnapshotsRequest
-		numSnapshots int
-		expErrCode   codes.Code
+		name          string
+		req           *csi.ListSnapshotsRequest
+		numSnapshots  int
+		numImages     int
+		expectedCount int
+		expErrCode    codes.Code
 	}{
 		{
 			name: "valid",
 			req: &csi.ListSnapshotsRequest{
 				SnapshotId: testSnapshotID + "0",
 			},
-			numSnapshots: 1,
+			numSnapshots:  3,
+			numImages:     2,
+			expectedCount: 1,
 		},
 		{
 			name: "invalid id",
 			req: &csi.ListSnapshotsRequest{
 				SnapshotId: testSnapshotID + "/foo",
 			},
-			numSnapshots: 0,
+			expectedCount: 0,
 		},
 		{
 			name: "no id",
 			req: &csi.ListSnapshotsRequest{
 				SnapshotId: "",
 			},
-			numSnapshots: 5,
+			numSnapshots:  2,
+			numImages:     3,
+			expectedCount: 5,
+		},
+		{
+			name: "with invalid token",
+			req: &csi.ListSnapshotsRequest{
+				StartingToken: "invalid",
+			},
+			expectedCount: 0,
+			expErrCode:    codes.Aborted,
+		},
+		{
+			name: "negative entries",
+			req: &csi.ListSnapshotsRequest{
+				MaxEntries: -1,
+			},
+
+			expErrCode: codes.InvalidArgument,
+		},
+		{
+			name: "max enries",
+			req: &csi.ListSnapshotsRequest{
+				MaxEntries: 4,
+			},
+			numSnapshots:  2,
+			numImages:     3,
+			expectedCount: 4,
 		},
 	}
 
@@ -278,7 +334,7 @@ func TestListSnapshotsArguments(t *testing.T) {
 		t.Logf("test case: %s", tc.name)
 
 		disks := []*gce.CloudDisk{}
-		for i := 0; i < tc.numSnapshots; i++ {
+		for i := 0; i < tc.numSnapshots+tc.numImages; i++ {
 			sname := fmt.Sprintf("%s%d", name, i)
 			disks = append(disks, createZonalCloudDisk(sname))
 		}
@@ -292,6 +348,21 @@ func TestListSnapshotsArguments(t *testing.T) {
 			createReq := &csi.CreateSnapshotRequest{
 				Name:           nameID,
 				SourceVolumeId: volumeID,
+				Parameters:     map[string]string{common.ParameterKeySnapshotType: common.DiskSnapshotType},
+			}
+			_, err := gceDriver.cs.CreateSnapshot(context.Background(), createReq)
+			if err != nil {
+				t.Errorf("error %v", err)
+			}
+		}
+
+		for i := 0; i < tc.numImages; i++ {
+			volumeID := fmt.Sprintf("%s%d", testVolumeID, i)
+			nameID := fmt.Sprintf("%s%d", name, i)
+			createReq := &csi.CreateSnapshotRequest{
+				Name:           nameID,
+				SourceVolumeId: volumeID,
+				Parameters:     map[string]string{common.ParameterKeySnapshotType: common.DiskImageType},
 			}
 			_, err := gceDriver.cs.CreateSnapshot(context.Background(), createReq)
 			if err != nil {
@@ -327,8 +398,8 @@ func TestListSnapshotsArguments(t *testing.T) {
 			// If one is nil or empty but not both
 			t.Fatalf("Expected snapshots number %v, got no snapshot", tc.numSnapshots)
 		}
-		if len(snapshots) != tc.numSnapshots {
-			errStr := fmt.Sprintf("Expected snapshot: %#v\n to equal snapshot: %#v\n", snapshots[0].Snapshot, tc.numSnapshots)
+		if len(snapshots) != tc.expectedCount {
+			errStr := fmt.Sprintf("Expected snapshot number to equal: %v", tc.numSnapshots)
 			t.Errorf(errStr)
 		}
 	}
@@ -827,6 +898,7 @@ func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 		name            string
 		project         string
 		volKey          *meta.Key
+		snapshotType    string
 		snapshotOnCloud bool
 		expErrCode      codes.Code
 	}{
@@ -834,12 +906,29 @@ func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 			name:            "success with data source of snapshot type",
 			project:         "test-project",
 			volKey:          meta.ZonalKey("my-disk", zone),
+			snapshotType:    common.DiskSnapshotType,
 			snapshotOnCloud: true,
 		},
 		{
 			name:            "fail with data source of snapshot type that doesn't exist",
 			project:         "test-project",
 			volKey:          meta.ZonalKey("my-disk", zone),
+			snapshotType:    common.DiskSnapshotType,
+			snapshotOnCloud: false,
+			expErrCode:      codes.NotFound,
+		},
+		{
+			name:            "success with data source of snapshot type",
+			project:         "test-project",
+			volKey:          meta.ZonalKey("my-disk", zone),
+			snapshotType:    common.DiskImageType,
+			snapshotOnCloud: true,
+		},
+		{
+			name:            "fail with data source of snapshot type that doesn't exist",
+			project:         "test-project",
+			volKey:          meta.ZonalKey("my-disk", zone),
+			snapshotType:    common.DiskImageType,
 			snapshotOnCloud: false,
 			expErrCode:      codes.NotFound,
 		},
@@ -851,7 +940,28 @@ func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 		// Setup new driver each time so no interference
 		gceDriver := initGCEDriver(t, nil)
 
+		snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(nil)
+		if err != nil {
+			t.Errorf("Got error extracting snapshot parameters: %v", err)
+		}
+
 		// Start Test
+		var snapshotID string
+		switch tc.snapshotType {
+		case common.DiskSnapshotType:
+			snapshotID = testSnapshotID
+			if tc.snapshotOnCloud {
+				gceDriver.cs.CloudProvider.CreateSnapshot(context.Background(), tc.project, tc.volKey, name, snapshotParams)
+			}
+		case common.DiskImageType:
+			snapshotID = testImageID
+			if tc.snapshotOnCloud {
+				gceDriver.cs.CloudProvider.CreateImage(context.Background(), tc.project, tc.volKey, name, snapshotParams)
+			}
+		default:
+			t.Errorf("Unknown snapshot type: %v", tc.snapshotType)
+		}
+
 		req := &csi.CreateVolumeRequest{
 			Name:               "test-name",
 			CapacityRange:      stdCapRange,
@@ -859,19 +969,12 @@ func TestCreateVolumeWithVolumeSourceFromSnapshot(t *testing.T) {
 			VolumeContentSource: &csi.VolumeContentSource{
 				Type: &csi.VolumeContentSource_Snapshot{
 					Snapshot: &csi.VolumeContentSource_SnapshotSource{
-						SnapshotId: testSnapshotID,
+						SnapshotId: snapshotID,
 					},
 				},
 			},
 		}
 
-		if tc.snapshotOnCloud {
-			snapshotParams, err := common.ExtractAndDefaultSnapshotParameters(req.GetParameters())
-			if err != nil {
-				t.Errorf("Got error extracting snapshot parameters: %v", err)
-			}
-			gceDriver.cs.CloudProvider.CreateSnapshot(context.Background(), tc.project, tc.volKey, name, snapshotParams)
-		}
 		resp, err := gceDriver.cs.CreateVolume(context.Background(), req)
 		//check response
 		if err != nil {
@@ -1717,6 +1820,7 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 		response := make(chan error)
 		go func() {
 			_, err := cs.CreateSnapshot(context.Background(), req)
+			t.Log(err)
 			response <- err
 		}()
 		return response