use chan of chan for testing instead of SourceAndTarget

Author: Hantao (Will) Wang

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

@@ -49,7 +49,7 @@ type FakeCloudProvider struct {
 
 var _ GCECompute = &FakeCloudProvider{}
 
-func FakeCreateCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*FakeCloudProvider, error) {
+func CreateFakeCloudProvider(project, zone string, cloudDisks []*CloudDisk) (*FakeCloudProvider, error) {
 	fcp := &FakeCloudProvider{
 		project:   project,
 		zone:      zone,
@@ -427,40 +427,20 @@ func (cloud *FakeCloudProvider) getGlobalSnapshotURI(snapshotName string) string
 
 type FakeBlockingCloudProvider struct {
 	*FakeCloudProvider
-	snapshotToExec chan SnapshotSourceAndTarget
-	readyToExecute chan struct{}
+	ReadyToExecute chan chan struct{}
 }
 
-type SnapshotSourceAndTarget struct {
-	Source string
-	Target string
-}
-
-// FakeBlockingCloudProvider's method adds two channels to the CreateSnapshot process in order to provide functionality to finely
-// control the order of execution of volume operations. readyToExecute signals that an CreateSnapshot operation has been called.
-// Then it cycles through the snapshotToExec channel, waiting for permission to actually create the snapshot.
+// 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) {
-	cloud.readyToExecute <- struct{}{}
-	for snapshotToExec := range cloud.snapshotToExec {
-		if snapshotToExec.Source == volKey.String() && snapshotToExec.Target == snapshotName {
-			break
-		} else {
-			cloud.snapshotToExec <- snapshotToExec
-		}
-	}
+	executeCreateSnapshot := make(chan struct{})
+	cloud.ReadyToExecute <- executeCreateSnapshot
+	<-executeCreateSnapshot
 	return cloud.FakeCloudProvider.CreateSnapshot(ctx, volKey, snapshotName)
 }
 
-func FakeCreateBlockingCloudProvider(project, zone string, cloudDisks []*CloudDisk, snapshotToExec chan SnapshotSourceAndTarget, readyToExecute chan struct{}) (*FakeBlockingCloudProvider, error) {
-	fcp, err := FakeCreateCloudProvider(project, zone, cloudDisks)
-	fbcp := &FakeBlockingCloudProvider{
-		FakeCloudProvider: fcp,
-		snapshotToExec:    snapshotToExec,
-		readyToExecute:    readyToExecute,
-	}
-	return fbcp, err
-}
-
 func notFoundError() *googleapi.Error {
 	return &googleapi.Error{
 		Errors: []googleapi.ErrorItem{

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

@@ -1415,18 +1415,15 @@ func TestPickRandAndConsecutive(t *testing.T) {
 }
 
 func TestVolumeOperationConcurrency(t *testing.T) {
-	snapshotToExecute := make(chan gce.SnapshotSourceAndTarget, 3)
-	readyToExecute := make(chan struct{}, 2)
-	reqFinished := make(chan error, 2)
-
-	gceDriver := initBlockingGCEDriver(t, nil, snapshotToExecute, readyToExecute)
+	readyToExecute := make(chan chan struct{}, 1)
+	gceDriver := initBlockingGCEDriver(t, nil, readyToExecute)
 	cs := gceDriver.cs
 
-	vol1CreateSnapshotReqA := &csi.CreateSnapshotRequest{
+	vol1CreateSnapshotAReq := &csi.CreateSnapshotRequest{
 		Name:           name + "1A",
 		SourceVolumeId: testVolumeId + "1",
 	}
-	vol1CreateSnapshotReqB := &csi.CreateSnapshotRequest{
+	vol1CreateSnapshotBReq := &csi.CreateSnapshotRequest{
 		Name:           name + "1B",
 		SourceVolumeId: testVolumeId + "1",
 	}
@@ -1435,54 +1432,51 @@ func TestVolumeOperationConcurrency(t *testing.T) {
 		SourceVolumeId: testVolumeId + "2",
 	}
 
-	runRequestInBackground := func(req *csi.CreateSnapshotRequest) {
-		_, err := cs.CreateSnapshot(context.Background(), req)
-		reqFinished <- err
-	}
-
-	sourceAndTargetForRequest := func(req *csi.CreateSnapshotRequest) gce.SnapshotSourceAndTarget {
-		volKey, err := common.VolumeIDToKey(req.SourceVolumeId)
-		if err != nil {
-			t.Fatalf("Could not find volume with ID %v: %v", req.SourceVolumeId, err)
-		}
-		return gce.SnapshotSourceAndTarget{
-			Source: volKey.String(),
-			Target: req.Name,
-		}
+	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 vol1CreateSnapshotReqA and block until it reaches the CreateSnapshot
-	go runRequestInBackground(vol1CreateSnapshotReqA)
-	<-readyToExecute
+	// Start first valid request vol1CreateSnapshotA and block until it reaches the CreateSnapshot
+	vol1CreateSnapshotAResp := runRequest(vol1CreateSnapshotAReq)
+	execVol1CreateSnapshotA := <-readyToExecute
 
-	// Check that vol1CreateSnapshotReqB is rejected, due to same volume ID
-	// Also allow vol1CreateSnapshotReqB to complete, in case it is allowed to Mount
-	snapshotToExecute <- sourceAndTargetForRequest(vol1CreateSnapshotReqB)
-	_, err := cs.CreateSnapshot(context.Background(), vol1CreateSnapshotReqB)
-	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.Fatalf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+	// 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.Fatalf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+			}
+		} else {
+			t.Fatalf("Expected error: %v, got no error", codes.Aborted)
 		}
-	} else {
-		t.Fatalf("Expected error: %v, got no error", codes.Aborted)
+	case <-readyToExecute:
+		t.Fatal("The operation for vol1PublishTargetB should have been aborted, but was started")
 	}
 
-	// Start second valid request vol2CreateSnapshotReq
-	go runRequestInBackground(vol2CreateSnapshotReq)
-
-	// Allow the vol2CreateSnapshotReq to complete, which it can concurrently with vol1CreateSnapshotReqA
-	snapshotToExecute <- sourceAndTargetForRequest(vol2CreateSnapshotReq)
-	if err = <-reqFinished; err != nil {
+	// 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)
 	}
 
-	// To clean up, allow the vol1CreateSnapshotReqA to complete
-	snapshotToExecute <- sourceAndTargetForRequest(vol1CreateSnapshotReqA)
-	if err = <-reqFinished; err != nil {
+	// To clean up, allow the vol1PublishTargetA to complete
+	execVol1CreateSnapshotA <- struct{}{}
+	if err := <-vol1CreateSnapshotAResp; err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 }

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

@@ -22,19 +22,23 @@ import (
 )
 
 func initGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk) *GCEDriver {
-	fakeCloudProvider, err := gce.FakeCreateCloudProvider(project, zone, cloudDisks)
+	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, snapshotToExec chan gce.SnapshotSourceAndTarget, readyToExecute chan struct{}) *GCEDriver {
-	fakeCloudProvider, err := gce.FakeCreateBlockingCloudProvider(project, zone, cloudDisks, snapshotToExec, readyToExecute)
+func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExecute chan chan struct{}) *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)
+	fakeBlockingBlockProvider := &gce.FakeBlockingCloudProvider{
+		FakeCloudProvider: fakeCloudProvider,
+		ReadyToExecute:    readyToExecute,
+	}
+	return initGCEDriverWithCloudProvider(t, fakeBlockingBlockProvider)
 }
 
 func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute) *GCEDriver {

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

@@ -15,9 +15,9 @@ limitations under the License.
 package gceGCEDriver
 
 import (
+	"context"
 	"testing"
 
-	"context"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -38,9 +38,9 @@ func getTestGCEDriver(t *testing.T) *GCEDriver {
 	return gceDriver
 }
 
-func getTestBlockingGCEDriver(t *testing.T, mountToRun chan mountmanager.MountSourceAndTarget, readyToMount chan struct{}) *GCEDriver {
+func getTestBlockingGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
-	err := gceDriver.SetupGCEDriver(nil, mountmanager.NewFakeSafeBlockingMounter(mountToRun, readyToMount), mountmanager.NewFakeDeviceUtils(), metadataservice.NewFakeService(), driver, "test-vendor")
+	err := gceDriver.SetupGCEDriver(nil, mountmanager.NewFakeSafeBlockingMounter(readyToExecute), mountmanager.NewFakeDeviceUtils(), metadataservice.NewFakeService(), driver, "test-vendor")
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}
@@ -389,12 +389,10 @@ func TestNodeGetCapabilities(t *testing.T) {
 }
 
 func TestConcurrentNodeOperations(t *testing.T) {
-	mountToRun := make(chan mountmanager.MountSourceAndTarget, 3)
-	readyToMount := make(chan struct{}, 2)
-	reqFinished := make(chan error, 2)
-
-	gceDriver := getTestBlockingGCEDriver(t, mountToRun, readyToMount)
+	readyToExecute := make(chan chan struct{}, 1)
+	gceDriver := getTestBlockingGCEDriver(t, readyToExecute)
 	ns := gceDriver.ns
+
 	vol1PublishTargetAReq := &csi.NodePublishVolumeRequest{
 		VolumeId:          defaultVolumeID + "1",
 		TargetPath:        defaultTargetPath + "a",
@@ -417,52 +415,51 @@ func TestConcurrentNodeOperations(t *testing.T) {
 		VolumeCapability:  stdVolCap,
 	}
 
-	runRequestInBackground := func(req *csi.NodePublishVolumeRequest) {
-		_, err := ns.NodePublishVolume(context.Background(), req)
-		reqFinished <- err
+	runRequest := func(req *csi.NodePublishVolumeRequest) chan error {
+		response := make(chan error)
+		go func() {
+			_, err := ns.NodePublishVolume(context.Background(), req)
+			response <- err
+		}()
+		return response
 	}
 
-	// Start first valid request vol1PublishTargetAReq and block until it reaches the Mount
-	go runRequestInBackground(vol1PublishTargetAReq)
-	<-readyToMount
+	// Start first valid request vol1PublishTargetA and block until it reaches the Mount
+	vol1PublishTargetAResp := runRequest(vol1PublishTargetAReq)
+	execVol1PublishTargetA := <-readyToExecute
 
-	// Check that vol1PublishTargetBReq is rejected, due to same volume ID
-	// Also allow vol1PublishTargetBReq to complete, in case it is allowed to Mount
-	mountToRun <- mountmanager.MountSourceAndTarget{
-		Source: vol1PublishTargetBReq.StagingTargetPath,
-		Target: vol1PublishTargetBReq.TargetPath,
-	}
-	_, err := ns.NodePublishVolume(context.Background(), vol1PublishTargetBReq)
-	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.Fatalf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+	// Start vol1PublishTargetB 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.
+	vol1PublishTargetBResp := runRequest(vol1PublishTargetBReq)
+	select {
+	case err := <-vol1PublishTargetBResp:
+		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.Fatalf("Expected error code: %v, got: %v. err : %v", codes.Aborted, serverError.Code(), err)
+			}
+		} else {
+			t.Fatalf("Expected error: %v, got no error", codes.Aborted)
 		}
-	} else {
-		t.Fatalf("Expected error: %v, got no error", codes.Aborted)
+	case <-readyToExecute:
+		t.Fatal("The operation for vol1PublishTargetB should have been aborted, but was started")
 	}
 
-	// Start second valid request vol2PublishTargetCReq
-	go runRequestInBackground(vol2PublishTargetCReq)
-
-	// Allow the vol2PublishTargetCReq to complete, which it can concurrently with vol1PublishTargetAReq
-	mountToRun <- mountmanager.MountSourceAndTarget{
-		Source: vol2PublishTargetCReq.StagingTargetPath,
-		Target: vol2PublishTargetCReq.TargetPath,
-	}
-	if err = <-reqFinished; err != nil {
+	// Start vol2PublishTargetC and allow it to execute to completion. Then check for success.
+	vol2PublishTargetCResp := runRequest(vol2PublishTargetCReq)
+	execVol2PublishTargetC := <-readyToExecute
+	execVol2PublishTargetC <- struct{}{}
+	if err := <-vol2PublishTargetCResp; err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	// To clean up, allow the vol1PublishTargetAReq to complete
-	mountToRun <- mountmanager.MountSourceAndTarget{
-		Source: vol1PublishTargetAReq.StagingTargetPath,
-		Target: vol1PublishTargetAReq.TargetPath,
-	}
-	if err = <-reqFinished; err != nil {
+	// To clean up, allow the vol1PublishTargetA to complete
+	execVol1PublishTargetA <- struct{}{}
+	if err := <-vol1PublishTargetAResp; err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 }

pkg/mount-manager/fake-safe-mounter.go

@@ -52,35 +52,24 @@ func NewFakeSafeMounter() *mount.SafeFormatAndMount {
 
 type FakeBlockingMounter struct {
 	*mount.FakeMounter
-	mountToRun   chan MountSourceAndTarget
-	readyToMount chan struct{}
+	ReadyToExecute chan chan struct{}
 }
 
-type MountSourceAndTarget struct {
-	Source string
-	Target string
-}
-
-// FakeBlockingMounter's method adds two channels to the Mount process in order to provide functionality to finely
-// control the order of execution of Mount calls. readToMount signals that a Mount operation has been called.
-// Then it cycles through the mountToRun channel, waiting for permission to actually make the mount operation.
+// FakeBlockingMounter's method adds functionality to finely control the order of execution of Mount calls.
+// Upon starting a Mount, it passes a chan 'executeMount' into readyToExecute, then blocks on executeMount.
+// The test calling this function can block on readyToExecute to ensure that the operation has started and
+// allowed the Mount to continue by passing a struct into executeMount.
 func (mounter *FakeBlockingMounter) Mount(source string, target string, fstype string, options []string) error {
-	mounter.readyToMount <- struct{}{}
-	for mountToRun := range mounter.mountToRun {
-		if mountToRun.Source == source && mountToRun.Target == target {
-			break
-		} else {
-			mounter.mountToRun <- mountToRun
-		}
-	}
+	executeMount := make(chan struct{})
+	mounter.ReadyToExecute <- executeMount
+	<-executeMount
 	return mounter.FakeMounter.Mount(source, target, fstype, options)
 }
 
-func NewFakeSafeBlockingMounter(mountToRun chan MountSourceAndTarget, readyToMount chan struct{}) *mount.SafeFormatAndMount {
+func NewFakeSafeBlockingMounter(readyToExecute chan chan struct{}) *mount.SafeFormatAndMount {
 	fakeBlockingMounter := &FakeBlockingMounter{
-		FakeMounter:  fakeMounter,
-		mountToRun:   mountToRun,
-		readyToMount: readyToMount,
+		FakeMounter:    fakeMounter,
+		ReadyToExecute: readyToExecute,
 	}
 	return &mount.SafeFormatAndMount{
 		Interface: fakeBlockingMounter,