Revert "Merge pull request kubernetes-sigs#316 from hantaowang/seralize-attach-ops"

This reverts commit 41f103f, reversing
changes made to 965ba6a.

Author: Hantao (Will) Wang

pkg/common/utils.go

@@ -73,17 +73,6 @@ func VolumeIDToKey(id string) (*meta.Key, error) {
 	}
 }
 
-func KeyToVolumeID(volKey *meta.Key, project string) (string, error) {
-	switch volKey.Type() {
-	case meta.Zonal:
-		return fmt.Sprintf(volIDZonalFmt, project, volKey.Zone, volKey.Zone), nil
-	case meta.Regional:
-		return fmt.Sprintf(volIDZonalFmt, project, volKey.Region, volKey.Zone), nil
-	default:
-		return "", fmt.Errorf("volume key %v neither zonal nor regional", volKey.Name)
-	}
-}
-
 func GenerateUnderspecifiedVolumeID(diskName string, isZonal bool) string {
 	if isZonal {
 		return fmt.Sprintf(volIDZonalFmt, UnspecifiedValue, UnspecifiedValue, diskName)

pkg/common/volume_lock.go

@@ -49,7 +49,7 @@ type FakeCloudProvider struct {
 
 var _ GCECompute = &FakeCloudProvider{}
 
-func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*FakeCloudProvider, error) {
+func FakeCreateCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*FakeCloudProvider, error) {
 	fcp := &FakeCloudProvider{
 		project:   project,
 		zone:      zone,
@@ -61,6 +61,7 @@ func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*Fa
 		fcp.disks[d.GetName()] = d
 	}
 	return fcp, nil
+
 }
 
 func (cloud *FakeCloudProvider) RepairUnderspecifiedVolumeKey(ctx context.Context, volumeKey *meta.Key) (*meta.Key, error) {
@@ -433,22 +434,6 @@ func (cloud *FakeCloudProvider) getGlobalSnapshotURI(snapshotName string) string
 		snapshotName)
 }
 
-type FakeBlockingCloudProvider struct {
-	*FakeCloudProvider
-	ReadyToExecute chan chan struct{}
-}
-
-// FakeBlockingCloudProvider's method adds functionality to finely control the order of execution of CreateSnapshot calls.
-// Upon starting a CreateSnapshot, it passes a chan 'executeCreateSnapshot' into readyToExecute, then blocks on executeCreateSnapshot.
-// The test calling this function can block on readyToExecute to ensure that the operation has started and
-// allowed the CreateSnapshot to continue by passing a struct into executeCreateSnapshot.
-func (cloud *FakeBlockingCloudProvider) CreateSnapshot(ctx context.Context, volKey *meta.Key, snapshotName string) (*compute.Snapshot, error) {
-	executeCreateSnapshot := make(chan struct{})
-	cloud.ReadyToExecute <- executeCreateSnapshot
-	<-executeCreateSnapshot
-	return cloud.FakeCloudProvider.CreateSnapshot(ctx, volKey, snapshotName)
-}
-
 func notFoundError() *googleapi.Error {
 	return &googleapi.Error{
 		Errors: []googleapi.ErrorItem{

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

@@ -42,10 +42,6 @@ type GCEControllerServer struct {
 	Driver          *GCEDriver
 	CloudProvider   gce.GCECompute
 	MetadataService metadataservice.MetadataService
-
-	// 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
-	volumeLocks *common.VolumeLocks
 }
 
 var _ csi.ControllerServer = &GCEControllerServer{}
@@ -143,15 +139,6 @@ 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.MetadataService.GetProject())
-	if err != nil {
-		return nil, status.Errorf(codes.Internal, "Failed to convert volume key to volume ID: %v", err)
-	}
-	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
-	}
-	defer gceCS.volumeLocks.Release(volumeID)
-
 	// Validate if disk already exists
 	existingDisk, err := gceCS.CloudProvider.GetDisk(ctx, volKey)
 	if err != nil {
@@ -235,11 +222,6 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
-	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
-	}
-	defer gceCS.volumeLocks.Release(volumeID)
-
 	err = gceCS.CloudProvider.DeleteDisk(ctx, volKey)
 	if err != nil {
 		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown Delete disk error: %v", err))
@@ -276,14 +258,6 @@ 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)
-	}
-	defer gceCS.volumeLocks.Release(lockingVolumeID)
-
 	// TODO(#253): Check volume capability matches for ALREADY_EXISTS
 	if err = validateVolumeCapability(volumeCapability); err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("VolumeCapabilities is invalid: %v", err))
@@ -369,14 +343,6 @@ 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)
-	}
-	defer gceCS.volumeLocks.Release(lockingVolumeID)
-
 	instanceZone, instanceName, err := common.NodeIDToZoneAndName(nodeID)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("could not split nodeID: %v", err))
