Up to date changes

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/common"
@@ -239,19 +241,27 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to get safe mounter: %v", err.Error())
 		}
+
 		deviceUtils := deviceutils.NewDeviceUtils()
 		statter := mountmanager.NewStatter(mounter)
 		meta, err := metadataservice.NewMetadataService()
 		if err != nil {
 			klog.Fatalf("Failed to set up metadata service: %v", err.Error())
 		}
+
+		kubeClient, err := instantiateKubeClient()
+		if err != nil {
+			klog.Fatalf("Failed to instantiate Kubernetes client: %v", err)
+		}
 		nsArgs := driver.NodeServerArgs{
 			EnableDeviceInUseCheck:   *enableDeviceInUseCheck,
 			DeviceInUseTimeout:       *deviceInUseTimeout,
 			EnableDataCache:          *enableDataCacheFlag,
 			DataCacheEnabledNodePool: isDataCacheEnabledNodePool(ctx, *nodeName),
+			KubeClient:               kubeClient,
 		}
 		nodeServer = driver.NewNodeServer(gceDriver, mounter, deviceUtils, meta, statter, nsArgs)
+
 		if *maxConcurrentFormatAndMount > 0 {
 			nodeServer = nodeServer.WithSerializedFormatAndMount(*formatAndMountTimeout, *maxConcurrentFormatAndMount)
 		}
@@ -288,6 +298,19 @@ func handle() {
 	gceDriver.Run(*endpoint, *grpcLogCharCap, *enableOtelTracing, metricsManager)
 }
 
+func instantiateKubeClient() (*kubernetes.Clientset, error) {
+	klog.V(2).Infof("Setting up kubeClient")
+	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
+}
+
 func notEmpty(v string) bool {
 	return v != ""
 }

pkg/common/constants.go

@@ -18,7 +18,8 @@ package common
 
 const (
 	// Keys for Topology. This key will be shared amongst drivers from GCP
-	TopologyKeyZone = "topology.gke.io/zone"
+	TopologyKeyPrefix = "topology.gke.io"
+	TopologyKeyZone   = TopologyKeyPrefix + "/zone"
 
 	// VolumeAttributes for Partition
 	VolumeAttributePartition = "partition"

pkg/common/utils.go

@@ -749,3 +749,11 @@ func ShortString(s string) string {
 	}
 	return string(short)
 }
+
+func IsGKETopologyLabel(key string) bool {
+	// This is the actual code
+	// return strings.HasPrefix(key, gkeTopologyLabelPrefix)
+
+	// More permissive code for testing
+	return strings.HasPrefix(key, "topology.gke")
+}

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

@@ -157,6 +157,7 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 		deviceInUseErrors:        newDeviceErrMap(args.DeviceInUseTimeout),
 		EnableDataCache:          args.EnableDataCache,
 		DataCacheEnabledNodePool: args.DataCacheEnabledNodePool,
+		KubeClient:               args.KubeClient,
 	}
 }
 

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

@@ -30,9 +30,11 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 
+	"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/deviceutils"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
@@ -49,6 +51,8 @@ type GCENodeServer struct {
 	EnableDataCache          bool
 	DataCacheEnabledNodePool bool
 
+	KubeClient *kubernetes.Clientset
+
 	// A map storing all volumes with ongoing operations so that additional operations
 	// for that same volume (as defined by VolumeID) return an Aborted error
 	volumeLocks *common.VolumeLocks
@@ -84,6 +88,9 @@ type NodeServerArgs struct {
 	EnableDataCache bool
 
 	DataCacheEnabledNodePool bool
+
+	// Kubernetes client for API server interactions
+	KubeClient *kubernetes.Clientset
 }
 
 var _ csi.NodeServer = &GCENodeServer{}
@@ -556,22 +563,58 @@ func (ns *GCENodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeG
 }
 
 func (ns *GCENodeServer) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+	labels, err := ns.gkeTopologyLabels(ctx, ns.MetadataService.GetName())
+	if err != nil {
+		// Perhaps we don't want to fail here.  We are introducing a new
+		// dependency and we might be better off allowing this failure to
+		// happen and moving on to retrieve the zone from GCE MDS.
+		return nil, err
+	}
+
+	// Each "Topology" struct will later be translated into an individual
+	// 'matchExpressions' block in the PV's NodeAffinity.  Because we always
+	// need to match on both the zone AND the disk type, both the zone and the
+	// supported disks belong as segments on a single Topology.
 	top := &csi.Topology{
-		Segments: map[string]string{common.TopologyKeyZone: ns.MetadataService.GetZone()},
+		Segments: labels,
 	}
 
 	nodeID := common.CreateNodeID(ns.MetadataService.GetProject(), ns.MetadataService.GetZone(), ns.MetadataService.GetName())
-
 	volumeLimits, err := ns.GetVolumeLimits()
 
 	resp := &csi.NodeGetInfoResponse{
 		NodeId:             nodeID,
 		MaxVolumesPerNode:  volumeLimits,
 		AccessibleTopology: top,
 	}
+
+	klog.V(2).Infof("Returning NodeGetInfoResponse: %+v", resp)
+
 	return resp, err
 }
 
+// gkeTopologyLabels retrieves the node labels with the prefix
+// `topology.gke.io/`.
+func (ns *GCENodeServer) gkeTopologyLabels(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 {
+		// We should retry instead.  Need to figure out how much wrong-ness can be tolerated and how often CSINode gets refreshed.
+		return nil, err
+	}
+
+	topology := make(map[string]string)
+	for k, v := range node.GetLabels() {
+		if common.IsGKETopologyLabel(k) {
+			klog.V(2).Infof("Including node topology label %q=%q", k, v)
+			topology[k] = v
+		}
+	}
+
+	return topology, nil
+}
+
 func (ns *GCENodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
 	if len(req.VolumeId) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty")