responding to comments

Author: Hantao (Will) Wang

pkg/common/volume_lock.go

@@ -39,6 +39,8 @@ func NewVolumeLocks() *VolumeLocks {
 	}
 }
 
+// Tries to acquire the lock for operating on volumeID and returns true if successful.
+// If another operation is already using volumeID, returns false.
 func (vl *VolumeLocks) TryAcquire(volumeID string) bool {
 	vl.mux.Lock()
 	defer vl.mux.Unlock()

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

@@ -63,10 +63,6 @@ func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*Fa
 	return fcp, nil
 }
 
-func (cloud *FakeCloudProvider) GetProject() string {
-	return cloud.project
-}
-
 func (cloud *FakeCloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, volumeKey *meta.Key) (*meta.Key, error) {
 	switch volumeKey.Type() {
 	case meta.Zonal:

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

@@ -60,12 +60,6 @@ type GCECompute interface {
 	GetSnapshot(ctx context.Context, snapshotName string) (*compute.Snapshot, error)
 	CreateSnapshot(ctx context.Context, volKey *meta.Key, snapshotName string) (*compute.Snapshot, error)
 	DeleteSnapshot(ctx context.Context, snapshotName string) error
-	// Metadata
-	GetProject() string
-}
-
-func (cloud *CloudProvider) GetProject() string {
-	return cloud.project
 }
 
 // RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified

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

@@ -143,7 +143,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("CreateVolume replication type '%s' is not supported", replicationType))
 	}
 
-	volumeID, err := common.KeyToVolumeID(volKey, gceCS.CloudProvider.GetProject())
+	volumeID, err := common.KeyToVolumeID(volKey, gceCS.MetadataService.GetProject())
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Failed to convert volume key to volume ID: %v", err)
 	}
@@ -276,6 +276,8 @@ func (gceCS *GCEControllerServer) ControllerPublishVolume(ctx context.Context, r
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
+	// Acquires the lock for the volume on that node only, because we need to support the ability
+	// to publish the same volume onto different nodes concurrently
 	lockingVolumeID := fmt.Sprintf("%s/%s", nodeID, volumeID)
 	if acquired := gceCS.volumeLocks.TryAcquire(lockingVolumeID); !acquired {
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID)
@@ -367,6 +369,8 @@ func (gceCS *GCEControllerServer) ControllerUnpublishVolume(ctx context.Context,
 		return nil, err
 	}
 
+	// Acquires the lock for the volume on that node only, because we need to support the ability
+	// to unpublish the same volume from different nodes concurrently
 	lockingVolumeID := fmt.Sprintf("%s/%s", nodeID, volumeID)
 	if acquired := gceCS.volumeLocks.TryAcquire(lockingVolumeID); !acquired {
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, lockingVolumeID)

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

@@ -1432,7 +1432,7 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 		SourceVolumeId: testVolumeId + "2",
 	}
 
-	runRequest := func(req *csi.CreateSnapshotRequest) chan error {
+	runRequest := func(req *csi.CreateSnapshotRequest) <-chan error {
 		response := make(chan error)
 		go func() {
 			_, err := cs.CreateSnapshot(context.Background(), req)
@@ -1454,29 +1454,29 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 		if err != nil {
 			serverError, ok := status.FromError(err)
 			if !ok {
-				t.Fatalf("Could not get error status code from err: %v", err)
+				t.Errorf("Could not get error status code from err: %v", err)
 			}
 			if serverError.Code() != codes.Aborted {
-				t.Fatalf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+				t.Errorf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
 			}
 		} else {
-			t.Fatalf("Expected error: %v, got no error", codes.Aborted)
+			t.Errorf("Expected error: %v, got no error", codes.Aborted)
 		}
 	case <-readyToExecute:
-		t.Fatal("The operation for vol1PublishTargetB should have been aborted, but was started")
+		t.Errorf("The operation for vol1PublishTargetB should have been aborted, but was started")
 	}
 
 	// Start vol2CreateSnapshot and allow it to execute to completion. Then check for success.
 	vol2CreateSnapshotResp := runRequest(vol2CreateSnapshotReq)
 	execVol2CreateSnapshot := <-readyToExecute
 	execVol2CreateSnapshot <- struct{}{}
 	if err := <-vol2CreateSnapshotResp; err != nil {
-		t.Fatalf("Unexpected error: %v", err)
+		t.Errorf("Unexpected error: %v", err)
 	}
 
-	// To clean up, allow the vol1PublishTargetA to complete
+	// To clean up, allow the vol1CreateSnapshotA to complete
 	execVol1CreateSnapshotA <- struct{}{}
 	if err := <-vol1CreateSnapshotAResp; err != nil {
-		t.Fatalf("Unexpected error: %v", err)
+		t.Errorf("Unexpected error: %v", err)
 	}
 }