use csiNode informer to prevent hit the API server for all csiNode resource request

Signed-off-by: Mucahit Kurt <[email protected]>

Author: mucahitkurt

Gopkg.lock

@@ -42,6 +42,8 @@ import (
 	"k8s.io/klog"
 
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/informers"
+	storagelisters "k8s.io/client-go/listers/storage/v1beta1"
 	utilflag "k8s.io/component-base/cli/flag"
 	csitrans "k8s.io/csi-translation-lib"
 )
@@ -181,11 +183,32 @@ func main() {
 		provisionerOptions = append(provisionerOptions, controller.AdditionalProvisionerNames([]string{supportsMigrationFromInTreePluginName}))
 	}
 
+	var csiNodeLister storagelisters.CSINodeLister
+	var factory informers.SharedInformerFactory
+	if ctrl.SupportsTopology(pluginCapabilities) {
+		// Create informer to prevent hit the API server for all resource request
+		factory = informers.NewSharedInformerFactory(clientset, ctrl.ResyncPeriodOfCsiNodeInformer)
+		csiNodeLister = factory.Storage().V1beta1().CSINodes().Lister()
+	}
+
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
-	csiProvisioner := ctrl.NewCSIProvisioner(clientset, *operationTimeout, identity, *volumeNamePrefix,
-		*volumeNameUUIDLength, grpcClient, snapClient, provisionerName, pluginCapabilities,
-		controllerCapabilities, supportsMigrationFromInTreePluginName, *strictTopology, translator)
+	csiProvisioner := ctrl.NewCSIProvisioner(
+		clientset,
+		*operationTimeout,
+		identity,
+		*volumeNamePrefix,
+		*volumeNameUUIDLength,
+		grpcClient,
+		snapClient,
+		provisionerName,
+		pluginCapabilities,
+		controllerCapabilities,
+		supportsMigrationFromInTreePluginName,
+		*strictTopology,
+		translator,
+		csiNodeLister)
+
 	provisionController = controller.NewProvisionController(
 		clientset,
 		provisionerName,
@@ -195,6 +218,17 @@ func main() {
 	)
 
 	run := func(context.Context) {
+		if factory != nil {
+			stopCh := context.Background().Done()
+			factory.Start(stopCh)
+			cacheSyncResult := factory.WaitForCacheSync(stopCh)
+			for _, v := range cacheSyncResult {
+				if !v {
+					klog.Fatalf("Failed to sync CsiNodeInformer!")
+				}
+			}
+		}
+
 		provisionController.Run(wait.NeverStop)
 	}
 

cmd/csi-provisioner/csi-provisioner.go

@@ -30,25 +30,24 @@ import (
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/kubernetes-csi/csi-lib-utils/connection"
-	"github.com/kubernetes-csi/external-provisioner/pkg/features"
 	snapapi "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1"
 	snapclientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/util"
 
-	v1 "k8s.io/api/core/v1"
+	"k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	_ "k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog"
 
 	"google.golang.org/grpc"
+	storagelisters "k8s.io/client-go/listers/storage/v1beta1"
 )
 
 //secretParamsMap provides a mapping of current as well as deprecated secret keys
@@ -115,6 +114,8 @@ const (
 	tokenPVCNameKey      = "pvc.name"
 	tokenPVCNameSpaceKey = "pvc.namespace"
 
+	ResyncPeriodOfCsiNodeInformer = 1 * time.Hour
+
 	deleteVolumeRetryCount = 5
 
 	annStorageProvisioner = "volume.beta.kubernetes.io/storage-provisioner"
@@ -195,6 +196,7 @@ type csiProvisioner struct {
 	supportsMigrationFromInTreePluginName string
 	strictTopology                        bool
 	translator                            ProvisionerCSITranslator
+	csiNodeLister                         storagelisters.CSINodeLister
 }
 
 var _ controller.Provisioner = &csiProvisioner{}
@@ -254,7 +256,8 @@ func NewCSIProvisioner(client kubernetes.Interface,
 	controllerCapabilities connection.ControllerCapabilitySet,
 	supportsMigrationFromInTreePluginName string,
 	strictTopology bool,
-	translator ProvisionerCSITranslator) controller.Provisioner {
+	translator ProvisionerCSITranslator,
+	csiNodeLister storagelisters.CSINodeLister) controller.Provisioner {
 
 	csiClient := csi.NewControllerClient(grpcClient)
 	provisioner := &csiProvisioner{
@@ -272,6 +275,7 @@ func NewCSIProvisioner(client kubernetes.Interface,
 		supportsMigrationFromInTreePluginName: supportsMigrationFromInTreePluginName,
 		strictTopology:                        strictTopology,
 		translator:                            translator,
+		csiNodeLister:                         csiNodeLister,
 	}
 	return provisioner
 }
@@ -499,7 +503,8 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 			options.PVC.Name,
 			options.StorageClass.AllowedTopologies,
 			options.SelectedNode,
-			p.strictTopology)
+			p.strictTopology,
+			p.csiNodeLister)
 		if err != nil {
 			return nil, controller.ProvisioningNoChange, fmt.Errorf("error generating accessibility requirements: %v", err)
 		}
@@ -664,8 +669,7 @@ func (p *csiProvisioner) ProvisionExt(options controller.ProvisionOptions) (*v1.
 }
 
 func (p *csiProvisioner) supportsTopology() bool {
-	return p.pluginCapabilities[csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS] &&
-		utilfeature.DefaultFeatureGate.Enabled(features.Topology)
+	return SupportsTopology(p.pluginCapabilities)
 }
 
 func removePrefixedParameters(param map[string]string) (map[string]string, error) {

pkg/controller/controller.go

@@ -39,7 +39,7 @@ import (
 	crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1"
 	"github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned/fake"
 	"google.golang.org/grpc"
-	v1 "k8s.io/api/core/v1"
+	"k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	storagev1beta1 "k8s.io/api/storage/v1beta1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -401,7 +401,7 @@ func TestCreateDriverReturnsInvalidCapacityDuringProvision(t *testing.T) {
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(nil, 5*time.Second, "test-provisioner", "test",
-		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+		5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil)
 
 	// Requested PVC with requestedBytes storage
 	deletePolicy := v1.PersistentVolumeReclaimDelete
@@ -1464,7 +1464,7 @@ func runProvisionTest(t *testing.T, k string, tc provisioningTestcase, requested
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New())
+		nil, provisionDriverName, pluginCaps, controllerCaps, supportsMigrationFromInTreePluginName, false, csitrans.New(), nil)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2031,7 +2031,7 @@ func TestProvisionFromSnapshot(t *testing.T) {
 
 		pluginCaps, controllerCaps := provisionFromSnapshotCapabilities()
 		csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+			client, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil)
 
 		out := &csi.CreateVolumeResponse{
 			Volume: &csi.Volume{
@@ -2186,7 +2186,7 @@ func TestProvisionWithTopologyEnabled(t *testing.T) {
 			}
 
 			nodes := buildNodes(tc.nodeLabels, k8sTopologyBetaVersion.String())
-			nodeInfos := buildNodeInfos(tc.topologyKeys)
+			csiNodes := buildCSINodes(tc.topologyKeys)
 
 			var (
 				pluginCaps     connection.PluginCapabilitySet
@@ -2199,9 +2199,13 @@ func TestProvisionWithTopologyEnabled(t *testing.T) {
 				pluginCaps, controllerCaps = provisionCapabilities()
 			}
 
-			clientSet := fakeclientset.NewSimpleClientset(nodes, nodeInfos)
+			clientSet := fakeclientset.NewSimpleClientset(nodes, csiNodes)
+
+			csiNodeLister, stopChan := csiNodeLister(clientSet, t)
+			defer close(stopChan)
+
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+				csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), csiNodeLister)
 
 			pv, err := csiProvisioner.Provision(controller.ProvisionOptions{
 				StorageClass: &storagev1.StorageClass{},
@@ -2256,7 +2260,7 @@ func TestProvisionWithTopologyDisabled(t *testing.T) {
 	clientSet := fakeclientset.NewSimpleClientset()
 	pluginCaps, controllerCaps := provisionWithTopologyCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil)
 
 	out := &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
@@ -2436,7 +2440,7 @@ func runDeleteTest(t *testing.T, k string, tc deleteTestcase) {
 
 	pluginCaps, controllerCaps := provisionCapabilities()
 	csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5,
-		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+		csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil)
 
 	err = csiProvisioner.Delete(tc.persistentVolume)
 	if tc.expectErr && err == nil {
@@ -3092,7 +3096,7 @@ func TestProvisionFromPVC(t *testing.T) {
 		}
 
 		csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner", "test", 5, csiConn.conn,
-			nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New())
+			nil, driverName, pluginCaps, controllerCaps, "", false, csitrans.New(), nil)
 
 		pv, err := csiProvisioner.Provision(tc.volOpts)
 		if tc.expectErr && err == nil {
@@ -3171,7 +3175,7 @@ func TestProvisionWithMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps,
-				inTreePluginName, false, mockTranslator)
+				inTreePluginName, false, mockTranslator, nil)
 
 			// Set up return values (AnyTimes to avoid overfitting on implementation)
 
@@ -3331,7 +3335,7 @@ func TestDeleteMigration(t *testing.T) {
 			pluginCaps, controllerCaps := provisionCapabilities()
 			csiProvisioner := NewCSIProvisioner(clientSet, 5*time.Second, "test-provisioner",
 				"test", 5, csiConn.conn, nil, driverName, pluginCaps, controllerCaps, "",
-				false, mockTranslator)
+				false, mockTranslator, nil)
 
 			// Set mock return values (AnyTimes to avoid overfitting on implementation details)
 			mockTranslator.EXPECT().IsPVMigratable(gomock.Any()).Return(tc.expectTranslation).AnyTimes()