Repository: kubernetes-sigs/gcp-compute-persistent-disk-csi-driver
Pull request #1313 (2023-07-21 23:46:01)

Cherrypick #1294
/kind bug

Cherry-pick #1294

```release-note
Add option for serializing formatAndMount, including fsck as well as mkfs.
```



Patch:
diff --git a/cmd/gce-pd-csi-driver/main.go b/cmd/gce-pd-csi-driver/main.go
index 0402ac64..479559af 100644
--- a/cmd/gce-pd-csi-driver/main.go
+++ b/cmd/gce-pd-csi-driver/main.go
@@ -63,6 +63,9 @@ var (
 	maxConcurrentFormat     = flag.Int("max-concurrent-format", 1, "The maximum number of concurrent format exec calls")
 	concurrentFormatTimeout = flag.Duration("concurrent-format-timeout", 1*time.Minute, "The maximum duration of a format operation before its concurrency token is released")
 
+	maxConcurrentFormatAndMount = flag.Int("max-concurrent-format-and-mount", 1, "If set then format and mount operations are serialized on each node. This is stronger than max-concurrent-format as it includes fsck and other mount operations")
+	formatAndMountTimeout       = flag.Duration("format-and-mount-timeout", 1*time.Minute, "The maximum duration of a format and mount operation before another such operation will be started. Used only if --serialize-format-and-mount")
+
 	version string
 )
 
@@ -153,6 +156,9 @@ func handle() {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter)
+		if *maxConcurrentFormatAndMount > 0 {
+			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
+		}
 	}
 
 	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, identityServer, controllerServer, nodeServer)
diff --git a/pkg/gce-cloud-provider/compute/fake-gce.go b/pkg/gce-cloud-provider/compute/fake-gce.go
index 4efc3fad..82a5e18f 100644
--- a/pkg/gce-cloud-provider/compute/fake-gce.go
+++ b/pkg/gce-cloud-provider/compute/fake-gce.go
@@ -229,10 +229,10 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
 	switch volKey.Type() {
 	case meta.Zonal:
 		computeDisk.Zone = volKey.Zone
-		computeDisk.SelfLink = fmt.Sprintf("%sprojects/%s/zones/%s/disks/%s", BasePath, project, volKey.Zone, volKey.Name)
+		computeDisk.SelfLink = fmt.Sprintf("projects/%s/zones/%s/disks/%s", project, volKey.Zone, volKey.Name)
 	case meta.Regional:
 		computeDisk.Region = volKey.Region
-		computeDisk.SelfLink = fmt.Sprintf("%sprojects/%s/regions/%s/disks/%s", BasePath, project, volKey.Region, volKey.Name)
+		computeDisk.SelfLink = fmt.Sprintf("projects/%s/regions/%s/disks/%s", project, volKey.Region, volKey.Name)
 	default:
 		return fmt.Errorf("could not create disk, key was neither zonal nor regional, instead got: %v", volKey.String())
 	}
diff --git a/pkg/gce-pd-csi-driver/controller.go b/pkg/gce-pd-csi-driver/controller.go
index 0fd00aa3..d27937b0 100644
--- a/pkg/gce-pd-csi-driver/controller.go
+++ b/pkg/gce-pd-csi-driver/controller.go
@@ -19,7 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
-	neturl "net/url"
+	"regexp"
 	"sort"
 	"strings"
 	"time"
@@ -138,14 +138,6 @@ const (
 
 	// Keys in the volume context.
 	contextForceAttach = "force-attach"
-
-	resourceApiScheme  = "https"
-	resourceApiService = "compute"
-	resourceProject    = "projects"
-)
-
-var (
-	validResourceApiVersions = map[string]bool{"v1": true, "alpha": true, "beta": true}
 )
 
 func isDiskReady(disk *gce.CloudDisk) (bool, error) {
@@ -327,7 +319,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 
 		// 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, zones, params)
+		return generateCreateVolumeResponse(existingDisk, zones, params), nil
 	}
 
 	snapshotID := ""
