feat: Provide hostname for zonal disk insert API

CL contains changes to pass in hostname for zonal disk.insert API as a
part of location hint field. Code changes for regional disk API will be
part of future CLs.

Author: Rishita-Golla

Diff for: cmd/gce-pd-csi-driver/main.go

@@ -27,6 +27,8 @@ import (
 	"strings"
 	"time"
 
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/strings/slices"
 
@@ -72,6 +74,7 @@ var (
 	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")
 	fallbackRequisiteZonesFlag  = flag.String("fallback-requisite-zones", "", "Comma separated list of requisite zones that will be used if there are not sufficient zones present in requisite topologies when provisioning a disk")
 	enableStoragePoolsFlag      = flag.Bool("enable-storage-pools", false, "If set to true, the CSI Driver will allow volumes to be provisioned in Storage Pools")
+	enableVMLocationHint        = flag.Bool("enable-vm-location-hint", false, "If set to true, the location hint field for create volume request will have hostname set")
 
 	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")
 	multiZoneVolumeHandleEnableFlag    = flag.Bool("multi-zone-volume-handle-enable", false, "If set to true, the multi-zone volumeHandle feature will be enabled")
@@ -244,7 +247,20 @@ func handle() {
 		}
 	}
 
-	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, extraTags, identityServer, controllerServer, nodeServer)
+	var kubeClient *kubernetes.Clientset
+	if *enableVMLocationHint {
+		cfg, err := rest.InClusterConfig()
+		if err != nil {
+			klog.Fatalf("Could not fetch in-cluster config: %v", err.Error())
+		}
+
+		kubeClient, err = kubernetes.NewForConfig(cfg)
+		if err != nil {
+			klog.Fatalf("Could not fetch in-cluster client: %v", err.Error())
+		}
+	}
+
+	err = gceDriver.SetupGCEDriver(driverName, version, extraVolumeLabels, extraTags, kubeClient, identityServer, controllerServer, nodeServer)
 	if err != nil {
 		klog.Fatalf("Failed to initialize GCE CSI Driver: %v", err.Error())
 	}

Diff for: pkg/gce-cloud-provider/compute/fake-gce.go

@@ -221,7 +221,7 @@ func (cloud *FakeCloudProvider) ValidateExistingDisk(ctx context.Context, resp *
 	return ValidateDiskParameters(resp, params)
 }
 
