[WIP] GKE-MT support for PDCSI

Author: cemakd

cmd/gce-pd-csi-driver/main.go

@@ -76,6 +76,7 @@ var (
 	enableStoragePoolsFlag      = flag.Bool("enable-storage-pools", false, "If set to true, the CSI Driver will allow volumes to be provisioned in Storage Pools")
 	enableHdHAFlag              = flag.Bool("allow-hdha-provisioning", false, "If set to true, will allow the driver to provision Hyperdisk-balanced High Availability disks")
 	enableDataCacheFlag         = flag.Bool("enable-data-cache", false, "If set to true, the CSI Driver will allow volumes to be provisioned with Data Cache configuration")
+	enableMultitenancyFlag      = flag.Bool("enable-multitenancy", false, "If set to true, the CSI Driver will support running on multitenant GKE clusters")
 	nodeName                    = flag.String("node-name", "", "The node this driver is running on")
 
 	multiZoneVolumeHandleDiskTypesFlag = flag.String("multi-zone-volume-handle-disk-types", "", "Comma separated list of allowed disk types that can use the multi-zone volumeHandle. Used only if --multi-zone-volume-handle-enable")
@@ -221,10 +222,15 @@ func handle() {
 	// Initialize requirements for the controller service
 	var controllerServer *driver.GCEControllerServer
 	if *runControllerService {
-		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment, waitForAttachConfig, listInstancesConfig)
+		cloudProvider, err := gce.CreateCloudProvider(ctx, version, *cloudConfigFilePath, computeEndpoint, computeEnvironment, waitForAttachConfig, listInstancesConfig, *enableMultitenancyFlag)
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %v", err.Error())
 		}
+		if *enableMultitenancyFlag {
+			ctx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			go cloudProvider.TenantInformer.Run(ctx.Done())
+		}
 		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
 		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig, *enableHdHAFlag)

deploy/kubernetes/base/controller/controller.yaml

@@ -145,6 +145,7 @@ spec:
             - "--supports-dynamic-iops-provisioning=hyperdisk-balanced,hyperdisk-extreme"
             - "--supports-dynamic-throughput-provisioning=hyperdisk-balanced,hyperdisk-throughput,hyperdisk-ml"
             - --enable-data-cache
+            - --enable-multitenancy
           command:
             - /gce-pd-csi-driver
           env:

deploy/kubernetes/base/node_linux/node.yaml

@@ -47,6 +47,7 @@ spec:
             - "--endpoint=unix:/csi/csi.sock"
             - "--run-controller-service=false"
             - "--enable-data-cache"
+            - "--enable-multitenancy"
             - "--node-name=$(KUBE_NODE_NAME)"
           securityContext:
             privileged: true

deploy/kubernetes/images/stable-master/image.yaml

@@ -49,7 +49,8 @@ metadata:
   name: imagetag-gcepd-driver
 imageTag:
   name: gke.gcr.io/gcp-compute-persistent-disk-csi-driver
-  # pdImagePlaceholder in test/k8s-integration/main.go is updated automatically with the newTag
-  newName: registry.k8s.io/cloud-provider-gcp/gcp-compute-persistent-disk-csi-driver
-  newTag: "v1.17.2"
+  # Don't change stable image without changing pdImagePlaceholder in
+  # test/k8s-integration/main.go
+  newName: us-central1-docker.pkg.dev/enginakdemir-gke-dev/csi-dev/gcp-compute-persistent-disk-csi-driver
+  newTag: "latest"
 ---

go.mod

