Only add disk support Topology if all nodes have a disk support label

Author: julianKatz

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

@@ -38,6 +38,7 @@ import (
 	driver "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-pd-csi-driver"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/metrics"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/nodelabels"
 )
 
 var (
@@ -96,7 +97,7 @@ var (
 
 	extraTagsStr = flag.String("extra-tags", "", "Extra tags to attach to each Compute Disk, Image, Snapshot created. It is a comma separated list of parent id, key and value like '<parent_id1>/<tag_key1>/<tag_value1>,...,<parent_idN>/<tag_keyN>/<tag_valueN>'. parent_id is the Organization or the Project ID or Project name where the tag key and the tag value resources exist. A maximum of 50 tags bindings is allowed for a resource. See https://cloud.google.com/resource-manager/docs/tags/tags-overview, https://cloud.google.com/resource-manager/docs/tags/tags-creating-and-managing for details")
 
-	diskTopology = flag.Bool("disk-topology", false, "If set to true, the driver will add a topology.gke.io/[disk-type] topology label to the Topologies returned in CreateVolumeResponse")
+	diskTopology = flag.Bool("disk-topology", false, "If set to true, the driver will add a topology.gke.io/[disk-type] topology label to the Topologies returned in CreateVolumeResponse.  Label is only added if all cluster nodes have at least one disk support label")
 
 	version string
 )
@@ -229,11 +230,27 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %v", err.Error())
 		}
+
 		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
 		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Millisecond
 		args := &driver.GCEControllerServerArgs{
 			EnableDiskTopology: *diskTopology,
 		}
