Add implementation and tests for Expand capability

Author: davidz627

deploy/kubernetes/overlays/alpha/controller_add_resizer.yaml

@@ -0,0 +1,17 @@
+kind: StatefulSet
+apiVersion: apps/v1
+metadata:
+  name: csi-gce-pd-controller
+spec:
+  template:
+    spec:
+      containers:
+        - name: csi-resizer
+          imagePullPolicy: Always
+          image: quay.io/k8scsi/csi-resizer:canary
+          args:
+            - "--v=5"
+            - "--csi-address=/csi/csi.sock"
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /csi

deploy/kubernetes/overlays/alpha/kustomization.yaml

@@ -4,6 +4,7 @@ bases:
 - ../stable
 patches:
 - controller_add_snapshotter.yaml
+- controller_add_resizer.yaml
 patchesJson6902:
 - target:
     group: rbac.authorization.k8s.io
@@ -13,3 +14,4 @@ patchesJson6902:
   path: rbac_add_snapshots_to_provisioner.yaml
 resources:
 - rbac_add_snapshotter.yaml
+- rbac_add_resizer.yaml

deploy/kubernetes/overlays/alpha/rbac_add_resizer.yaml

@@ -0,0 +1,32 @@
+# Resizer must be able to work with PVCs, PVs, SCs.
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: external-resizer-role
+rules:
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch", "update", "patch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims/status"]
+    verbs: ["update", "patch"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["list", "watch", "create", "update", "patch"]
+
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-resizer-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-controller-sa
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: external-resizer-role
+  apiGroup: rbac.authorization.k8s.io

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

@@ -124,6 +124,17 @@ func (d *CloudDisk) GetSizeGb() int64 {
 	}
 }
 
