Merge pull request #3797 from landreasyan/cherry-pick-3668-release-1.33

[release-1.33] fix: use node informer to get node labels (cherry-pick of #3668)

Author: landreasyan

pkg/azuredisk/azure_controller_common.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kwait "k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
 	cloudprovider "k8s.io/cloud-provider"
 	volerr "k8s.io/cloud-provider/volume/errors"
 	"k8s.io/klog/v2"
@@ -88,6 +89,7 @@ type controllerCommon struct {
 	lockMap       *lockMap
 	cloud         *provider.Cloud
 	clientFactory azclient.ClientFactory
+	nodeLister    cache.GenericLister
 	// disk queue that is waiting for attach or detach on specific node
 	// <nodeName, map<diskURI, *provider.AttachDiskOptions/DetachDiskOptions>>
 	attachDiskMap sync.Map
@@ -224,7 +226,13 @@ func (c *controllerCommon) AttachDisk(ctx context.Context, diskName, diskURI str
 
 	numDisksAllowed := math.MaxInt
 	if c.CheckDiskCountForBatching {
-		_, instanceType, err := getNodeInfoFromLabels(ctx, string(nodeName), c.cloud.KubeClient)
+		var instanceType string
+		var err error
+		if c.nodeLister != nil {
+			_, instanceType, err = GetNodeInfoFromNodeLister(string(nodeName), c.nodeLister)
+		} else {
+			_, instanceType, err = GetNodeInfoFromLabels(ctx, string(nodeName), c.cloud.KubeClient)
+		}
 		if err != nil {
 			klog.Errorf("failed to get node info from labels: %v", err)
 		} else if instanceType != "" {

pkg/azuredisk/azuredisk.go

@@ -36,6 +36,7 @@ import (
 
 	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	k8stypes "k8s.io/apimachinery/pkg/types"
@@ -50,6 +51,11 @@ import (
 	"k8s.io/mount-utils"
 	"k8s.io/utils/ptr"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/metadata"
+	"k8s.io/client-go/metadata/metadatainformer"
+	"k8s.io/client-go/tools/cache"
+
 	consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
 	"sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
 	csicommon "sigs.k8s.io/azuredisk-csi-driver/pkg/csi-common"
@@ -156,6 +162,9 @@ type Driver struct {
 	enableMigrationMonitor      bool
 	// whether to convert ReadWrite cachingMode to ReadOnly for intree PVs to avoid issues
 	convertRWCachingModeForIntreePV bool
+	nodeLister                      cache.GenericLister
+	nodeInformerSynced              cache.InformerSynced
+	nodeInformerFactory             metadatainformer.SharedInformerFactory
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
@@ -240,10 +249,17 @@ func NewDriver(options *DriverOptions) *Driver {
 	userAgent := GetUserAgent(driver.Name, driver.customUserAgent, driver.userAgentSuffix)
 	klog.V(2).Infof("driver userAgent: %s", userAgent)
 
-	kubeClient, err := azureutils.GetKubeClient(options.Kubeconfig, options.KubeAPIQPS, options.KubeAPIBurst)
+	kubeConfig, err := azureutils.GetKubeConfig(options.Kubeconfig, options.KubeAPIQPS, options.KubeAPIBurst)
 	if err != nil {
 		klog.Warningf("get kubeconfig(%s) failed with error: %v", options.Kubeconfig, err)
 	}
+	var kubeClient clientset.Interface
+	if kubeConfig != nil {
+		kubeClient, err = clientset.NewForConfig(kubeConfig)
+		if err != nil {
+			klog.Warningf("get kubeclient failed with error: %v", err)
+		}
+	}
 	driver.kubeClient = kubeClient
 
 	cloud, err := azureutils.GetCloudProviderFromClient(context.Background(), kubeClient, driver.cloudConfigSecretName, driver.cloudConfigSecretNamespace,
@@ -325,6 +341,23 @@ func NewDriver(options *DriverOptions) *Driver {
 		}
 	}
 
+	if kubeConfig != nil && driver.checkDiskCountForBatching && driver.NodeID == "" {
+		// Create a metadata-only node informer to cache node labels locally (controller only)
+		metadataClient, err := metadata.NewForConfig(kubeConfig)
+		if err != nil {
+			klog.Warningf("failed to create metadata client: %v, node informer will not be used", err)
+		} else {
+			driver.nodeInformerFactory = metadatainformer.NewSharedInformerFactory(metadataClient, 0)
+			nodeGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "nodes"}
+			nodeInformer := driver.nodeInformerFactory.ForResource(nodeGVR)
+			driver.nodeLister = nodeInformer.Lister()
+			driver.nodeInformerSynced = nodeInformer.Informer().HasSynced
+			if driver.diskController != nil {
+				driver.diskController.nodeLister = driver.nodeLister
+			}
+		}
+	}
+
 	driver.deviceHelper = optimization.NewSafeDeviceHelper()
 
 	if driver.getPerfOptimizationEnabled() {
@@ -413,6 +446,19 @@ func (d *Driver) Run(ctx context.Context) error {
 	csi.RegisterControllerServer(s, d)
 	csi.RegisterNodeServer(s, d)
 
+	// Start the node informer if it was set up during driver initialization
+	if d.nodeInformerFactory != nil {
+		d.nodeInformerFactory.Start(ctx.Done())
+		syncCtx, syncCancel := context.WithTimeout(ctx, 30*time.Second)
+		defer syncCancel()
+		if !cache.WaitForCacheSync(syncCtx.Done(), d.nodeInformerSynced) {
+			klog.Warningf("metadata node informer cache has not synced yet, will continue to sync in background")
+		} else {
+			klog.V(2).Infof("metadata node informer cache synced successfully")
+		}
+		klog.V(2).Infof("started metadata node informer for GetNodeInfoFromLabels caching")
+	}
+
 	go func() {
 		//graceful shutdown
 		<-ctx.Done()
@@ -422,6 +468,11 @@ func (d *Driver) Run(ctx context.Context) error {
 			d.migrationMonitor.Stop()
 		}
 
+		// Shutdown node informer if it was started
+		if d.nodeInformerFactory != nil {
+			d.nodeInformerFactory.Shutdown()
+		}
+
 		s.GracefulStop()
 	}()
 	// Driver d act as IdentityServer, ControllerServer and NodeServer
@@ -665,8 +716,38 @@ func (d *Driver) getUsedLunsFromNode(ctx context.Context, nodeName types.NodeNam
 	return usedLuns, nil
 }
 
-// getNodeInfoFromLabels get zone, instanceType from node labels
-func getNodeInfoFromLabels(ctx context.Context, nodeName string, kubeClient clientset.Interface) (string, string, error) {
+// GetNodeInfoFromNodeLister gets zone, instanceType from node labels using the cached nodeLister.
+func GetNodeInfoFromNodeLister(nodeName string, nodeLister cache.GenericLister) (string, string, error) {
+	if nodeLister == nil {
+		return "", "", fmt.Errorf("nodeLister is nil")
+	}
+
+	obj, err := nodeLister.Get(nodeName)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			klog.V(4).Infof("GetNodeInfoFromNodeLister: node(%s) not found in lister cache", nodeName)
+			return "", "", nil
+		}
+		return "", "", fmt.Errorf("get node(%s) from lister failed: %v", nodeName, err)
+	}
+
+	pom, ok := obj.(*metav1.PartialObjectMetadata)
+	if !ok {
+		return "", "", fmt.Errorf("node(%s) from lister is not *metav1.PartialObjectMetadata", nodeName)
+	}
+
+	if len(pom.Labels) == 0 {
+		return "", "", fmt.Errorf("node(%s) label is empty", nodeName)
+	}
+
+	zone := pom.Labels[consts.WellKnownTopologyKey]
+	instanceType := pom.Labels[consts.InstanceTypeKey]
+	klog.V(4).Infof("GetNodeInfoFromNodeLister: node(%s): zone=%s, instanceType=%s", nodeName, zone, instanceType)
+	return zone, instanceType, nil
+}
+
+// GetNodeInfoFromLabels gets zone, instanceType from node labels via the kubeClient API server.
+func GetNodeInfoFromLabels(ctx context.Context, nodeName string, kubeClient clientset.Interface) (string, string, error) {
 	if kubeClient == nil || kubeClient.CoreV1() == nil {
 		return "", "", fmt.Errorf("kubeClient is nil")
 	}
@@ -679,7 +760,11 @@ func getNodeInfoFromLabels(ctx context.Context, nodeName string, kubeClient clie
 	if len(node.Labels) == 0 {
 		return "", "", fmt.Errorf("node(%s) label is empty", nodeName)
 	}
-	return node.Labels[consts.WellKnownTopologyKey], node.Labels[consts.InstanceTypeKey], nil
+
+	zone := node.Labels[consts.WellKnownTopologyKey]
+	instanceType := node.Labels[consts.InstanceTypeKey]
+	klog.V(4).Infof("GetNodeInfoFromLabels: node(%s) from API server: zone=%s, instanceType=%s", nodeName, zone, instanceType)
+	return zone, instanceType, nil
 }
 
 // getDefaultDiskIOPSReadWrite according to requestGiB

pkg/azuredisk/azuredisk_test.go

@@ -30,8 +30,15 @@ import (
 	"go.uber.org/mock/gomock"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/status"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/ptr"
 	consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
 	"sigs.k8s.io/cloud-provider-azure/pkg/azclient/diskclient/mock_diskclient"
@@ -290,24 +297,161 @@ func TestDriver_CheckDiskExists_Success(t *testing.T) {
 	assert.Equal(t, err, nil)
 }
 
+// fakeErrorNodeLister is a GenericLister that always returns a specified error.
+type fakeErrorNodeLister struct {
+	err error
+}
+
+func (f *fakeErrorNodeLister) List(selector labels.Selector) ([]runtime.Object, error) {
+	return nil, f.err
+}
+
+func (f *fakeErrorNodeLister) Get(name string) (runtime.Object, error) {
+	return nil, f.err
+}
+
+func (f *fakeErrorNodeLister) ByNamespace(namespace string) cache.GenericNamespaceLister {
+	return nil
+}
+
+func TestGetNodeInfoFromNodeLister(t *testing.T) {
+	nodeGR := schema.GroupResource{Group: "", Resource: "nodes"}
+	tests := []struct {
+		name           string
+		nodeName       string
+		nodeLister     cache.GenericLister
+		expectedZone   string
+		expectedType   string
+		expectedError  error
+		expectErrorNil bool
+	}{
+		{
+			name:          "nil lister",
+			nodeName:      "node1",
+			nodeLister:    nil,
+			expectedError: fmt.Errorf("nodeLister is nil"),
+		},
+		{
+			name:     "lister returns node with labels",
+			nodeName: "node1",
+			nodeLister: func() cache.GenericLister {
+				indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+				node := &metav1.PartialObjectMetadata{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "node1",
+						Labels: map[string]string{
+							consts.WellKnownTopologyKey: "westus2-1",
+							consts.InstanceTypeKey:      "Standard_DS2_v2",
+						},
+					},
+				}
+				_ = indexer.Add(node)
+				return cache.NewGenericLister(indexer, nodeGR)
+			}(),
+			expectedZone:   "westus2-1",
+			expectedType:   "Standard_DS2_v2",
+			expectErrorNil: true,
+		},
+		{
+			name:     "lister returns node with empty labels",
+			nodeName: "node1",
+			nodeLister: func() cache.GenericLister {
+				indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+				node := &metav1.PartialObjectMetadata{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   "node1",
+						Labels: map[string]string{},
+					},
+				}
+				_ = indexer.Add(node)
+				return cache.NewGenericLister(indexer, nodeGR)
+			}(),
+			expectedError: fmt.Errorf("node(node1) label is empty"),
+		},
+		{
+			name:     "lister does not have the node - returns nil (NotFound is not an error)",
+			nodeName: "missing-node",
+			nodeLister: func() cache.GenericLister {
+				indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+				return cache.NewGenericLister(indexer, nodeGR)
+			}(),
+			expectedZone:   "",
+			expectedType:   "",
+			expectErrorNil: true,
+		},
+		{
+			name:          "lister returns non-NotFound error - propagates error",
+			nodeName:      "node1",
+			nodeLister:    &fakeErrorNodeLister{err: fmt.Errorf("connection refused")},
+			expectedError: fmt.Errorf("get node(node1) from lister failed: connection refused"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			zone, instanceType, err := GetNodeInfoFromNodeLister(test.nodeName, test.nodeLister)
+			if test.expectErrorNil {
+				assert.NoError(t, err)
+				assert.Equal(t, test.expectedZone, zone)
+				assert.Equal(t, test.expectedType, instanceType)
+			} else {
+				assert.EqualError(t, err, test.expectedError.Error())
+			}
+		})
+	}
+}
+
 func TestGetNodeInfoFromLabels(t *testing.T) {
 	tests := []struct {
-		nodeName      string
-		kubeClient    clientset.Interface
-		expectedError error
+		name           string
+		nodeName       string
+		kubeClient     clientset.Interface
+		expectedZone   string
+		expectedType   string
+		expectedError  error
+		expectErrorNil bool
 	}{
 		{
+			name:          "nil kubeClient",
 			nodeName:      "",
 			kubeClient:    nil,
 			expectedError: fmt.Errorf("kubeClient is nil"),
 		},
+		{
+			name:     "kubeClient returns node with labels",
+			nodeName: "node1",
+			kubeClient: fake.NewSimpleClientset(&corev1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "node1",
+					Labels: map[string]string{
+						consts.WellKnownTopologyKey: "centralus-1",
+						consts.InstanceTypeKey:      "Standard_E8s_v3",
+					},
+				},
+			}),
+			expectedZone:   "centralus-1",
+			expectedType:   "Standard_E8s_v3",
+			expectErrorNil: true,
+		},
+		{
+			name:          "kubeClient node not found",
+			nodeName:      "missing-node",
+			kubeClient:    fake.NewSimpleClientset(),
+			expectedError: fmt.Errorf("get node(missing-node) failed with nodes \"missing-node\" not found"),
+		},
 	}
 
 	for _, test := range tests {
-		_, _, err := getNodeInfoFromLabels(context.TODO(), test.nodeName, test.kubeClient)
-		if !reflect.DeepEqual(err, test.expectedError) {
-			t.Errorf("Unexpected result: %v, expected result: %v", err, test.expectedError)
-		}
+		t.Run(test.name, func(t *testing.T) {
+			zone, instanceType, err := GetNodeInfoFromLabels(context.TODO(), test.nodeName, test.kubeClient)
+			if test.expectErrorNil {
+				assert.NoError(t, err)
+				assert.Equal(t, test.expectedZone, zone)
+				assert.Equal(t, test.expectedType, instanceType)
+			} else {
+				assert.EqualError(t, err, test.expectedError.Error())
+			}
+		})
 	}
 }