@@ -41,6 +41,7 @@ require (
 	k8s.io/mount-utils v0.32.3
 	k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738
 	sigs.k8s.io/boskos v0.0.0-20220711194915-6cb8a6fb2dd1
+	sigs.k8s.io/yaml v1.4.0
 )
 
 require (
@@ -116,7 +117,6 @@ require (
 	k8s.io/test-infra v0.0.0-20210730160938-8ad9b8c53bd8 // indirect
 	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.6.0 // indirect
-	sigs.k8s.io/yaml v1.4.0 // indirect
 )
 
 replace k8s.io/client-go => k8s.io/client-go v0.32.2

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

@@ -378,7 +378,7 @@ func (cloud *FakeCloudProvider) InsertInstance(instance *computev1.Instance, ins
 	return
 }
 
-func (cloud *FakeCloudProvider) GetInstanceOrError(ctx context.Context, instanceZone, instanceName string) (*computev1.Instance, error) {
+func (cloud *FakeCloudProvider) GetInstanceOrError(ctx context.Context, project, instanceZone, instanceName string) (*computev1.Instance, error) {
 	instance, ok := cloud.instances[instanceName]
 	if !ok {
 		return nil, notFoundError()

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

@@ -32,6 +32,7 @@ import (
 	"github.com/googleapis/gax-go/v2/apierror"
 	"golang.org/x/oauth2"
 	computebeta "google.golang.org/api/compute/v0.beta"
+	"google.golang.org/api/compute/v1"
 	computev1 "google.golang.org/api/compute/v1"
 	"google.golang.org/api/googleapi"
 	"google.golang.org/api/iterator"
@@ -117,7 +118,7 @@ type GCECompute interface {
 	// Regional Disk Methods
 	GetReplicaZoneURI(project string, zone string) string
 	// Instance Methods
-	GetInstanceOrError(ctx context.Context, instanceZone, instanceName string) (*computev1.Instance, error)
+	GetInstanceOrError(ctx context.Context, project, instanceZone, instanceName string) (*computev1.Instance, error)
 	// Zone Methods
 	ListZones(ctx context.Context, region string) ([]string, error)
 	ListSnapshots(ctx context.Context, filter string) ([]*computev1.Snapshot, string, error)
@@ -160,40 +161,75 @@ func (cloud *CloudProvider) listDisksInternal(ctx context.Context, fields []goog
 	if err != nil {
 		return nil, "", err
 	}
-	items := []*computev1.Disk{}
+	disks := []*computev1.Disk{}
 
 	// listing out regional disks in the region
-	rlCall := cloud.service.RegionDisks.List(cloud.project, region)
+	rDisks, err := listRegionalDisksForProject(cloud.service, cloud.project, region, fields, filter)
+	if err != nil {
+		return nil, "", err
+	}
+	disks = append(disks, rDisks...)
+	for tProject, tService := range cloud.tenantServiceMap {
+		rDisks, err := listRegionalDisksForProject(tService, tProject, region, fields, filter)
+		if err != nil {
+			return nil, "", err
+		}
+		disks = append(disks, rDisks...)
+	}
+
+	// listing out zonal disks in all zones of the region
+	zDisks, err := listZonalDisksForProject(cloud.service, cloud.project, zones, fields, filter)
+	if err != nil {
+		return nil, "", err
+	}
+	disks = append(disks, zDisks...)
+	for tProject, tService := range cloud.tenantServiceMap {
+		zDisks, err := listZonalDisksForProject(tService, tProject, zones, fields, filter)
+		if err != nil {
+			return nil, "", err
+		}
+		disks = append(disks, zDisks...)
+	}
+
+	return disks, "", nil
+}
+
+func listRegionalDisksForProject(service *computev1.Service, project string, region string, fields []googleapi.Field, filter string) ([]*computev1.Disk, error) {
+	items := []*computev1.Disk{}
+	rlCall := service.RegionDisks.List(project, region)
 	rlCall.Fields(fields...)
 	rlCall.Filter(filter)
 	nextPageToken := "pageToken"
 	for nextPageToken != "" {
 		rDiskList, err := rlCall.Do()
 		if err != nil {
-			return nil, "", err
+			return nil, err
 		}
 		items = append(items, rDiskList.Items...)
 		nextPageToken = rDiskList.NextPageToken
 		rlCall.PageToken(nextPageToken)
 	}
+	return items, nil
+}
 
-	// listing out zonal disks in all zones of the region
+func listZonalDisksForProject(service *computev1.Service, project string, zones []string, fields []googleapi.Field, filter string) ([]*computev1.Disk, error) {
+	items := []*computev1.Disk{}
 	for _, zone := range zones {
-		lCall := cloud.service.Disks.List(cloud.project, zone)
+		lCall := service.Disks.List(project, zone)
 		lCall.Fields(fields...)
 		lCall.Filter(filter)
 		nextPageToken := "pageToken"
 		for nextPageToken != "" {
 			diskList, err := lCall.Do()
 			if err != nil {
-				return nil, "", err
+				return nil, err
 			}
 			items = append(items, diskList.Items...)
 			nextPageToken = diskList.NextPageToken
 			lCall.PageToken(nextPageToken)
 		}
 	}
-	return items, "", nil
+	return items, nil
 }
 
 // ListInstances lists instances based on maxEntries and pageToken for the project and region
@@ -210,8 +246,27 @@ func (cloud *CloudProvider) ListInstances(ctx context.Context, fields []googleap
 	}
 	items := []*computev1.Instance{}
 
+	instances, err := cloud.listInstancesForProject(cloud.service, cloud.project, zones, fields)
+	if err != nil {
+		return nil, "", err
+	}
+	items = append(items, instances...)
+	for tProject, tService := range cloud.tenantServiceMap {
+		instances, err := cloud.listInstancesForProject(tService, tProject, zones, fields)
+		if err != nil {
+			return nil, "", err
+		}
+		items = append(items, instances...)
+	}
+
+	return items, "", nil
+}
+
+func (cloud *CloudProvider) listInstancesForProject(service *computev1.Service, project string, zones []string, fields []googleapi.Field) ([]*computev1.Instance, error) {
+	items := []*computev1.Instance{}
+
 	for _, zone := range zones {
-		lCall := cloud.service.Instances.List(cloud.project, zone)
+		lCall := service.Instances.List(project, zone)
 		for _, filter := range cloud.listInstancesConfig.Filters {
 			lCall = lCall.Filter(filter)
 		}
@@ -220,15 +275,14 @@ func (cloud *CloudProvider) ListInstances(ctx context.Context, fields []googleap
 		for nextPageToken != "" {
 			instancesList, err := lCall.Do()
 			if err != nil {
-				return nil, "", err
+				return nil, err
 			}
 			items = append(items, instancesList.Items...)
 			nextPageToken = instancesList.NextPageToken
 			lCall.PageToken(nextPageToken)
 		}
 	}
-
-	return items, "", nil
+	return items, nil
 }
 
 // RepairUnderspecifiedVolumeKey will query the cloud provider and check each zone for the disk specified
@@ -857,7 +911,11 @@ func (cloud *CloudProvider) AttachDisk(ctx context.Context, project string, volK
 		ForceAttach: forceAttach,
 	}
 
-	op, err := cloud.service.Instances.AttachDisk(project, instanceZone, instanceName, attachedDiskV1).Context(ctx).ForceAttach(forceAttach).Do()
+	service := cloud.service
+	if _, ok := cloud.tenantServiceMap[project]; ok {
+		service = cloud.tenantServiceMap[project]
+	}
+	op, err := service.Instances.AttachDisk(project, instanceZone, instanceName, attachedDiskV1).Context(ctx).ForceAttach(forceAttach).Do()
 	if err != nil {
 		return fmt.Errorf("failed cloud service attach disk call: %w", err)
 	}
@@ -872,7 +930,11 @@ func (cloud *CloudProvider) AttachDisk(ctx context.Context, project string, volK
 
 func (cloud *CloudProvider) DetachDisk(ctx context.Context, project, deviceName, instanceZone, instanceName string) error {
 	klog.V(5).Infof("Detaching disk %v from %v", deviceName, instanceName)
-	op, err := cloud.service.Instances.DetachDisk(project, instanceZone, instanceName, deviceName).Context(ctx).Do()
+	service := cloud.service
+	if _, ok := cloud.tenantServiceMap[project]; ok {
+		service = cloud.tenantServiceMap[project]
+	}
+	op, err := service.Instances.DetachDisk(project, instanceZone, instanceName, deviceName).Context(ctx).Do()
 	if err != nil {
 		return err
 	}
@@ -1041,7 +1103,7 @@ func (cloud *CloudProvider) waitForAttachOnInstance(ctx context.Context, project
 	start := time.Now()
 	return wait.ExponentialBackoff(AttachDiskBackoff, func() (bool, error) {
 		klog.V(6).Infof("Polling instances.get for attach of disk %v to instance %v to complete for %v", volKey.Name, instanceName, time.Since(start))
-		instance, err := cloud.GetInstanceOrError(ctx, instanceZone, instanceName)
+		instance, err := cloud.GetInstanceOrError(ctx, project, instanceZone, instanceName)
 		if err != nil {
 			return false, fmt.Errorf("GetInstance failed to get instance: %w", err)
 		}
@@ -1145,10 +1207,13 @@ func opIsDone(op *computev1.Operation) (bool, error) {
 	return true, nil
 }
 
-func (cloud *CloudProvider) GetInstanceOrError(ctx context.Context, instanceZone, instanceName string) (*computev1.Instance, error) {
+func (cloud *CloudProvider) GetInstanceOrError(ctx context.Context, project, instanceZone, instanceName string) (*computev1.Instance, error) {
 	klog.V(5).Infof("Getting instance %v from zone %v", instanceName, instanceZone)
-	project := cloud.project
-	instance, err := cloud.service.Instances.Get(project, instanceZone, instanceName).Do()
+	service := cloud.service
+	if _, ok := cloud.tenantServiceMap[project]; ok {
+		service = cloud.tenantServiceMap[project]
+	}
+	instance, err := service.Instances.Get(project, instanceZone, instanceName).Do()
 	if err != nil {
 		return nil, err
 	}
@@ -1426,6 +1491,22 @@ func (cloud *CloudProvider) waitForSnapshotCreation(ctx context.Context, project
 	}
 }
 
+func (cloud *CloudProvider) getTenantService(ctx context.Context, project string) (*compute.Service, error) {
+	s, exists := cloud.tenantServiceMap[project]
+	if !exists {
+		var err error
+		s, err = cloud.createTenantService(project)
+		if err != nil {
+			return nil, fmt.Errorf("error during tenant compute client creation: %w", err)
+		}
+	}
+	return s, nil
+}
+
+func (cloud *CloudProvider) createTenantService(project string) (*compute.Service, error) {
+	return nil, nil
+}
+
 // getResourceManagerTags returns the map of tag keys and values. The tag keys are in the form `tagKeys/{tag_key_id}`
 // and the tag values are in the format `tagValues/456`.
 func getResourceManagerTags(ctx context.Context, tokenSource oauth2.TokenSource, tagsMap map[string]string) (map[string]string, error) {