+
+		if *diskTopology {
+			klog.V(2).Infof("Setting up kubeClient")
+			kubeClient, err := instantiateKubeClient()
+			if err != nil {
+				klog.Fatalf("Failed to instantiate Kubernetes client: %v", err)
+			}
+			klog.V(2).Infof("Setting up node lister with informer")
+			labelVerifier, err := nodelabels.NewVerifier(ctx, kubeClient)
+			if err != nil {
+				klog.Fatalf("Failed to create node label verifier: %v", err)
+			}
+			args.LabelVerifier = labelVerifier
+		}
+
 		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration, fallbackRequisiteZones, *enableStoragePoolsFlag, *enableDataCacheFlag, multiZoneVolumeHandleConfig, listVolumesConfig, provisionableDisksConfig, *enableHdHAFlag, args)
 	} else if *cloudConfigFilePath != "" {
 		klog.Warningf("controller service is disabled but cloud config given - it has no effect")

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

@@ -40,6 +40,7 @@ import (
 	"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"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/nodelabels"
 )
 
 type GCEControllerServer struct {
@@ -122,10 +123,12 @@ type GCEControllerServer struct {
 	csi.UnimplementedControllerServer
 
 	EnableDiskTopology bool
+	LabelVerifier      *nodelabels.Verifier
 }
 
 type GCEControllerServerArgs struct {
 	EnableDiskTopology bool
+	LabelVerifier      *nodelabels.Verifier
 }
 
 type MultiZoneVolumeHandleConfig struct {
@@ -471,9 +474,7 @@ func (gceCS *GCEControllerServer) getMultiZoneProvisioningZones(ctx context.Cont
 }
 
 func (gceCS *GCEControllerServer) createMultiZoneDisk(ctx context.Context, req *csi.CreateVolumeRequest, params common.DiskParameters, dataCacheParams common.DataCacheParameters, enableDataCache bool) (*csi.CreateVolumeResponse, error) {
-	// Determine the zones that are needed.
 	var err error
-
 	// For multi-zone, we either select:
 	// 1) The zones specified in requisite topology requirements
 	// 2) All zones in the region that are compatible with the selected disk type
@@ -523,7 +524,12 @@ func (gceCS *GCEControllerServer) createMultiZoneDisk(ctx context.Context, req *
 	// Use the first response as a template
 	volumeId := fmt.Sprintf("projects/%s/zones/%s/disks/%s", gceCS.CloudProvider.GetDefaultProject(), common.MultiZoneValue, req.GetName())
 	klog.V(4).Infof("CreateVolume succeeded for multi-zone disks in zones %s: %v", zones, multiZoneVolKey)
-	return gceCS.generateCreateVolumeResponseWithVolumeId(createdDisks[0], zones, params, dataCacheParams, enableDataCache, volumeId), nil
+
+	resp, err := gceCS.generateCreateVolumeResponseWithVolumeId(createdDisks[0], zones, params, dataCacheParams, enableDataCache, volumeId)
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate createVolumeResponse: %w", err)
+	}
+	return resp, nil
 }
 
 func (gceCS *GCEControllerServer) getZonesWithDiskNameAndType(ctx context.Context, name string, diskType string) ([]string, error) {
@@ -623,13 +629,22 @@ func (gceCS *GCEControllerServer) createSingleDeviceDisk(ctx context.Context, re
 		return nil, status.Errorf(codes.Aborted, common.VolumeOperationAlreadyExistsFmt, volumeID)
 	}
 	defer gceCS.volumeLocks.Release(volumeID)
+<<<<<<< HEAD
 	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones, accessMode)
+=======
+>>>>>>> 474af8f4 (Only add disk support Topology if all nodes have a disk support label)
 
+	disk, err := gceCS.createSingleDisk(ctx, req, params, volKey, zones)
 	if err != nil {
 		return nil, common.LoggedError("CreateVolume failed: %v", err)
 	}
 
-	return gceCS.generateCreateVolumeResponseWithVolumeId(disk, zones, params, dataCacheParams, enableDataCache, volumeID), err
+	resp, err := gceCS.generateCreateVolumeResponseWithVolumeId(disk, zones, params, dataCacheParams, enableDataCache, volumeID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate createVolumeResponse: %w", err)
+	}
+
+	return resp, nil
 }
 
 func getAccessMode(req *csi.CreateVolumeRequest, params common.DiskParameters) (string, error) {
@@ -2412,7 +2427,7 @@ func extractVolumeContext(context map[string]string) (*PDCSIContext, error) {
 	return info, nil
 }
 
-func (gceCS *GCEControllerServer) generateCreateVolumeResponseWithVolumeId(disk *gce.CloudDisk, zones []string, params common.DiskParameters, dataCacheParams common.DataCacheParameters, enableDataCache bool, volumeId string) *csi.CreateVolumeResponse {
+func (gceCS *GCEControllerServer) generateCreateVolumeResponseWithVolumeId(disk *gce.CloudDisk, zones []string, params common.DiskParameters, dataCacheParams common.DataCacheParameters, enableDataCache bool, volumeId string) (*csi.CreateVolumeResponse, error) {
 	tops := []*csi.Topology{}
 	for _, zone := range zones {
 		top := &csi.Topology{
@@ -2422,7 +2437,14 @@ func (gceCS *GCEControllerServer) generateCreateVolumeResponseWithVolumeId(disk
 		}
 
 		if gceCS.EnableDiskTopology {
-			top.Segments[common.TopologyLabelKey(params.DiskType)] = "true"
+			klog.V(4).Infof("Verifying disk support labels on cluster nodes before adding disk support topology")
+			addDiskSupportTopology, err := gceCS.allNodesHaveDiskSupportLabel()
+			if err != nil {
+				return nil, fmt.Errorf("failed to check if all nodes have disk support label: %w", err)
+			}
+			if addDiskSupportTopology {
+				top.Segments[common.TopologyLabelKey(params.DiskType)] = "true"
+			}
 		}
 
 		tops = append(tops, top)
@@ -2479,7 +2501,15 @@ func (gceCS *GCEControllerServer) generateCreateVolumeResponseWithVolumeId(disk
 		}
 		createResp.Volume.ContentSource = contentSource
 	}
-	return createResp
+	return createResp, nil
+}
+
+func (gceCS *GCEControllerServer) allNodesHaveDiskSupportLabel() (bool, error) {
+	allNodesHaveDiskSupportLabel, err := gceCS.LabelVerifier.AllNodesHaveDiskSupportLabel()
+	if err != nil {
+		return false, fmt.Errorf("failed to check if all nodes have disk support label: %w", err)
+	}
+	return allNodesHaveDiskSupportLabel, nil
 }
 
 func getResourceId(resourceLink string) (string, error) {

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

@@ -37,6 +37,8 @@ import (
 	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/klog/v2"
@@ -47,6 +49,7 @@ import (
 	"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"
 	gcecloudprovider "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/nodelabels"
 )
 
 const (
@@ -617,7 +620,7 @@ func TestCreateVolumeArguments(t *testing.T) {
 		enableStoragePools bool
 		expVol             *csi.Volume
 		expErrCode         codes.Code
-		EnableDiskTopology bool
+		enableDiskTopology bool
 	}{
 		{
 			name: "success default",
@@ -1371,7 +1374,7 @@ func TestCreateVolumeArguments(t *testing.T) {
 					},
 				},
 			},
-			EnableDiskTopology: true,
+			enableDiskTopology: true,
 		},
 		{
 			// Desired as the disk type label should match the `type` parameter,
@@ -1414,19 +1417,31 @@ func TestCreateVolumeArguments(t *testing.T) {
 					},
 				},
 			},
-			EnableDiskTopology: true,
+			enableDiskTopology: true,
 		},
 	}
 
 	// Run test cases
 	for _, tc := range testCases {
 		t.Logf("test case: %s", tc.name)
+
 		// Setup new driver each time so no interference
 		args := &GCEControllerServerArgs{
-			EnableDiskTopology: tc.EnableDiskTopology,
+			EnableDiskTopology: tc.enableDiskTopology,
+			LabelVerifier: nodelabels.NewFakeVerifier(t, []*corev1.Node{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "node-with-disk-support-label",
+						Labels: map[string]string{
+							common.TopologyLabelKey(stdDiskType): "true",
+						},
+					},
+				},
+			}),
 		}
 		gceDriver := initGCEDriver(t, nil, args)
 		gceDriver.cs.enableStoragePools = tc.enableStoragePools
+
 		// Start Test
 		resp, err := gceDriver.cs.CreateVolume(context.Background(), tc.req)
 		if err != nil {

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

@@ -177,6 +177,7 @@ func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, err
 		provisionableDisksConfig:    provisionableDisksConfig,
 		enableHdHA:                  enableHdHA,
 		EnableDiskTopology:          args.EnableDiskTopology,
+		LabelVerifier:               args.LabelVerifier,
 	}
 }
 

pkg/nodelabels/fake.go

@@ -0,0 +1,36 @@
+package nodelabels
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func NewFakeVerifier(t *testing.T, nodes []*corev1.Node) *Verifier {
+	ctx := context.Background()
+	clientset := fakeKubeClient(nodes)
+
+	verifier, err := NewVerifier(ctx, clientset)
+	if err != nil {
+		t.Fatalf("failed to create verifier: %v", err)
+	}
+
+	return verifier
+}
+
+func fakeKubeClient(nodes []*corev1.Node) kubernetes.Interface {
+	// Convert the list of nodes to a slice of runtime.Object
+	var objects []runtime.Object
+	for _, node := range nodes {
+		objects = append(objects, node)
+	}
+
+	// Create a fake clientset with the predefined objects
+	clientset := fake.NewSimpleClientset(objects...)
+
+	return clientset
+}

pkg/nodelabels/verifier.go

@@ -0,0 +1,89 @@
+package nodelabels
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	listers "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+)
+
+// resyncInterval is the interval at which the informer resyncs.  Should the
+// informer cache miss an event, the resync will correct the cache.
+const resyncInterval = 10 * time.Minute
+
+// Verifier checks node labels for GKE topology labels.
+type Verifier struct {
+	nodeLister listers.NodeLister
+}
+
+// NewVerifier creates an informer for listing nodes and returns a
+// lister.  It waits for the informer to sync and connects the factory's stop
+// channel to the context's done channel to prevent goroutine leaks when the
+// binary is stopped.
+func NewVerifier(ctx context.Context, clientset kubernetes.Interface) (*Verifier, error) {
+	stopCh := make(chan struct{})
+	// Should the context be cancelled upstream, we want to stop the informer.
+	// Documentation suggests that this prevents goroutine leaks.
+	go func() {
+		<-ctx.Done()
+		close(stopCh)
+	}()
+
+	factory := informers.NewSharedInformerFactory(clientset, resyncInterval)
+	nodeInformer := factory.Core().V1().Nodes().Informer()
+	nodeLister := factory.Core().V1().Nodes().Lister()
+
+	factory.Start(stopCh)
+	synced := cache.WaitForCacheSync(stopCh, nodeInformer.HasSynced)
+	if !synced {
+		return nil, fmt.Errorf("failed to sync node informer")
+	}
+
+	return &Verifier{
+		nodeLister: nodeLister,
+	}, nil
+}
+
+func (v *Verifier) AllNodesHaveDiskSupportLabel() (bool, error) {
+	nodes, err := v.nodeLister.List(labels.Everything())
+	if err != nil {
+		return false, fmt.Errorf("failed to list nodes: %v", err)
+	}
+	return allNodesHaveDiskSupportLabel(nodes), nil
+}
+
+func allNodesHaveDiskSupportLabel(nodes []*corev1.Node) bool {
+	if len(nodes) == 0 {
+		return false
+	}
+
+	for _, node := range nodes {
+		if !nodeHasDiskSupportLabel(node) {
+			klog.V(4).Infof("Node %s does not have disk support label", node.GetName())
+			return false
+		}
+	}
+	return true
+}
+
+func nodeHasDiskSupportLabel(node *corev1.Node) bool {
+	labels := node.GetLabels()
+	for key, _ := range labels {
+		if isNonZoneGKETopologyLabel(key) {
+			return true
+		}
+	}
+	return false
+}
+
+func isNonZoneGKETopologyLabel(key string) bool {
+	return common.IsGKETopologyLabel(key) && key != common.TopologyKeyZone
+}