@@ -423,12 +389,6 @@ func (gceCS *GCEControllerServer) ValidateVolumeCapabilities(ctx context.Context
 	if err != nil {
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Volume ID is of improper format, got %v", volumeID))
 	}
-
-	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
-	}
-	defer gceCS.volumeLocks.Release(volumeID)
-
 	_, err = gceCS.CloudProvider.GetDisk(ctx, volKey)
 	if err != nil {
 		if gce.IsGCEError(err, "notFound") {
@@ -536,11 +496,6 @@ func (gceCS *GCEControllerServer) CreateSnapshot(ctx context.Context, req *csi.C
 		return nil, status.Error(codes.NotFound, fmt.Sprintf("Could not find volume with ID %v: %v", volumeID, err))
 	}
 
-	if acquired := gceCS.volumeLocks.TryAcquire(volumeID); !acquired {
-		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
-	}
-	defer gceCS.volumeLocks.Release(volumeID)
-
 	// Check if snapshot already exists
 	var snapshot *compute.Snapshot
 	snapshot, err = gceCS.CloudProvider.GetSnapshot(ctx, req.Name)

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

@@ -1413,70 +1413,3 @@ func TestPickRandAndConsecutive(t *testing.T) {
 
 	}
 }
-
-func TestVolumeOperationConcurrency(t *testing.T) {
-	readyToExecute := make(chan chan struct{}, 1)
-	gceDriver := initBlockingGCEDriver(t, nil, readyToExecute)
-	cs := gceDriver.cs
-
-	vol1CreateSnapshotAReq := &csi.CreateSnapshotRequest{
-		Name:           name + "1A",
-		SourceVolumeId: testVolumeId + "1",
-	}
-	vol1CreateSnapshotBReq := &csi.CreateSnapshotRequest{
-		Name:           name + "1B",
-		SourceVolumeId: testVolumeId + "1",
-	}
-	vol2CreateSnapshotReq := &csi.CreateSnapshotRequest{
-		Name:           name + "2",
-		SourceVolumeId: testVolumeId + "2",
-	}
-
-	runRequest := func(req *csi.CreateSnapshotRequest) <-chan error {
-		response := make(chan error)
-		go func() {
-			_, err := cs.CreateSnapshot(context.Background(), req)
-			response <- err
-		}()
-		return response
-	}
-
-	// Start first valid request vol1CreateSnapshotA and block until it reaches the CreateSnapshot
-	vol1CreateSnapshotAResp := runRequest(vol1CreateSnapshotAReq)
-	execVol1CreateSnapshotA := <-readyToExecute
-
-	// Start vol1CreateSnapshotB and allow it to execute to completion. Then check for Aborted error.
-	// If a non Abort error is received or if the operation was started, then there is a problem
-	// with volume locking
-	vol1CreateSnapshotBResp := runRequest(vol1CreateSnapshotBReq)
-	select {
-	case err := <-vol1CreateSnapshotBResp:
-		if err != nil {
-			serverError, ok := status.FromError(err)
-			if !ok {
-				t.Fatalf("Could not get error status code from err: %v", err)
-			}
-			if serverError.Code() != codes.Aborted {
-				t.Errorf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
-			}
-		} else {
-			t.Errorf("Expected error: %v, got no error", codes.Aborted)
-		}
-	case <-readyToExecute:
-		t.Errorf("The operation for vol1CreateSnapshotB 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.Errorf("Unexpected error: %v", err)
-	}
-
-	// To clean up, allow the vol1CreateSnapshotA to complete
-	execVol1CreateSnapshotA <- struct{}{}
-	if err := <-vol1CreateSnapshotAResp; err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-}

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

@@ -22,7 +22,6 @@ import (
 	"google.golang.org/grpc/status"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/util/mount"
-	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
@@ -139,7 +138,6 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 		Mounter:         mounter,
 		DeviceUtils:     deviceUtils,
 		MetadataService: meta,
-		volumeLocks:     common.NewVolumeLocks(),
 	}
 }
 
@@ -148,7 +146,6 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, met
 		Driver:          gceDriver,
 		CloudProvider:   cloudProvider,
 		MetadataService: meta,
-		volumeLocks:     common.NewVolumeLocks(),
 	}
 }
 

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

@@ -22,29 +22,13 @@ import (
 )
 
 func initGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk) *GCEDriver {
-	fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
-	if err != nil {
-		t.Fatalf("Failed to create fake cloud provider: %v", err)
-	}
-	return initGCEDriverWithCloudProvider(t, fakeCloudProvider)
-}
-
-func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan struct{}) *GCEDriver {
-	fakeCloudProvider, err := gce.CreateFakeCloudProvider(project, zone, cloudDisks)
+	vendorVersion := "test-vendor"
+	gceDriver := GetGCEDriver()
+	fakeCloudProvider, err := gce.FakeCreateCloudProvider(project, zone, cloudDisks)
 	if err != nil {
 		t.Fatalf("Failed to create fake cloud provider: %v", err)
 	}
-	fakeBlockingBlockProvider := &gce.FakeBlockingCloudProvider{
-		FakeCloudProvider: fakeCloudProvider,
-		ReadyToExecute:    readyToExecute,
-	}
-	return initGCEDriverWithCloudProvider(t, fakeBlockingBlockProvider)
-}
-
-func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute) *GCEDriver {
-	vendorVersion := "test-vendor"
-	gceDriver := GetGCEDriver()
-	err := gceDriver.SetupGCEDriver(cloudProvider, nil, nil, metadataservice.NewFakeService(), driver, vendorVersion)
+	err = gceDriver.SetupGCEDriver(fakeCloudProvider, nil, nil, metadataservice.NewFakeService(), driver, vendorVersion)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}