+// setSizeGb sets the size of the disk used ONLY
+// for testing purposes.
+func (d *CloudDisk) setSizeGb(size int64) {
+	switch d.Type() {
+	case Zonal:
+		d.ZonalDisk.SizeGb = size
+	case Regional:
+		d.RegionalDisk.SizeGb = size
+	}
+}
+
 func (d *CloudDisk) GetZone() string {
 	switch d.Type() {
 	case Zonal:

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

@@ -368,6 +368,18 @@ func (cloud *FakeCloudProvider) CreateSnapshot(ctx context.Context, volKey *meta
 	return snapshotToCreate, nil
 }
 
+func (cloud *FakeCloudProvider) ResizeDisk(ctx context.Context, volKey *meta.Key, requestBytes int64) (int64, error) {
+	disk, ok := cloud.disks[volKey.Name]
+	if !ok {
+		return -1, notFoundError()
+	}
+
+	disk.setSizeGb(common.BytesToGb(requestBytes))
+
+	return requestBytes, nil
+
+}
+
 // Snapshot Methods
 func (cloud *FakeCloudProvider) DeleteSnapshot(ctx context.Context, snapshotName string) error {
 	delete(cloud.snapshots, snapshotName)

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

@@ -50,6 +50,7 @@ type GCECompute interface {
 	GetDiskSourceURI(volKey *meta.Key) string
 	GetDiskTypeURI(volKey *meta.Key, diskType string) string
 	WaitForAttach(ctx context.Context, volKey *meta.Key, instanceZone, instanceName string) error
+	ResizeDisk(ctx context.Context, volKey *meta.Key, requestBytes int64) (int64, error)
 	// Regional Disk Methods
 	GetReplicaZoneURI(zone string) string
 	// Instance Methods
@@ -593,6 +594,65 @@ func (cloud *CloudProvider) CreateSnapshot(ctx context.Context, volKey *meta.Key
 	}
 }
 
+func (cloud *CloudProvider) ResizeDisk(ctx context.Context, volKey *meta.Key, requestBytes int64) (int64, error) {
+	cloudDisk, err := cloud.GetDisk(ctx, volKey)
+	if err != nil {
+		return -1, fmt.Errorf("failed to get disk: %v", err)
+	}
+
+	sizeGb := cloudDisk.GetSizeGb()
+	requestGb := common.BytesToGb(requestBytes)
+
+	// If disk is already of size equal or greater than requested size, we simply return
+	if sizeGb >= requestGb {
+		return requestBytes, nil
+	}
+
+	switch volKey.Type() {
+	case meta.Zonal:
+		return cloud.resizeZonalDisk(ctx, volKey, requestGb)
+	case meta.Regional:
+		return cloud.resizeRegionalDisk(ctx, volKey, requestGb)
+	default:
+		return -1, fmt.Errorf("could not resize disk, key was neither zonal nor regional, instead got: %v", volKey.String())
+	}
+}
+
+func (cloud *CloudProvider) resizeZonalDisk(ctx context.Context, volKey *meta.Key, requestGb int64) (int64, error) {
+	resizeReq := &compute.DisksResizeRequest{
+		SizeGb: requestGb,
+	}
+	op, err := cloud.service.Disks.Resize(cloud.project, volKey.Zone, volKey.Name, resizeReq).Context(ctx).Do()
+	if err != nil {
+		return -1, fmt.Errorf("failed to resize zonal volume %v: %v", volKey.String(), err)
+	}
+
+	err = cloud.waitForZonalOp(ctx, op, volKey.Zone)
+	if err != nil {
+		return -1, fmt.Errorf("failed waiting for op for zonal resize for %s: %v", volKey.String(), err)
+	}
+
+	return requestGb, nil
+}
+
+func (cloud *CloudProvider) resizeRegionalDisk(ctx context.Context, volKey *meta.Key, requestGb int64) (int64, error) {
+	resizeReq := &computebeta.RegionDisksResizeRequest{
+		SizeGb: requestGb,
+	}
+
+	op, err := cloud.betaService.RegionDisks.Resize(cloud.project, volKey.Region, volKey.Name, resizeReq).Context(ctx).Do()
+	if err != nil {
+		return -1, fmt.Errorf("failed to resize regional volume %v: %v", volKey.String(), err)
+	}
+
+	err = cloud.waitForRegionalOp(ctx, op, volKey.Region)
+	if err != nil {
+		return -1, fmt.Errorf("failed waiting for op for regional resize for %s: %v", volKey.String(), err)
+	}
+
+	return requestGb, nil
+}
+
 func (cloud *CloudProvider) createZonalDiskSnapshot(ctx context.Context, volKey *meta.Key, snapshotName string) (*compute.Snapshot, error) {
 	snapshotToCreate := &compute.Snapshot{
 		Name: snapshotName,

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

@@ -614,7 +614,30 @@ func (gceCS *GCEControllerServer) ListSnapshots(ctx context.Context, req *csi.Li
 }
 
 func (gceCS *GCEControllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "ControllerExpandVolume is not yet implemented")
+	volumeID := req.GetVolumeId()
+	if len(volumeID) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "ControllerExpandVolume volume ID must be provided")
+	}
+	capacityRange := req.GetCapacityRange()
+	reqBytes, err := getRequestCapacity(capacityRange)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume capacity range is invalid: %v", err))
+	}
+
+	volKey, err := common.VolumeIDToKey(volumeID)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("ControllerExpandVolume volume ID is invalid: %v", err))
+	}
+
+	resizedGb, err := gceCS.CloudProvider.ResizeDisk(ctx, volKey, reqBytes)
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("ControllerExpandVolume failed to resize disk: %v", err))
+	}
+
+	return &csi.ControllerExpandVolumeResponse{
+		CapacityBytes:         common.GbToBytes(resizedGb),
+		NodeExpansionRequired: true,
+	}, nil
 }
 
 func (gceCS *GCEControllerServer) getSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {

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

@@ -65,10 +65,12 @@ func (gceDriver *GCEDriver) SetupGCEDriver(cloudProvider gce.GCECompute, mounter
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 		csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
 		csi.ControllerServiceCapability_RPC_PUBLISH_READONLY,
+		csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
 	}
 	gceDriver.AddControllerServiceCapabilities(csc)
 	ns := []csi.NodeServiceCapability_RPC_Type{
 		csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
+		csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
 	}
 	gceDriver.AddNodeServiceCapabilities(ns)
 

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

@@ -16,6 +16,7 @@ package gceGCEDriver
 
 import (
 	"context"
+
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -58,6 +59,20 @@ func (gceIdentity *GCEIdentityServer) GetPluginCapabilities(ctx context.Context,
 					},
 				},
 			},
+			{
+				Type: &csi.PluginCapability_VolumeExpansion_{
+					VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+						Type: csi.PluginCapability_VolumeExpansion_ONLINE,
+					},
+				},
+			},
+			{
+				Type: &csi.PluginCapability_VolumeExpansion_{
+					VolumeExpansion: &csi.PluginCapability_VolumeExpansion{
+						Type: csi.PluginCapability_VolumeExpansion_OFFLINE,
+					},
+				},
+			},
 		},
 	}, nil
 }

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

@@ -18,6 +18,7 @@ import (
 	"testing"
 
 	"context"
+
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 )
@@ -58,11 +59,21 @@ func TestGetPluginCapabilities(t *testing.T) {
 	}
 
 	for _, capability := range resp.GetCapabilities() {
-		switch capability.GetService().GetType() {
-		case csi.PluginCapability_Service_CONTROLLER_SERVICE:
-		case csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS:
-		default:
-			t.Fatalf("Unknown capability: %v", capability.GetService().GetType())
+		if capability.GetVolumeExpansion() != nil {
+			switch capability.GetVolumeExpansion().GetType() {
+			case csi.PluginCapability_VolumeExpansion_ONLINE:
+			case csi.PluginCapability_VolumeExpansion_OFFLINE:
+			default:
+				t.Fatalf("Unknown capability: %v", capability.GetVolumeExpansion().GetType())
+			}
+		}
+		if capability.GetService() != nil {
+			switch capability.GetService().GetType() {
+			case csi.PluginCapability_Service_CONTROLLER_SERVICE:
+			case csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS:
+			default:
+				t.Fatalf("Unknown capability: %v", capability.GetService().GetType())
+			}
 		}
 	}
 }