Add support for multi-writer PD

examples/kubernetes/demo-pod-shared-pd.yaml

@@ -0,0 +1,50 @@
+kind: PersistentVolumeClaim
+apiVersion: v1
+metadata:
+  name: podpvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  volumeMode: Block
+  storageClassName: csi-gce-pd
+  resources:
+    requests:
+      storage: 200Gi
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: web-server
+spec:
+  selector:
+    matchLabels:
+      app: web
+  replicas: 2
+  template:
+    metadata:
+      labels:
+        app: web
+    spec:
+      affinity:
+        podAntiAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+          - labelSelector:
+              matchExpressions:
+              - key: app
+                operator: In
+                values:
+                - web
+            topologyKey: "kubernetes.io/hostname"
+      containers:
+        - name: web-server
+          image: nginx
+          volumeDevices:
+            - name: mypvc
+              devicePath: /dev/mypvc
+      volumes:
+        - name: mypvc
+          persistentVolumeClaim:
+            claimName: podpvc
+            readOnly: false

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

@@ -15,12 +15,15 @@ limitations under the License.
 package gcecloudprovider
 
 import (
+	computealpha "google.golang.org/api/compute/v0.alpha"
 	computev1 "google.golang.org/api/compute/v1"
 )
 
 type CloudDisk struct {
-	ZonalDisk    *computev1.Disk
-	RegionalDisk *computev1.Disk
+	ZonalDisk         *computev1.Disk
+	RegionalDisk      *computev1.Disk
+	ZonalAlphaDisk    *computealpha.Disk
+	RegionalAlphaDisk *computealpha.Disk
 }
 
 type CloudDiskType string
@@ -30,6 +33,10 @@ const (
 	Zonal = "zonal"
 	// Regional key type.
 	Regional = "regional"
+	// ZonalAlpha key type.
+	ZonalAlpha = "zonalAlpha"
+	// RegionalAlpha key type.
+	RegionalAlpha = "regionalAlpha"
 	// Global key type.
 	Global = "global"
 )
@@ -46,12 +53,28 @@ func RegionalCloudDisk(disk *computev1.Disk) *CloudDisk {
 	}
 }
 
+func ZonalAlphaCloudDisk(disk *computealpha.Disk) *CloudDisk {
+	return &CloudDisk{
+		ZonalAlphaDisk: disk,
+	}
+}
+
+func RegionalAlphaCloudDisk(disk *computealpha.Disk) *CloudDisk {
+	return &CloudDisk{
+		RegionalAlphaDisk: disk,
+	}
+}
+
 func (d *CloudDisk) Type() CloudDiskType {
 	switch {
 	case d.ZonalDisk != nil:
 		return Zonal
 	case d.RegionalDisk != nil:
 		return Regional
+	case d.ZonalAlphaDisk != nil:
+		return ZonalAlpha
+	case d.RegionalAlphaDisk != nil:
+		return RegionalAlpha
 	default:
 		return Global
 	}
@@ -63,6 +86,10 @@ func (d *CloudDisk) GetUsers() []string {
 		return d.ZonalDisk.Users
 	case Regional:
 		return d.RegionalDisk.Users
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Users
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Users
 	default:
 		return nil
 	}
@@ -74,6 +101,10 @@ func (d *CloudDisk) GetName() string {
 		return d.ZonalDisk.Name
 	case Regional:
 		return d.RegionalDisk.Name
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Name
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Name
 	default:
 		return ""
 	}
@@ -85,6 +116,10 @@ func (d *CloudDisk) GetKind() string {
 		return d.ZonalDisk.Kind
 	case Regional:
 		return d.RegionalDisk.Kind
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Kind
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Kind
 	default:
 		return ""
 	}
@@ -96,6 +131,10 @@ func (d *CloudDisk) GetType() string {
 		return d.ZonalDisk.Type
 	case Regional:
 		return d.RegionalDisk.Type
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Type
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Type
 	default:
 		return ""
 	}
@@ -107,6 +146,10 @@ func (d *CloudDisk) GetSelfLink() string {
 		return d.ZonalDisk.SelfLink
 	case Regional:
 		return d.RegionalDisk.SelfLink
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SelfLink
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SelfLink
 	default:
 		return ""
 	}
@@ -118,6 +161,10 @@ func (d *CloudDisk) GetSizeGb() int64 {
 		return d.ZonalDisk.SizeGb
 	case Regional:
 		return d.RegionalDisk.SizeGb
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SizeGb
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SizeGb
 	default:
 		return -1
 	}
@@ -131,6 +178,10 @@ func (d *CloudDisk) setSizeGb(size int64) {
 		d.ZonalDisk.SizeGb = size
 	case Regional:
 		d.RegionalDisk.SizeGb = size
+	case ZonalAlpha:
+		d.ZonalAlphaDisk.SizeGb = size
+	case RegionalAlpha:
+		d.RegionalAlphaDisk.SizeGb = size
 	}
 }
 
