Hooked up NodeGetInfo to provide disk-type labels from nodes

Author: julianKatz

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"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/constants"
@@ -99,6 +101,8 @@ var (
 
 	diskTopology = flag.Bool("disk-topology", false, "If set to true, the driver will add a disk-type.gke.io/[disk-type] topology label when the StorageClass has the use-allowed-disk-topology parameter set to true. That topology label is included in the Topologies returned in CreateVolumeResponse. This flag is disabled by default.")
 
+	dynamicVolumes = flag.Bool("dynamic-volumes", false, "If set to true, the CSI driver will automatically select a compatible disk type based on the presence of the dynamic-volume parameter and disk types defined in the StorageClass. Disabled by default.")
+
 	diskCacheSyncPeriod = flag.Duration("disk-cache-sync-period", 10*time.Minute, "Period for the disk cache to check the /dev/disk/by-id/ directory and evaluate the symlinks")
 
 	enableDiskSizeValidation = flag.Bool("enable-disk-size-validation", false, "If set to true, the driver will validate that the requested disk size is matches the physical disk size. This flag is disabled by default.")
@@ -256,6 +260,7 @@ func handle() {
 		args := &driver.GCEControllerServerArgs{
 			EnableDiskTopology:       *diskTopology,
 			EnableDiskSizeValidation: *enableDiskSizeValidation,
+			EnableDynamicVolumes:     *dynamicVolumes,
 		}
 
 		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig, *enableHdHAFlag, args)
@@ -299,6 +304,16 @@ func handle() {
 			MetricsManager:           metricsManager,
 			DeviceCache:              deviceCache,
 		}
+
+		if *dynamicVolumes {
+			klog.V(2).Infof("Setting up kubeClient")
+			kc, err := instantiateKubeClient()
+			if err != nil {
+				klog.Fatalf("Failed to instantiate Kubernetes client: %v", err)
+			}
+			nsArgs.KubeClient = kc
+			nsArgs.EnableDynamicVolumes = *dynamicVolumes
+		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
 
 		if *maxConcurrentFormatAndMount > 0 {
@@ -467,3 +482,15 @@ func setupDataCache(ctx context.Context, nodeName string, nodeId string) error {
 	klog.V(4).Infof("LSSD caching is setup for the Data Cache enabled node %s", nodeName)
 	return nil
 }
+
+func instantiateKubeClient() (*kubernetes.Clientset, error) {
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create REST Config for k8s client: %w", err)
+	}
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create k8s client: %w", err)
+	}
+	return kubeClient, nil
+}

pkg/common/utils.go

@@ -491,6 +491,11 @@ func MapNumber(vCPUs int64, limitMap []constants.MachineHyperdiskLimit) int64 {
 	return 15
 }
 
+// HasDiskTypeLabelKeyPrefix checks if the label key starts with the DiskTypeKeyPrefix.
+func HasDiskTypeLabelKeyPrefix(labelKey string) bool {
+	return strings.HasPrefix(labelKey, constants.DiskTypeKeyPrefix+"/")
+}
+
 func DiskTypeLabelKey(diskType string) string {
 	return fmt.Sprintf("%s/%s", constants.DiskTypeKeyPrefix, diskType)
 }

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

@@ -132,6 +132,7 @@ type GCEControllerServer struct {
 type GCEControllerServerArgs struct {
 	EnableDiskTopology       bool
 	EnableDiskSizeValidation bool
+	EnableDynamicVolumes     bool
 }
 
 type MultiZoneVolumeHandleConfig struct {

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

@@ -35,6 +35,8 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/constants"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
@@ -83,6 +85,9 @@ type GCENodeServer struct {
 	metricsManager *metrics.MetricsManager
 	// A cache of the device paths for the volumes that are attached to the node.
 	DeviceCache *linkcache.DeviceCache
+
+	KubeClient           kubernetes.Interface
+	EnableDynamicVolumes bool
 }
 
 type NodeServerArgs struct {
@@ -101,6 +106,10 @@ type NodeServerArgs struct {
 
 	MetricsManager *metrics.MetricsManager
 	DeviceCache    *linkcache.DeviceCache
+
+	KubeClient           kubernetes.Interface
+	EnableDiskTopology   bool
+	EnableDynamicVolumes bool
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -717,6 +726,17 @@ func (ns *GCENodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRe
 		Segments: map[string]string{constants.TopologyKeyZone: ns.MetadataService.GetZone()},
 	}
 
+	if ns.EnableDynamicVolumes {
+		labels, err := ns.fetchGKETopologyLabels(ctx, ns.MetadataService.GetName())
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch GKE topology labels: %v", err)
+		}
+
+		for k, v := range labels {
+			top.Segments[k] = v
+		}
+	}
+
 	nodeID := common.CreateNodeID(ns.MetadataService.GetProject(), ns.MetadataService.GetZone(), ns.MetadataService.GetName())
 
 	volumeLimits, err := ns.GetVolumeLimits(ctx)
@@ -970,3 +990,25 @@ func GetAttachLimitsOverrideFromNodeLabel(ctx context.Context, nodeName string)
 	}
 	return 0, nil
 }
+
+// fetchGKETopologyLabels retrieves the node labels with the prefix
+// `topology.gke.io/` for the specified node.
+func (ns *GCENodeServer) fetchGKETopologyLabels(ctx context.Context, nodeName string) (map[string]string, error) {
+	klog.V(2).Infof("Retrieving node topology labels for node %q", nodeName)
+
+	node, err := ns.KubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	if err != nil {
+		// Q: Should we retry if we fail to get the node?
+		return nil, err
+	}
+
+	topology := make(map[string]string)
+	for k, v := range node.GetLabels() {
+		if common.HasDiskTypeLabelKeyPrefix(k) {
+			klog.V(2).Infof("Including node topology label %q=%q", k, v)
+			topology[k] = v
+		}
+	}
+
+	return topology, nil
+}