@@ -442,7 +434,7 @@ func (gceCS *GCEControllerServer) CreateVolume(ctx context.Context, req *csi.Cre
 	}
 
 	klog.V(4).Infof("CreateVolume succeeded for disk %v", volKey)
-	return generateCreateVolumeResponse(disk, zones, params)
+	return generateCreateVolumeResponse(disk, zones, params), nil
 
 }
 
@@ -877,23 +869,13 @@ func (gceCS *GCEControllerServer) ListVolumes(ctx context.Context, req *csi.List
 	entries := []*csi.ListVolumesResponse_Entry{}
 	for i := 0; i+offset < len(gceCS.disks) && i < maxEntries; i++ {
 		d := gceCS.disks[i+offset]
-		diskRsrc, err := getResourceId(d.SelfLink)
-		if err != nil {
-			klog.Warningf("Bad ListVolumes disk resource %s, skipped: %v (%+v)", d.SelfLink, err, d)
-			continue
-		}
 		users := []string{}
 		for _, u := range d.Users {
-			rsrc, err := getResourceId(u)
-			if err != nil {
-				klog.Warningf("Bad ListVolumes user %s, skipped: %v", u, err)
-			} else {
-				users = append(users, rsrc)
-			}
+			users = append(users, cleanSelfLink(u))
 		}
 		entries = append(entries, &csi.ListVolumesResponse_Entry{
 			Volume: &csi.Volume{
-				VolumeId: diskRsrc,
+				VolumeId: cleanSelfLink(d.SelfLink),
 			},
 			Status: &csi.ListVolumesResponse_VolumeStatus{
 				PublishedNodeIds: users,
@@ -1006,10 +988,6 @@ func (gceCS *GCEControllerServer) createPDSnapshot(ctx context.Context, project
 			return nil, common.LoggedError("Failed to create snapshot: ", err)
 		}
 	}
-	snapshotId, err := getResourceId(snapshot.SelfLink)
-	if err != nil {
-		return nil, common.LoggedError(fmt.Sprintf("Cannot extract resource id from snapshot %s", snapshot.SelfLink), err)
-	}
 
 	err = gceCS.validateExistingSnapshot(snapshot, volKey)
 	if err != nil {
@@ -1028,7 +1006,7 @@ func (gceCS *GCEControllerServer) createPDSnapshot(ctx context.Context, project
 
 	return &csi.Snapshot{
 		SizeBytes:      common.GbToBytes(snapshot.DiskSizeGb),
-		SnapshotId:     snapshotId,
+		SnapshotId:     cleanSelfLink(snapshot.SelfLink),
 		SourceVolumeId: volumeID,
 		CreationTime:   timestamp,
 		ReadyToUse:     ready,
@@ -1057,10 +1035,6 @@ func (gceCS *GCEControllerServer) createImage(ctx context.Context, project strin
 			return nil, common.LoggedError("Failed to create image: ", err)
 		}
 	}
-	imageId, err := getResourceId(image.SelfLink)
-	if err != nil {
-		return nil, common.LoggedError(fmt.Sprintf("Cannot extract resource id from snapshot %s", image.SelfLink), err)
-	}
 
 	err = gceCS.validateExistingImage(image, volKey)
 	if err != nil {
@@ -1079,7 +1053,7 @@ func (gceCS *GCEControllerServer) createImage(ctx context.Context, project strin
 
 	return &csi.Snapshot{
 		SizeBytes:      common.GbToBytes(image.DiskSizeGb),
-		SnapshotId:     imageId,
+		SnapshotId:     cleanSelfLink(image.SelfLink),
 		SourceVolumeId: volumeID,
 		CreationTime:   timestamp,
 		ReadyToUse:     ready,
@@ -1091,13 +1065,9 @@ func (gceCS *GCEControllerServer) validateExistingImage(image *compute.Image, vo
 		return fmt.Errorf("disk does not exist")
 	}
 
-	sourceId, err := getResourceId(image.SourceDisk)
-	if err != nil {
-		return fmt.Errorf("failed to get source id from %s: %w", image.SourceDisk, err)
-	}
-	_, sourceKey, err := common.VolumeIDToKey(sourceId)
+	_, sourceKey, err := common.VolumeIDToKey(cleanSelfLink(image.SourceDisk))
 	if err != nil {
-		return fmt.Errorf("failed to get source disk key %s: %w", image.SourceDisk, err)
+		return fmt.Errorf("fail to get source disk key %s, %w", image.SourceDisk, err)
 	}
 
 	if sourceKey.String() != volKey.String() {
@@ -1146,11 +1116,7 @@ func (gceCS *GCEControllerServer) validateExistingSnapshot(snapshot *compute.Sna
 		return fmt.Errorf("disk does not exist")
 	}
 
-	sourceId, err := getResourceId(snapshot.SourceDisk)
-	if err != nil {
-		return fmt.Errorf("failed to get source id from %s: %w", snapshot.SourceDisk, err)
-	}
-	_, sourceKey, err := common.VolumeIDToKey(sourceId)
+	_, sourceKey, err := common.VolumeIDToKey(cleanSelfLink(snapshot.SourceDisk))
 	if err != nil {
 		return fmt.Errorf("fail to get source disk key %s, %w", snapshot.SourceDisk, err)
 	}
@@ -1193,7 +1159,7 @@ func (gceCS *GCEControllerServer) DeleteSnapshot(ctx context.Context, req *csi.D
 	if err != nil {
 		// Cannot get snapshot ID from the passing request
 		// This is a success according to the spec
-		klog.Warningf("Snapshot id does not have the correct format %s: %v", snapshotID, err)
+		klog.Warningf("Snapshot id does not have the correct format %s", snapshotID)
 		return &csi.DeleteSnapshotResponse{}, nil
 	}
 
@@ -1384,7 +1350,7 @@ func (gceCS *GCEControllerServer) getSnapshotByID(ctx context.Context, snapshotI
 				return &csi.ListSnapshotsResponse{}, nil
 			}
 		}
-		e, err := generateDiskImageEntry(image)
+		e, err := generateImageEntry(image)
 		if err != nil {
 			return nil, fmt.Errorf("failed to generate image entry: %w", err)
 		}
@@ -1406,15 +1372,6 @@ func generateDiskSnapshotEntry(snapshot *compute.Snapshot) (*csi.ListSnapshotsRe
 		return nil, fmt.Errorf("Failed to covert creation timestamp: %w", err)
 	}
 
-	snapshotId, err := getResourceId(snapshot.SelfLink)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get snapshot id from %s: %w", snapshot.SelfLink, err)
-	}
-	sourceId, err := getResourceId(snapshot.SourceDisk)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get source id from %s: %w", snapshot.SourceDisk, err)
-	}
-
 	// We ignore the error intentionally here since we are just listing snapshots
 	// TODO: If the snapshot is in "FAILED" state we need to think through what this
 	// should actually look like.
@@ -1423,8 +1380,8 @@ func generateDiskSnapshotEntry(snapshot *compute.Snapshot) (*csi.ListSnapshotsRe
 	entry := &csi.ListSnapshotsResponse_Entry{
 		Snapshot: &csi.Snapshot{
 			SizeBytes:      common.GbToBytes(snapshot.DiskSizeGb),
-			SnapshotId:     snapshotId,
-			SourceVolumeId: sourceId,
+			SnapshotId:     cleanSelfLink(snapshot.SelfLink),
+			SourceVolumeId: cleanSelfLink(snapshot.SourceDisk),
 			CreationTime:   tp,
 			ReadyToUse:     ready,
 		},
@@ -1440,23 +1397,35 @@ func generateDiskImageEntry(image *compute.Image) (*csi.ListSnapshotsResponse_En
 		return nil, fmt.Errorf("failed to covert creation timestamp: %w", err)
 	}
 
-	imageId, err := getResourceId(image.SelfLink)
-	if err != nil {
-		return nil, fmt.Errorf("cannot get image id from %s: %w", image.SelfLink, err)
+	ready, _ := isImageReady(image.Status)
+
+	entry := &csi.ListSnapshotsResponse_Entry{
+		Snapshot: &csi.Snapshot{
+			SizeBytes:      common.GbToBytes(image.DiskSizeGb),
+			SnapshotId:     cleanSelfLink(image.SelfLink),
+			SourceVolumeId: cleanSelfLink(image.SourceDisk),
+			CreationTime:   tp,
+			ReadyToUse:     ready,
+		},
 	}
-	sourceId, err := getResourceId(image.SourceDisk)
+	return entry, nil
+}
+
+func generateImageEntry(image *compute.Image) (*csi.ListSnapshotsResponse_Entry, error) {
+	timestamp, err := parseTimestamp(image.CreationTimestamp)
 	if err != nil {
-		return nil, fmt.Errorf("cannot get source id from %s: %w", image.SourceDisk, err)
+		return nil, fmt.Errorf("Failed to covert creation timestamp: %w", err)
 	}
 
+	// ignore the error intentionally here since we are just listing images
 	ready, _ := isImageReady(image.Status)
 
 	entry := &csi.ListSnapshotsResponse_Entry{
 		Snapshot: &csi.Snapshot{
 			SizeBytes:      common.GbToBytes(image.DiskSizeGb),
-			SnapshotId:     imageId,
-			SourceVolumeId: sourceId,
-			CreationTime:   tp,
+			SnapshotId:     cleanSelfLink(image.SelfLink),
+			SourceVolumeId: cleanSelfLink(image.SourceDisk),
+			CreationTime:   timestamp,
 			ReadyToUse:     ready,
 		},
 	}
@@ -1722,12 +1691,7 @@ func extractVolumeContext(context map[string]string) (*PDCSIContext, error) {
 	return info, nil
 }
 
-func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params common.DiskParameters) (*csi.CreateVolumeResponse, error) {
-	volumeId, err := getResourceId(disk.GetSelfLink())
-	if err != nil {
-		return nil, fmt.Errorf("cannot get volume id from %s: %w", disk.GetSelfLink(), err)
-	}
-
+func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params common.DiskParameters) *csi.CreateVolumeResponse {
 	tops := []*csi.Topology{}
 	for _, zone := range zones {
 		tops = append(tops, &csi.Topology{
@@ -1738,7 +1702,7 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 	createResp := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
 			CapacityBytes:      realDiskSizeBytes,
-			VolumeId:           volumeId,
+			VolumeId:           cleanSelfLink(disk.GetSelfLink()),
 			VolumeContext:      paramsToVolumeContext(params),
 			AccessibleTopology: tops,
 		},
@@ -1777,36 +1741,12 @@ func generateCreateVolumeResponse(disk *gce.CloudDisk, zones []string, params co
 		}
 		createResp.Volume.ContentSource = contentSource
 	}
-	return createResp, nil
+	return createResp
 }
 
-func getResourceId(resourceLink string) (string, error) {
-	url, err := neturl.Parse(resourceLink)
-	if err != nil {
-		return "", fmt.Errorf("Could not parse resource %s: %w", resourceLink, err)
-	}
-	if url.Scheme != resourceApiScheme {
-		return "", fmt.Errorf("Unexpected API scheme for resource %s", resourceLink)
-	}
-
-	// Note that the resource host can basically be anything, if we are running in
-	// a distributed cloud or trusted partner environment.
-
-	// The path should be /compute/VERSION/project/....
-	elts := strings.Split(url.Path, "/")
-	if len(elts) < 4 {
-		return "", fmt.Errorf("Short resource path %s", resourceLink)
-	}
-	if elts[1] != resourceApiService {
-		return "", fmt.Errorf("Bad resource service %s in %s", elts[1], resourceLink)
-	}
-	if _, ok := validResourceApiVersions[elts[2]]; !ok {
-		return "", fmt.Errorf("Bad version %s in %s", elts[2], resourceLink)
-	}
-	if elts[3] != resourceProject {
-		return "", fmt.Errorf("Expected %v to start with %s in resource %s", elts[3:], resourceProject, resourceLink)
-	}
-	return strings.Join(elts[3:], "/"), nil
+func cleanSelfLink(selfLink string) string {
+	r, _ := regexp.Compile("https:\\/\\/www.*apis.com\\/.*(v1|beta|alpha)\\/")
+	return r.ReplaceAllString(selfLink, "")
 }
 
 func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool) (*gce.CloudDisk, error) {
diff --git a/pkg/gce-pd-csi-driver/node.go b/pkg/gce-pd-csi-driver/node.go
index 8349721c..d9fe45fc 100644
--- a/pkg/gce-pd-csi-driver/node.go
+++ b/pkg/gce-pd-csi-driver/node.go
@@ -20,6 +20,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"time"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -46,6 +47,15 @@ type GCENodeServer struct {
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
 	volumeLocks *common.VolumeLocks
+
+	// If set, this semaphore will be used to serialize formatAndMount. It will be raised
+	// when the operation starts, and lowered either when finished, or when
+	// formatAndMountTimeout has expired.
+	//
+	// This is used only on linux (where memory problems for concurrent fsck and mkfs have
+	// been observed).
+	formatAndMountSemaphore chan any
+	formatAndMountTimeout   time.Duration
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -86,6 +96,14 @@ func (ns *GCENodeServer) isVolumePathMounted(path string) bool {
 	return false
 }
 
+func (ns *GCENodeServer) WithSerializedFormatAndMount(timeout time.Duration, maxConcurrent int) *GCENodeServer {
+	if maxConcurrent > 0 {
+		ns.formatAndMountSemaphore = make(chan any, maxConcurrent)
+		ns.formatAndMountTimeout = timeout
+	}
+	return ns
+}
+
 func (ns *GCENodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	// Validate Arguments
 	targetPath := req.GetTargetPath()
@@ -318,7 +336,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 		klog.V(4).Infof("CSI volume is read-only, mounting with extra option ro")
 	}
 
-	err = formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
+	err = ns.formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
 	if err != nil {
 		// If a volume is created from a content source like snapshot or cloning, the filesystem might get marked
 		// as "dirty" even if it is otherwise consistent and ext3/4 will try to restore to a consistent state by replaying
@@ -329,7 +347,7 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 			klog.V(4).Infof("Failed to mount CSI volume read-only, retry mounting with extra option noload")
 
 			options = append(options, "noload")
-			err = formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
+			err = ns.formatAndMount(devicePath, stagingTargetPath, fstype, options, ns.Mounter)
 			if err == nil {
 				klog.V(4).Infof("NodeStageVolume succeeded with \"noload\" option on %v to %s", volumeID, stagingTargetPath)
 				return &csi.NodeStageVolumeResponse{}, nil
diff --git a/pkg/gce-pd-csi-driver/utils_linux.go b/pkg/gce-pd-csi-driver/utils_linux.go
index 8f1f0367..805fb6e8 100644
--- a/pkg/gce-pd-csi-driver/utils_linux.go
+++ b/pkg/gce-pd-csi-driver/utils_linux.go
@@ -20,6 +20,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -47,7 +48,33 @@ func getDevicePath(ns *GCENodeServer, volumeID, partition string) (string, error
 	return devicePath, nil
 }
 
-func formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
+func (ns *GCENodeServer) formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
+	if ns.formatAndMountSemaphore != nil {
+		done := make(chan any)
+		defer close(done)
+
+		// Aquire the semaphore. This will block if another formatAndMount has put an item
+		// into the semaphore channel.
+		ns.formatAndMountSemaphore <- struct{}{}
+
+		go func() {
+			defer func() { <-ns.formatAndMountSemaphore }()
+
+			// Add a timeout where so the semaphore will be released even if
+			// formatAndMount is still working. This allows the node to make progress on
+			// volumes if some error causes one formatAndMount to get stuck. The
+			// motivation for this serialization is to reduce memory usage; if stuck
+			// processes cause OOMs then the containers will be killed and restarted,
+			// including the stuck threads and with any luck making progress.
+			timeout := time.NewTimer(ns.formatAndMountTimeout)
+			defer timeout.Stop()
+
+			select {
+			case <-done:
+			case <-timeout.C:
+			}
+		}()
+	}
 	return m.FormatAndMount(source, target, fstype, options)
 }
 
diff --git a/pkg/gce-pd-csi-driver/utils_windows.go b/pkg/gce-pd-csi-driver/utils_windows.go
index 4661cd89..b12f4f82 100644
--- a/pkg/gce-pd-csi-driver/utils_windows.go
+++ b/pkg/gce-pd-csi-driver/utils_windows.go
@@ -24,7 +24,7 @@ import (
 	mounter "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 )
 
-func formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
+func (ns *GCENodeServer) formatAndMount(source, target, fstype string, options []string, m *mount.SafeFormatAndMount) error {
 	if !strings.EqualFold(fstype, defaultWindowsFsType) {
 		return fmt.Errorf("GCE PD CSI driver can only supports %s file system, it does not support %s", defaultWindowsFsType, fstype)
 	}


Test patch:
diff --git a/pkg/gce-pd-csi-driver/controller_test.go b/pkg/gce-pd-csi-driver/controller_test.go
index fbe0a06a..d97815f1 100644
--- a/pkg/gce-pd-csi-driver/controller_test.go
+++ b/pkg/gce-pd-csi-driver/controller_test.go
@@ -930,11 +930,7 @@ func TestListVolumePagination(t *testing.T) {
 			var d []*gce.CloudDisk
 			for i := 0; i < tc.diskCount; i++ {
 				// Create diskCount dummy disks
-				name := fmt.Sprintf("disk-%v", i)
-				d = append(d, gce.CloudDiskFromV1(&compute.Disk{
-					Name:     name,
-					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/project/zones/zone/disk/%s", name),
-				}))
+				d = append(d, gce.CloudDiskFromV1(&compute.Disk{Name: fmt.Sprintf("%v", i)}))
 			}
 			gceDriver := initGCEDriver(t, d)
 			tok := ""
@@ -992,11 +988,7 @@ func TestListVolumeArgs(t *testing.T) {
 			var d []*gce.CloudDisk
 			for i := 0; i < diskCount; i++ {
 				// Create 600 dummy disks
-				name := fmt.Sprintf("disk-%v", i)
-				d = append(d, gce.CloudDiskFromV1(&compute.Disk{
-					Name:     name,
-					SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/project/zones/zone/disk/%s", name),
-				}))
+				d = append(d, gce.CloudDiskFromV1(&compute.Disk{Name: fmt.Sprintf("%v", i)}))
 			}
 			gceDriver := initGCEDriver(t, d)
 			lvr := &csi.ListVolumesRequest{
@@ -1895,8 +1887,7 @@ func TestCreateVolumeRandomRequisiteTopology(t *testing.T) {
 
 func createZonalCloudDisk(name string) *gce.CloudDisk {
 	return gce.CloudDiskFromV1(&compute.Disk{
-		Name:     name,
-		SelfLink: fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/project/zones/zone/name/%s", name),
+		Name: name,
 	})
 }
 
@@ -3261,12 +3252,11 @@ func TestControllerPublishBackoff(t *testing.T) {
 	}
 }
 
-func TestGetResource(t *testing.T) {
+func TestCleanSelfLink(t *testing.T) {
 	testCases := []struct {
-		name  string
-		in    string
-		want  string
-		error bool
+		name string
+		in   string
+		want string
 	}{
 		{
 			name: "v1 full standard w/ endpoint prefix",
@@ -3284,38 +3274,24 @@ func TestGetResource(t *testing.T) {
 			want: "projects/project/zones/zone/disks/disk",
 		},
 		{
-			name:  "no prefix",
-			in:    "projects/project/zones/zone/disks/disk",
-			error: true,
+			name: "no prefix",
+			in:   "projects/project/zones/zone/disks/disk",
+			want: "projects/project/zones/zone/disks/disk",
 		},
+
 		{
-			name:  "no prefix + project omitted",
-			in:    "zones/zone/disks/disk",
-			error: true,
+			name: "no prefix + project omitted",
+			in:   "zones/zone/disks/disk",
+			want: "zones/zone/disks/disk",
 		},
 		{
-			name: "Compute prefix, www google api",
+			name: "Compute prefix, google api",
 			in:   "https://www.compute.googleapis.com/compute/v1/projects/project/zones/zone/disks/disk",
 			want: "projects/project/zones/zone/disks/disk",
 		},
 		{
 			name: "Compute prefix, partner api",
-			in:   "https://www.compute.partnerapis.com/compute/v1/projects/project/zones/zone/disks/disk",
-			want: "projects/project/zones/zone/disks/disk",
-		},
-		{
-			name: "Compute, alternate googleapis host",
-			in:   "https://content-compute.googleapis.com/compute/v1/projects/project/zones/zone/disks/disk",
-			want: "projects/project/zones/zone/disks/disk",
-		},
-		{
-			name: "Compute, partner host",
-			in:   "https://compute.blahapis.com/compute/v1/projects/project/zones/zone/disks/disk",
-			want: "projects/project/zones/zone/disks/disk",
-		},
-		{
-			name: "Alternate partner host with mtls domain",
-			in:   "https://content-compute.us-central1.rep.mtls.googleapis.com/compute/v1/projects/project/zones/zone/disks/disk",
+			in:   "https://www.compute.PARTNERapis.com/compute/v1/projects/project/zones/zone/disks/disk",
 			want: "projects/project/zones/zone/disks/disk",
 		},
 		{
@@ -3328,66 +3304,13 @@ func TestGetResource(t *testing.T) {
 			in:   "https://www.partnerapis.com/compute/alpha/projects/project/zones/zone/disks/disk",
 			want: "projects/project/zones/zone/disks/disk",
 		},
-		{
-			name: "alpha project",
-			in:   "https://www.googleapis.com/compute/alpha/projects/alphaproject/zones/zone/disks/disk",
-			want: "projects/alphaproject/zones/zone/disks/disk",
-		},
-		{
-			name: "beta project",
-			in:   "https://www.googleapis.com/compute/alpha/projects/betabeta/zones/zone/disks/disk",
-			want: "projects/betabeta/zones/zone/disks/disk",
-		},
-		{
-			name: "v1 project",
-			in:   "https://www.googleapis.com/compute/alpha/projects/projectv1/zones/zone/disks/disk",
-			want: "projects/projectv1/zones/zone/disks/disk",
-		},
-		{
-			name: "random host",
-			in:   "https://npr.org/compute/v1/projects/project/zones/zone/disks/disk",
-			want: "projects/project/zones/zone/disks/disk",
-		},
-		{
-			name:  "no prefix",
-			in:    "projects/project/zones/zone/disks/disk",
-			error: true,
-		},
-		{
-			name:  "bad scheme",
-			in:    "ftp://www.googleapis.com/compute/v1/projects/project/zones/zone/disks/disk",
-			error: true,
-		},
-		{
-			name:  "insecure scheme",
-			in:    "http://www.googleapis.com/compute/v1/projects/project/zones/zone/disks/disk",
-			error: true,
-		},
-		{
-			name:  "bad service",
-			in:    "https://www.googleapis.com/computers/v1/projects/project/zones/zone/disks/disk",
-			error: true,
-		},
-		{
-			name:  "bad version",
-			in:    "https://www..googleapis.com/compute/zeta/projects/project/zones/zone/disks/disk",
-			error: true,
-		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			got, err := getResourceId(tc.in)
-			if tc.error {
-				if err == nil {
-					t.Errorf("Expected error, but got none")
-				}
-			} else {
-				if err != nil {
-					t.Errorf("Unexpected error %v", err)
-				} else if got != tc.want {
-					t.Errorf("Expected cleaned self link: %v, got: %v", tc.want, got)
-				}
+			got := cleanSelfLink(tc.in)
+			if got != tc.want {
+				t.Errorf("Expected cleaned self link: %v, got: %v", tc.want, got)
 			}
 		})
 	}
diff --git a/pkg/gce-pd-csi-driver/node_test.go b/pkg/gce-pd-csi-driver/node_test.go
index 3125c030..cba9f240 100644
--- a/pkg/gce-pd-csi-driver/node_test.go
+++ b/pkg/gce-pd-csi-driver/node_test.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"k8s.io/utils/exec"
 	testingexec "k8s.io/utils/exec/testing"
@@ -57,7 +58,7 @@ func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, dev
 	return gceDriver
 }
 
-func getTestBlockingGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
+func getTestBlockingMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
 	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
@@ -68,6 +69,18 @@ func getTestBlockingGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *
 	return gceDriver
 }
 
+func getTestBlockingFormatAndMountGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
+	gceDriver := GetGCEDriver()
+	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter)).WithSerializedFormatAndMount(5*time.Second, 1)
+
+	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nodeServer)
+	if err != nil {
+		t.Fatalf("Failed to setup GCE Driver: %v", err)
+	}
+	return gceDriver
+}
+
 func makeFakeCmd(fakeCmd *testingexec.FakeCmd, cmd string, args ...string) testingexec.FakeCommandAction {
 	c := cmd
 	a := args
@@ -762,9 +775,7 @@ func TestNodeGetCapabilities(t *testing.T) {
 	}
 }
 
-func TestConcurrentNodeOperations(t *testing.T) {
-	readyToExecute := make(chan chan struct{}, 1)
-	gceDriver := getTestBlockingGCEDriver(t, readyToExecute)
+func runBlockingFormatAndMount(t *testing.T, gceDriver *GCEDriver, readyToExecute chan chan struct{}) {
 	ns := gceDriver.ns
 	tempDir, err := ioutil.TempDir("", "cno")
 	if err != nil {
@@ -844,3 +855,15 @@ func TestConcurrentNodeOperations(t *testing.T) {
 		t.Errorf("Unexpected error: %v", err)
 	}
 }
+
+func TestBlockingMount(t *testing.T) {
+	readyToExecute := make(chan chan struct{}, 1)
+	gceDriver := getTestBlockingMountGCEDriver(t, readyToExecute)
+	runBlockingFormatAndMount(t, gceDriver, readyToExecute)
+}
+
+func TestBlockingFormatAndMount(t *testing.T) {
+	readyToExecute := make(chan chan struct{}, 1)
+	gceDriver := getTestBlockingFormatAndMountGCEDriver(t, readyToExecute)
+	runBlockingFormatAndMount(t, gceDriver, readyToExecute)
+}
diff --git a/test/e2e/utils/utils.go b/test/e2e/utils/utils.go
index f57c09d9..582fb88d 100644
--- a/test/e2e/utils/utils.go
+++ b/test/e2e/utils/utils.go
@@ -53,16 +53,18 @@ func GCEClientAndDriverSetup(instance *remote.InstanceInfo, computeEndpoint stri
 	binPath := path.Join(pkgPath, "bin/gce-pd-csi-driver")
 
 	endpoint := fmt.Sprintf("tcp://localhost:%s", port)
-	computeFlag := ""
+	extra_flags := []string{
+		fmt.Sprintf("--extra-labels=%s=%s", DiskLabelKey, DiskLabelValue),
+		"--max-concurrent-format-and-mount=10", // otherwise the serialization times out.
+	}
 	if computeEndpoint != "" {
-		computeFlag = fmt.Sprintf("--compute-endpoint %s", computeEndpoint)
+		extra_flags = append(extra_flags, fmt.Sprintf("--compute-endpoint %s", computeEndpoint))
 	}
-
 	workspace := remote.NewWorkspaceDir("gce-pd-e2e-")
 	// Log at V(6) as the compute API calls are emitted at that level and it's
 	// useful to see what's happening when debugging tests.
-	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s %s --extra-labels=%s=%s 2> %s/prog.out < /dev/null > /dev/null &'",
-		workspace, endpoint, computeFlag, DiskLabelKey, DiskLabelValue, workspace)
+	driverRunCmd := fmt.Sprintf("sh -c '/usr/bin/nohup %s/gce-pd-csi-driver -v=6 --endpoint=%s %s 2> %s/prog.out < /dev/null > /dev/null &'",
+		workspace, endpoint, strings.Join(extra_flags, " "), workspace)
 
 	config := &remote.ClientConfig{
 		PkgPath:      pkgPath,