-func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
+func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode, hostName string) error {
 	if disk, ok := cloud.disks[volKey.String()]; ok {
 		err := cloud.ValidateExistingDisk(ctx, disk, params,
 			int64(capacityRange.GetRequiredBytes()),
@@ -242,6 +242,7 @@ func (cloud *FakeCloudProvider) InsertDisk(ctx context.Context, project string,
 		Labels:                params.Labels,
 		ProvisionedIops:       params.ProvisionedIOPSOnCreate,
 		ProvisionedThroughput: params.ProvisionedThroughputOnCreate,
+		LocationHint:          cloud.GetLocationHintURI(project, volKey.Zone, hostName),
 	}
 
 	if snapshotID != "" {
@@ -382,6 +383,10 @@ func (cloud *FakeCloudProvider) GetDiskTypeURI(project string, volKey *meta.Key,
 	}
 }
 
+func (cloud *FakeCloudProvider) GetLocationHintURI(project, zone, hostName string) string {
+	return fmt.Sprintf(locationHintURITemplate, project, zone, hostName)
+}
+
 func (cloud *FakeCloudProvider) getZonalDiskTypeURI(project, zone, diskType string) string {
 	return fmt.Sprintf(diskTypeURITemplateSingleZone, project, zone, diskType)
 }

Diff for: pkg/gce-cloud-provider/compute/gce-compute.go

@@ -102,7 +102,7 @@ type GCECompute interface {
 	GetDisk(ctx context.Context, project string, volumeKey *meta.Key, gceAPIVersion GCEAPIVersion) (*CloudDisk, error)
 	RepairUnderspecifiedVolumeKey(ctx context.Context, project string, volumeKey *meta.Key) (string, *meta.Key, error)
 	ValidateExistingDisk(ctx context.Context, disk *CloudDisk, params common.DiskParameters, reqBytes, limBytes int64, multiWriter bool) error
-	InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error
+	InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode, hostName string) error
 	DeleteDisk(ctx context.Context, project string, volumeKey *meta.Key) error
 	UpdateDisk(ctx context.Context, project string, volKey *meta.Key, existingDisk *CloudDisk, params common.ModifyVolumeParameters) error
 	AttachDisk(ctx context.Context, project string, volKey *meta.Key, readWrite, diskType, instanceZone, instanceName string, forceAttach bool) error
@@ -436,7 +436,7 @@ func ValidateDiskParameters(disk *CloudDisk, params common.DiskParameters) error
 	return nil
 }
 
-func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
+func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode, hostName string) error {
 	klog.V(5).Infof("Inserting disk %v", volKey)
 
 	description, err := encodeTags(params.Tags)
@@ -449,7 +449,7 @@ func (cloud *CloudProvider) InsertDisk(ctx context.Context, project string, volK
 		if description == "" {
 			description = "Disk created by GCE-PD CSI Driver"
 		}
-		return cloud.insertZonalDisk(ctx, project, volKey, params, capBytes, capacityRange, snapshotID, volumeContentSourceVolumeID, description, multiWriter, accessMode)
+		return cloud.insertZonalDisk(ctx, project, volKey, params, capBytes, capacityRange, snapshotID, volumeContentSourceVolumeID, description, multiWriter, accessMode, hostName)
 	case meta.Regional:
 		if description == "" {
 			description = "Regional disk created by GCE-PD CSI Driver"
@@ -772,7 +772,7 @@ func (cloud *CloudProvider) insertZonalDisk(
 	volumeContentSourceVolumeID string,
 	description string,
 	multiWriter bool,
-	accessMode string) error {
+	accessMode, hostName string) error {
 	var (
 		err           error
 		opName        string
@@ -788,6 +788,7 @@ func (cloud *CloudProvider) insertZonalDisk(
 		Description: description,
 		Type:        cloud.GetDiskTypeURI(project, volKey, params.DiskType),
 		Labels:      params.Labels,
+		LocationHint: cloud.GetLocationHintURI(project, volKey.Zone, hostName),
 	}
 
 	if params.ProvisionedIOPSOnCreate > 0 {
@@ -1107,6 +1108,10 @@ func (cloud *CloudProvider) GetDiskTypeURI(project string, volKey *meta.Key, dis
 	}
 }
 
+func (cloud *CloudProvider) GetLocationHintURI(project, zone, hostName string) string {
+	return fmt.Sprintf(locationHintURITemplate, project, zone, hostName)
+}
+
 func (cloud *CloudProvider) getZonalDiskTypeURI(project string, zone, diskType string) string {
 	return cloud.service.BasePath + fmt.Sprintf(diskTypeURITemplateSingleZone, project, zone, diskType)
 }

Diff for: pkg/gce-cloud-provider/compute/gce.go

@@ -51,6 +51,7 @@ const (
 	diskTypeURITemplateRegional     = "projects/%s/regions/%s/diskTypes/%s" // {gce.projectID}/regions/{disk.Region}/diskTypes/{disk.Type}"
 
 	regionURITemplate = "projects/%s/regions/%s"
+	locationHintURITemplate = "projects/%s/zone/%s/instance/%s"
 
 	replicaZoneURITemplateSingleZone             = "projects/%s/zones/%s" // {gce.projectID}/zones/{disk.Zone}
 	EnvironmentStaging               Environment = "staging"

Diff for: pkg/gce-pd-csi-driver/controller.go

@@ -38,6 +38,8 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/utils/strings/slices"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
@@ -208,6 +210,8 @@ const (
 	listDisksUsersField = googleapi.Field("items/users")
 
 	readOnlyManyAccessMode = "READ_ONLY_MANY"
+
+	annSelectedNode = "volume.kubernetes.io/selected-node"
 )
 
 var (
@@ -720,21 +724,23 @@ func (gceCS *GCEControllerServer) createSingleDisk(ctx context.Context, req *csi
 	// Create the disk
 	var disk *gce.CloudDisk
 	name := req.GetName()
+	pvcName, pvcNamespace := req.Parameters[common.ParameterKeyPVCName], req.Parameters[common.ParameterKeyPVCNamespace]
+	hostName := getHostNameFromPVC(ctx, pvcName, pvcNamespace, gceCS.Driver.kubeClient)
 
 	switch params.ReplicationType {
 	case replicationTypeNone:
 		if len(zones) != 1 {
 			return nil, status.Errorf(codes.Internal, "CreateVolume failed to get a single zone for creating zonal disk, instead got: %v", zones)
 		}
-		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+		disk, err = createSingleZoneDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode, hostName)
 		if err != nil {
 			return nil, common.LoggedError("CreateVolume failed to create single zonal disk "+name+": ", err)
 		}
 	case replicationTypeRegionalPD:
 		if len(zones) != 2 {
 			return nil, status.Errorf(codes.Internal, "CreateVolume failed to get a 2 zones for creating regional disk, instead got: %v", zones)
 		}
-		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+		disk, err = createRegionalDisk(ctx, gceCS.CloudProvider, name, zones, params, capacityRange, capBytes, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode, "")
 		if err != nil {
 			return nil, common.LoggedError("CreateVolume failed to create regional disk "+name+": ", err)
 		}
@@ -872,6 +878,24 @@ func (gceCS *GCEControllerServer) DeleteVolume(ctx context.Context, req *csi.Del
 	return gceCS.deleteSingleDeviceDisk(ctx, req, project, volKey)
 }
 
+func getHostNameFromPVC(ctx context.Context, pvcName, pvcNamespace string, kubeClient *kubernetes.Clientset) string {
+	if kubeClient == nil {
+		// Client is initialized only when location hint is enabled
+		return ""
+	}
+
+	pvc, err := kubeClient.CoreV1().PersistentVolumeClaims(pvcNamespace).Get(ctx, pvcName, metav1.GetOptions{})
+	if err != nil {
+		klog.Errorf("Failed to get persistent volume claims: %v", err)
+		// Don't log the error as create volume request can be continued without hostname
+		return ""
+	}
+	if hostName, ok := pvc.Annotations[annSelectedNode]; ok {
+		return hostName
+	}
+	return ""
+}
+
 func getGCEApiVersion(multiWriter bool) gce.GCEAPIVersion {
 	if multiWriter {
 		return gce.GCEAPIVersionBeta
@@ -2482,7 +2506,7 @@ func getResourceId(resourceLink string) (string, error) {
 	return strings.Join(elts[3:], "/"), nil
 }
 
-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, accessMode string) (*gce.CloudDisk, error) {
+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, accessMode, hostName string) (*gce.CloudDisk, error) {
 	project := cloudProvider.GetDefaultProject()
 	region, err := common.GetRegionFromZones(zones)
 	if err != nil {
@@ -2495,7 +2519,7 @@ func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name
 			fullyQualifiedReplicaZones, cloudProvider.GetReplicaZoneURI(project, replicaZone))
 	}
 
-	err = cloudProvider.InsertDisk(ctx, project, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+	err = cloudProvider.InsertDisk(ctx, project, meta.RegionalKey(name, region), params, capBytes, capacityRange, fullyQualifiedReplicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode, hostName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert regional disk: %w", err)
 	}
@@ -2512,13 +2536,13 @@ func createRegionalDisk(ctx context.Context, cloudProvider gce.GCECompute, name
 	return disk, nil
 }
 
-func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) (*gce.CloudDisk, error) {
+func createSingleZoneDisk(ctx context.Context, cloudProvider gce.GCECompute, name string, zones []string, params common.DiskParameters, capacityRange *csi.CapacityRange, capBytes int64, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode, hostName string) (*gce.CloudDisk, error) {
 	project := cloudProvider.GetDefaultProject()
 	if len(zones) != 1 {
 		return nil, fmt.Errorf("got wrong number of zones for zonal create volume: %v", len(zones))
 	}
 	diskZone := zones[0]
-	err := cloudProvider.InsertDisk(ctx, project, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+	err := cloudProvider.InsertDisk(ctx, project, meta.ZonalKey(name, diskZone), params, capBytes, capacityRange, nil, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode, hostName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to insert zonal disk: %w", err)
 	}

Diff for: pkg/gce-pd-csi-driver/controller_test.go

@@ -1596,12 +1596,12 @@ func (cloud *FakeCloudProviderInsertDiskErr) AddDiskForErr(volKey *meta.Key, err
 	cloud.insertDiskErrors[volKey.String()] = err
 }
 
-func (cloud *FakeCloudProviderInsertDiskErr) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode string) error {
+func (cloud *FakeCloudProviderInsertDiskErr) InsertDisk(ctx context.Context, project string, volKey *meta.Key, params common.DiskParameters, capBytes int64, capacityRange *csi.CapacityRange, replicaZones []string, snapshotID string, volumeContentSourceVolumeID string, multiWriter bool, accessMode, hostName string) error {
 	if err, ok := cloud.insertDiskErrors[volKey.String()]; ok {
 		return err
 	}
 
-	return cloud.FakeCloudProvider.InsertDisk(ctx, project, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode)
+	return cloud.FakeCloudProvider.InsertDisk(ctx, project, volKey, params, capBytes, capacityRange, replicaZones, snapshotID, volumeContentSourceVolumeID, multiWriter, accessMode, hostName)
 }
 
 func TestMultiZoneVolumeCreationErrHandling(t *testing.T) {
@@ -1949,7 +1949,7 @@ func TestVolumeModifyOperation(t *testing.T) {
 			t.Fatalf("Failed convert key: %v", err)
 		}
 
-		err = fcp.InsertDisk(context.Background(), project, volKey, *tc.params, 200000, nil, nil, "", "", false, "")
+		err = fcp.InsertDisk(context.Background(), project, volKey, *tc.params, 200000, nil, nil, "", "", false, "", "")
 		if err != nil {
 			t.Fatalf("Failed to insert disk: %v", err)
 		}

Diff for: pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -21,6 +21,7 @@ import (
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
@@ -37,6 +38,7 @@ type GCEDriver struct {
 	vendorVersion     string
 	extraVolumeLabels map[string]string
 	extraTags         map[string]string
+	kubeClient        *kubernetes.Clientset // In-cluster client
 
 	ids *GCEIdentityServer
 	ns  *GCENodeServer
@@ -51,7 +53,7 @@ func GetGCEDriver() *GCEDriver {
 	return &GCEDriver{}
 }
 
-func (gceDriver *GCEDriver) SetupGCEDriver(name, vendorVersion string, extraVolumeLabels map[string]string, extraTags map[string]string, identityServer *GCEIdentityServer, controllerServer *GCEControllerServer, nodeServer *GCENodeServer) error {
+func (gceDriver *GCEDriver) SetupGCEDriver(name, vendorVersion string, extraVolumeLabels map[string]string, extraTags map[string]string, kubeClient *kubernetes.Clientset, identityServer *GCEIdentityServer, controllerServer *GCEControllerServer, nodeServer *GCENodeServer) error {
 	if name == "" {
 		return fmt.Errorf("Driver name missing")
 	}
@@ -87,6 +89,7 @@ func (gceDriver *GCEDriver) SetupGCEDriver(name, vendorVersion string, extraVolu
 	gceDriver.vendorVersion = vendorVersion
 	gceDriver.extraVolumeLabels = extraVolumeLabels
 	gceDriver.extraTags = extraTags
+	gceDriver.kubeClient = kubeClient
 	gceDriver.ids = identityServer
 	gceDriver.cs = controllerServer
 	gceDriver.ns = nodeServer

Diff for: pkg/gce-pd-csi-driver/gce-pd-driver_test.go

@@ -56,7 +56,7 @@ func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute)
 	}
 
 	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration, fallbackRequisiteZones, enableStoragePools, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig)
-	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, controllerServer, nil)
+	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, nil, controllerServer, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}

Diff for: pkg/gce-pd-csi-driver/identity_test.go

@@ -26,7 +26,7 @@ func TestGetPluginInfo(t *testing.T) {
 	vendorVersion := "test-vendor"
 	gceDriver := GetGCEDriver()
 	identityServer := NewIdentityServer(gceDriver)
-	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, identityServer, nil, nil)
+	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, nil, identityServer, nil, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}
@@ -49,7 +49,7 @@ func TestGetPluginInfo(t *testing.T) {
 func TestGetPluginCapabilities(t *testing.T) {
 	gceDriver := GetGCEDriver()
 	identityServer := NewIdentityServer(gceDriver)
-	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, identityServer, nil, nil)
+	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, identityServer, nil, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}
@@ -82,7 +82,7 @@ func TestGetPluginCapabilities(t *testing.T) {
 func TestProbe(t *testing.T) {
 	gceDriver := GetGCEDriver()
 	identityServer := NewIdentityServer(gceDriver)
-	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, identityServer, nil, nil)
+	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, identityServer, nil, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)
 	}