Return success in CreateVolume when disk is READY

Author: saikat-royc

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

@@ -92,6 +92,17 @@ func (d *CloudDisk) GetKind() string {
 	}
 }
 
+func (d *CloudDisk) GetStatus() string {
+	switch d.Type() {
+	case Zonal:
+		return d.ZonalDisk.Status
+	case Regional:
+		return d.RegionalDisk.Status
+	default:
+		return "Unknown"
+	}
+}
+
 // GetPDType returns the type of the PD as either 'pd-standard' or 'pd-ssd' The
 // "Type" field on the compute disk is stored as a url like
 // projects/project/zones/zone/diskTypes/pd-standard

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

@@ -250,6 +250,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key
 			Type:             cloud.GetDiskTypeURI(volKey, params.DiskType),
 			SelfLink:         fmt.Sprintf("projects/%s/zones/%s/disks/%s", cloud.project, volKey.Zone, volKey.Name),
 			SourceSnapshotId: snapshotID,
+			Status:           "READY",
 		}
 		if params.DiskEncryptionKMSKey != "" {
 			diskToCreateGA.DiskEncryptionKey = &computev1.CustomerEncryptionKey{
@@ -265,6 +266,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key
 			Type:             cloud.GetDiskTypeURI(volKey, params.DiskType),
 			SelfLink:         fmt.Sprintf("projects/%s/regions/%s/disks/%s", cloud.project, volKey.Region, volKey.Name),
 			SourceSnapshotId: snapshotID,
+			Status:           "READY",
 		}
 		if params.DiskEncryptionKMSKey != "" {
 			diskToCreateV1.DiskEncryptionKey = &computev1.CustomerEncryptionKey{

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

@@ -62,6 +62,29 @@ const (
 	replicationTypeRegionalPD = "regional-pd"
 )
 
+func isDiskReady(disk *gce.CloudDisk) (bool, error) {
+	status := disk.GetStatus()
+	switch status {
+	case "READY":
+		return true, nil
+	case "FAILED":
+		return false, fmt.Errorf("Disk %s status is FAILED", disk.GetName())
+	case "CREATING":
+		klog.V(4).Infof("Disk %s status is CREATING", disk.GetName())
+		return false, nil
+	case "DELETING":
+		klog.V(4).Infof("Disk %s status is DELETING", disk.GetName())
+		return false, nil
+	case "RESTORING":
+		klog.V(4).Infof("Disk %s status is RESTORING", disk.GetName())
+		return false, nil
+	default:
+		klog.V(4).Infof("Disk %s status is: %s", disk.GetName(), status)
+	}
+
+	return false, nil
+}
+
 func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	var err error
 	// Validate arguments
@@ -143,6 +166,16 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		if err != nil {
 			return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("CreateVolume disk already exists with same name and is incompatible: %v", err))
 		}
+
+		ready, err := isDiskReady(existingDisk)
+		if err != nil {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume disk %v had error checking ready status: %v", volKey, err))
+		}
+
+		if !ready {
+			return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume disk %v is not ready", volKey))
+		}
+
 		// If there is no validation error, immediately return success
 		klog.V(4).Infof("CreateVolume succeeded for disk %v, it already exists and was compatible", volKey)
 		return generateCreateVolumeResponse(existingDisk, capBytes, zones), nil
@@ -187,6 +220,15 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	default:
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("CreateVolume replication type '%s' is not supported", params.ReplicationType))
 	}
+
+	ready, err := isDiskReady(disk)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume disk %v had error checking ready status: %v", volKey, err))
+	}
+	if !ready {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume disk %v is not ready", volKey))
+	}
+
 	klog.V(4).Infof("CreateVolume succeeded for disk %v", volKey)
 	return generateCreateVolumeResponse(disk, capBytes, zones), nil