Fix image creation and clone

This patch fixed two issues in image creation and clone:
- add forceCreate flag to avoid image creation failures when source disk
  is attached
- fix the missing content source in CreateVolume response when restoring
  an image

Author: luohao

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

@@ -188,6 +188,17 @@ func (d *CloudDisk) GetSourceDiskId() string {
 	}
 }
 
+func (d *CloudDisk) GetImageId() string {
+	switch {
+	case d.disk != nil:
+		return d.disk.SourceImageId
+	case d.betaDisk != nil:
+		return d.betaDisk.SourceImageId
+	default:
+		return ""
+	}
+}
+
 func (d *CloudDisk) GetKMSKeyName() string {
 	switch {
 	case d.disk != nil:

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

@@ -196,16 +196,30 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
 	}
 
 	computeDisk := &computev1.Disk{
-		Name:             volKey.Name,
-		SizeGb:           common.BytesToGbRoundUp(capBytes),
-		Description:      "Disk created by GCE-PD CSI Driver",
-		Type:             cloud.GetDiskTypeURI(project, volKey, params.DiskType),
-		SourceSnapshotId: snapshotID,
-		SourceDiskId:     volumeContentSourceVolumeID,
-		SourceImageId:    snapshotID,
-		Status:           cloud.mockDiskStatus,
-		Labels:           params.Labels,
+		Name:         volKey.Name,
+		SizeGb:       common.BytesToGbRoundUp(capBytes),
+		Description:  "Disk created by GCE-PD CSI Driver",
+		Type:         cloud.GetDiskTypeURI(project, volKey, params.DiskType),
+		SourceDiskId: volumeContentSourceVolumeID,
+		Status:       cloud.mockDiskStatus,
+		Labels:       params.Labels,
 	}
+
+	if snapshotID != "" {
+		_, snapshotType, _, err := common.SnapshotIDToProjectKey(snapshotID)
+		if err != nil {
+			return err
+		}
+		switch snapshotType {
+		case common.DiskSnapshotType:
+			computeDisk.SourceSnapshotId = snapshotID
+		case common.DiskImageType:
+			computeDisk.SourceImageId = snapshotID
+		default:
+			return fmt.Errorf("invalid snapshot type in snapshot ID: %s", snapshotType)
+		}
+	}
+
 	if params.DiskEncryptionKMSKey != "" {
 		computeDisk.DiskEncryptionKey = &computev1.CustomerEncryptionKey{
 			KmsKeyName: params.DiskEncryptionKMSKey,

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

@@ -871,7 +871,7 @@ func (cloud *CloudProvider) CreateImage(ctx context.Context, project string, vol
 		StorageLocations: snapshotParams.StorageLocations,
 	}
 
-	_, err = cloud.service.Images.Insert(project, image).Context(ctx).Do()
+	_, err = cloud.service.Images.Insert(project, image).Context(ctx).ForceCreate(true).Do()
 	if err != nil {
 		return nil, err
 	}

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

@@ -1480,8 +1480,9 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string) *csi.Crea
 		},
 	}
 	snapshotID := disk.GetSnapshotId()
+	imageID := disk.GetImageId()
 	diskID := disk.GetSourceDiskId()
-	if diskID != "" || snapshotID != "" {
+	if diskID != "" || snapshotID != "" || imageID != "" {
 		contentSource := &csi.VolumeContentSource{}
 		if snapshotID != "" {
 			contentSource = &csi.VolumeContentSource{
@@ -1501,6 +1502,15 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string) *csi.Crea
 				},
 			}
 		}
+		if imageID != "" {
+			contentSource = &csi.VolumeContentSource{
+				Type: &csi.VolumeContentSource_Snapshot{
+					Snapshot: &csi.VolumeContentSource_SnapshotSource{
+						SnapshotId: imageID,
+					},
+				},
+			}
+		}
 		createResp.Volume.ContentSource = contentSource
 	}
 	return createResp