[WIP] GKE-MT support for PDCSI

[WIP] Refactor token source code into the token_source.go file

Cleaning up a few logs

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

@@ -117,7 +117,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 +160,67 @@ 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)
+	// listing out regional disks in the region for each project
+	for p, s := range cloud.tenantServiceMap {
+		klog.Infof("Getting regional disks for project: %s", p)
+		rDisks, err := listRegionalDisksForProject(s, p, region, fields, filter)
+		if err != nil {
+			return nil, "", err
+		}
+		disks = append(disks, rDisks...)
+	}
+
+	// listing out zonal disks in all zones of the region for each project
+	for p, s := range cloud.tenantServiceMap {
+		klog.Infof("Getting zonal disks for project: %s", p)
+		zDisks, err := listZonalDisksForProject(s, p, 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 +237,22 @@ func (cloud *CloudProvider) ListInstances(ctx context.Context, fields []googleap
 	}
 	items := []*computev1.Instance{}
 
+	for p, s := range cloud.tenantServiceMap {
+		instances, err := cloud.listInstancesForProject(s, p, 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 +261,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 +897,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 +916,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 +1089,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 +1193,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
 	}

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

@@ -22,13 +22,16 @@ import (
 	"net/url"
 	"os"
 	"runtime"
+	"sync"
 	"time"
 
 	"golang.org/x/oauth2/google"
 	"golang.org/x/time/rate"
 	"google.golang.org/api/option"
 	"gopkg.in/gcfg.v1"
+	"k8s.io/client-go/tools/cache"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute/tenancy"
 
 	"cloud.google.com/go/compute/metadata"
 	rscmgr "cloud.google.com/go/resourcemanager/apiv3"
@@ -81,7 +84,8 @@ var (
 	// snapshotsType is the resource type of compute snapshots.
 	snapshotsType ResourceType = "snapshots"
 	// imagesType is the resource type of compute images.
-	imagesType ResourceType = "images"
+	imagesType         ResourceType = "images"
+	tenantServiceMutex sync.Mutex
 )
 
 // CloudProvider only supports GCE v1/beta Disk APIs. See
@@ -102,6 +106,10 @@ type CloudProvider struct {
 
 	listInstancesConfig ListInstancesConfig
 
+	TenantInformer tenancy.TenantsInformer
+	// tenantServiceMap maintains Compute Services for the default project as well as any tenant projects for any tenant-aware GCE operations
+	tenantServiceMap map[string]*compute.Service
+
 	enableHdHA bool
 }
 
@@ -133,7 +141,7 @@ type ConfigGlobal struct {
 	Zone      string `gcfg:"zone"`
 }
 
-func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig, listInstancesConfig ListInstancesConfig) (*CloudProvider, error) {
+func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath string, computeEndpoint *url.URL, computeEnvironment Environment, waitForAttachConfig WaitForAttachConfig, listInstancesConfig ListInstancesConfig, multiTenancyEnabled bool) (*CloudProvider, error) {
 	configFile, err := readConfig(configPath)
 	if err != nil {
 		return nil, err
@@ -162,10 +170,10 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
 
 	project, zone, err := getProjectAndZone(configFile)
 	if err != nil {
-		return nil, fmt.Errorf("Failed getting Project and Zone: %w", err)
+		return nil, fmt.Errorf("failed getting Project and Zone: %w", err)
 	}
 
-	return &CloudProvider{
+	cp := &CloudProvider{
 		service:             svc,
 		betaService:         betasvc,
 		tokenSource:         tokenSource,
@@ -177,8 +185,69 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
 		// GCP has a rate limit of 600 requests per minute, restricting
 		// here to 8 requests per second.
 		tagsRateLimiter: common.NewLimiter(gcpTagsRequestRateLimit, gcpTagsRequestTokenBucketSize, true),
-	}, nil
+	}
+
+	if multiTenancyEnabled {
+		klog.Info("Setting up multitenancy")
+		// Setup informant for tenant CR to automatically create tenant specific clients with tenant identities
+		ti, err := tenancy.NewTenantsInformer(multiTenancyEnabled)
+		if err != nil {
+			return nil, fmt.Errorf("failed initializing tenant informer: %w", err)
+		}
+		cp.TenantInformer = ti
+		cp.tenantServiceMap = map[string]*compute.Service{}
+		cp.TenantInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj any) {
+				// Handle tenant creation
+				klog.Infof("Tenant %s created", obj)
+
+				tenantMeta, err := tenancy.GetMetadataFromTenantCR(obj)
+				if err != nil {
+					klog.Errorf("error while extracting tenant metadata: %v", err)
+				}
+
+				tenantServiceMutex.Lock()
+				defer tenantServiceMutex.Unlock()
+
+				if _, ok := cp.tenantServiceMap[tenantMeta.ProjectNumber]; ok {
+					klog.Infof("tenant GCE client already exists, skipping GCE client instantiation for tenant(%s) with project number(%s)", tenantMeta.TenantName, tenantMeta.ProjectNumber)
+					return
+				}
+
+				region, err := common.GetRegionFromZones([]string{zone})
+				if err != nil {
+					klog.Errorf("error getting region from zone(%s): %v", zone, err)
+					return
+				}
+				tokenSource, err := NewTenantTokenSource(tenantMeta, region, configFile.Global.TokenURL, configFile.Global.TokenBody)
+				if err != nil {
+					klog.Errorf("error during tenant token generation: %v", err.Error())
+				}
+				klog.Infof("Compute endpoint in add func: %s", computeEndpoint)
+				klog.Infof("Compute environment in add func: %s", computeEnvironment)
+
+				svc, err := createCloudService(ctx, vendorVersion, tokenSource, computeEndpoint, computeEnvironment)
+				if err != nil {
+					klog.Errorf("error while creating compute service with tenant identity: %v", err)
+					return
+				}
+				cp.tenantServiceMap[tenantMeta.ProjectNumber] = svc
+			},
+			UpdateFunc: func(oldObj, newObj any) {},
+			DeleteFunc: func(obj any) {
+				klog.Infof("Tenant %s deleted", obj)
+				tenantMeta, err := tenancy.GetMetadataFromTenantCR(obj)
+				if err != nil {
+					klog.Errorf("error while extracting teantn metadata: %v", err)
+				}
+				tenantServiceMutex.Lock()
+				defer tenantServiceMutex.Unlock()
+				delete(cp.tenantServiceMap, tenantMeta.ProjectNumber)
+			},
+		})
+	}
 
+	return cp, nil
 }
 
 func generateTokenSource(ctx context.Context, configFile *ConfigFile) (oauth2.TokenSource, error) {