@@ -140,6 +191,10 @@ func (d *CloudDisk) GetZone() string {
 		return d.ZonalDisk.Zone
 	case Regional:
 		return d.RegionalDisk.Zone
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.Zone
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.Zone
 	default:
 		return ""
 	}
@@ -151,7 +206,26 @@ func (d *CloudDisk) GetSnapshotId() string {
 		return d.ZonalDisk.SourceSnapshotId
 	case Regional:
 		return d.RegionalDisk.SourceSnapshotId
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.SourceSnapshotId
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.SourceSnapshotId
 	default:
 		return ""
 	}
 }
+
+func (d *CloudDisk) GetMultiWriter() bool {
+	switch d.Type() {
+	case Zonal:
+		return false
+	case Regional:
+		return false
+	case ZonalAlpha:
+		return d.ZonalAlphaDisk.MultiWriter
+	case RegionalAlpha:
+		return d.RegionalAlphaDisk.MultiWriter
+	default:
+		return false
+	}
+}

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

@@ -155,15 +155,15 @@ func (cloud *FakeCloudProvider) ListSnapshots(ctx context.Context, filter string
 }
 
 // Disk Methods
-func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, volKey *meta.Key) (*CloudDisk, error) {
+func (cloud *FakeCloudProvider) GetDisk(ctx context.Context, volKey *meta.Key, api GCEAPIVersion) (*CloudDisk, error) {
 	disk, ok := cloud.disks[volKey.Name]
 	if !ok {
 		return nil, notFoundError()
 	}
 	return disk, nil
 }
 
-func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, diskType string, reqBytes, limBytes int64) error {
+func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *CloudDisk, diskType string, reqBytes, limBytes int64, multiWriter bool) error {
 	if resp == nil {
 		return fmt.Errorf("disk does not exist")
 	}
@@ -182,15 +182,22 @@ func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *
 		return fmt.Errorf("disk already exists with incompatible type. Need %v. Got %v",
 			diskType, respType[len(respType)-1])
 	}
+
+	// We are assuming here that a multiWriter disk could be used as non-multiWriter
+	if multiWriter && !resp.GetMultiWriter() {
+		return fmt.Errorf("disk already exists with incompatible capability. Need MultiWriter. Got non-MultiWriter")
+	}
+
 	klog.V(4).Infof("Compatible disk already exists")
 	return nil
 }
 
-func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, diskType string, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID, diskEncryptionKmsKey string) error {
+func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, volKey *meta.Key, diskType string, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID, diskEncryptionKmsKey string, multiWriter bool) error {
 	if disk, ok := cloud.disks[volKey.Name]; ok {
 		err := cloud.ValidateExistingDisk(ctx, disk, diskType,
 			int64(capacityRange.GetRequiredBytes()),
-			int64(capacityRange.GetLimitBytes()))
+			int64(capacityRange.GetLimitBytes()),
+			multiWriter)
 		if err != nil {
 			return err
 		}