Feature-gate CSINode and CSIDriver informer starts

cmd/kube-controller-manager/app/BUILD

@@ -124,6 +124,7 @@ go_library(
         "//staging/src/k8s.io/client-go/discovery/cached/memory:go_default_library",
         "//staging/src/k8s.io/client-go/dynamic:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/storage/v1beta1:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/metadata:go_default_library",
         "//staging/src/k8s.io/client-go/metadata/metadatainformer:go_default_library",

cmd/kube-controller-manager/app/core.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	cacheddiscovery "k8s.io/client-go/discovery/cached/memory"
+	storagev1beta1informer "k8s.io/client-go/informers/storage/v1beta1"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/metadata"
 	restclient "k8s.io/client-go/rest"
@@ -278,6 +279,17 @@ func startAttachDetachController(ctx ControllerContext) (http.Handler, bool, err
 		return nil, true, fmt.Errorf("duration time must be greater than one second as set via command line option reconcile-sync-loop-period")
 	}
 
+	var (
+		csiNodeInformer   storagev1beta1informer.CSINodeInformer
+		csiDriverInformer storagev1beta1informer.CSIDriverInformer
+	)
+	if utilfeature.DefaultFeatureGate.Enabled(features.CSINodeInfo) {
+		csiNodeInformer = ctx.InformerFactory.Storage().V1beta1().CSINodes()
+	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.CSIDriverRegistry) {
+		csiDriverInformer = ctx.InformerFactory.Storage().V1beta1().CSIDrivers()
+	}
+
 	attachDetachController, attachDetachControllerErr :=
 		attachdetach.NewAttachDetachController(
 			ctx.ClientBuilder.ClientOrDie("attachdetach-controller"),
@@ -285,8 +297,8 @@ func startAttachDetachController(ctx ControllerContext) (http.Handler, bool, err
 			ctx.InformerFactory.Core().V1().Nodes(),
 			ctx.InformerFactory.Core().V1().PersistentVolumeClaims(),
 			ctx.InformerFactory.Core().V1().PersistentVolumes(),
-			ctx.InformerFactory.Storage().V1beta1().CSINodes(),
+			csiNodeInformer,
-			ctx.InformerFactory.Storage().V1beta1().CSIDrivers(),
+			csiDriverInformer,
 			ctx.Cloud,
 			ProbeAttachableVolumePlugins(),
 			GetDynamicPluginProber(ctx.ComponentConfig.PersistentVolumeBinderController.VolumeConfiguration),

pkg/scheduler/algorithm/predicates/predicates.go

@@ -446,12 +446,18 @@ func (c *MaxPDVolumeCountChecker) predicate(pod *v1.Pod, meta PredicateMetadata,
 		return false, nil, fmt.Errorf("node not found")
 	}
 
-	csiNode, err := c.csiNodeLister.Get(node.Name)
+	var (
+		csiNode *v1beta1storage.CSINode
+		err     error
+	)
+	if c.csiNodeLister != nil {
+		csiNode, err = c.csiNodeLister.Get(node.Name)
 		if err != nil {
 			// we don't fail here because the CSINode object is only necessary
 			// for determining whether the migration is enabled or not
 			klog.V(5).Infof("Could not get a CSINode object for the node: %v", err)
 		}
+	}
 
 	// if a plugin has been migrated to a CSI driver, defer to the CSI predicate
 	if c.filter.IsMigrated(csiNode) {

pkg/scheduler/framework/plugins/nodevolumelimits/BUILD

@@ -6,18 +6,23 @@ go_library(
         "azure.go",
         "cinder.go",
         "csi.go",
+        "csinode_helper.go",
         "ebs.go",
         "gce.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/scheduler/framework/plugins/nodevolumelimits",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/features:go_default_library",
         "//pkg/scheduler/algorithm/predicates:go_default_library",
         "//pkg/scheduler/framework/plugins/migration:go_default_library",
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/storage/v1beta1:go_default_library",
     ],
 )
 

pkg/scheduler/framework/plugins/nodevolumelimits/azure.go

@@ -52,12 +52,11 @@ func (pl *AzureDiskLimits) Filter(ctx context.Context, _ *framework.CycleState,
 // NewAzureDisk returns function that initializes a new plugin and returns it.
 func NewAzureDisk(_ *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()
-	csiNodeLister := informerFactory.Storage().V1beta1().CSINodes().Lister()
 	pvLister := informerFactory.Core().V1().PersistentVolumes().Lister()
 	pvcLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
 	scLister := informerFactory.Storage().V1().StorageClasses().Lister()
 
 	return &AzureDiskLimits{
-		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.AzureDiskVolumeFilterType, csiNodeLister, scLister, pvLister, pvcLister),
+		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.AzureDiskVolumeFilterType, getCSINodeListerIfEnabled(informerFactory), scLister, pvLister, pvcLister),
 	}, nil
 }

pkg/scheduler/framework/plugins/nodevolumelimits/cinder.go

@@ -52,11 +52,10 @@ func (pl *CinderLimits) Filter(ctx context.Context, _ *framework.CycleState, pod
 // NewCinder returns function that initializes a new plugin and returns it.
 func NewCinder(_ *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()
-	csiNodeLister := informerFactory.Storage().V1beta1().CSINodes().Lister()
 	pvLister := informerFactory.Core().V1().PersistentVolumes().Lister()
 	pvcLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
 	scLister := informerFactory.Storage().V1().StorageClasses().Lister()
 	return &CinderLimits{
-		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.CinderVolumeFilterType, csiNodeLister, scLister, pvLister, pvcLister),
+		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.CinderVolumeFilterType, getCSINodeListerIfEnabled(informerFactory), scLister, pvLister, pvcLister),
 	}, nil
 }

pkg/scheduler/framework/plugins/nodevolumelimits/csi.go

@@ -52,12 +52,11 @@ func (pl *CSILimits) Filter(ctx context.Context, _ *framework.CycleState, pod *v
 // NewCSI initializes a new plugin and returns it.
 func NewCSI(_ *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()
-	csiNodeLister := informerFactory.Storage().V1beta1().CSINodes().Lister()
 	pvLister := informerFactory.Core().V1().PersistentVolumes().Lister()
 	pvcLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
 	scLister := informerFactory.Storage().V1().StorageClasses().Lister()
 
 	return &CSILimits{
-		predicate: predicates.NewCSIMaxVolumeLimitPredicate(csiNodeLister, pvLister, pvcLister, scLister),
+		predicate: predicates.NewCSIMaxVolumeLimitPredicate(getCSINodeListerIfEnabled(informerFactory), pvLister, pvcLister, scLister),
 	}, nil
 }

pkg/scheduler/framework/plugins/nodevolumelimits/csinode_helper.go

@@ -0,0 +1,32 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package nodevolumelimits
+
+import (
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/informers"
+	v1beta1 "k8s.io/client-go/listers/storage/v1beta1"
+	kubefeatures "k8s.io/kubernetes/pkg/features"
+)
+
+// getCSINodeListerIfEnabled returns the CSINode lister or nil if the feature is disabled
+func getCSINodeListerIfEnabled(factory informers.SharedInformerFactory) v1beta1.CSINodeLister {
+	if !utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CSINodeInfo) {
+		return nil
+	}
+	return factory.Storage().V1beta1().CSINodes().Lister()
+}

pkg/scheduler/framework/plugins/nodevolumelimits/ebs.go

@@ -52,12 +52,11 @@ func (pl *EBSLimits) Filter(ctx context.Context, _ *framework.CycleState, pod *v
 // NewEBS returns function that initializes a new plugin and returns it.
 func NewEBS(_ *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()
-	csiNodeLister := informerFactory.Storage().V1beta1().CSINodes().Lister()
 	pvLister := informerFactory.Core().V1().PersistentVolumes().Lister()
 	pvcLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
 	scLister := informerFactory.Storage().V1().StorageClasses().Lister()
 
 	return &EBSLimits{
-		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.EBSVolumeFilterType, csiNodeLister, scLister, pvLister, pvcLister),
+		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.EBSVolumeFilterType, getCSINodeListerIfEnabled(informerFactory), scLister, pvLister, pvcLister),
 	}, nil
 }

pkg/scheduler/framework/plugins/nodevolumelimits/gce.go

@@ -52,12 +52,11 @@ func (pl *GCEPDLimits) Filter(ctx context.Context, _ *framework.CycleState, pod
 // NewGCEPD returns function that initializes a new plugin and returns it.
 func NewGCEPD(_ *runtime.Unknown, handle framework.FrameworkHandle) (framework.Plugin, error) {
 	informerFactory := handle.SharedInformerFactory()
-	csiNodeLister := informerFactory.Storage().V1beta1().CSINodes().Lister()
 	pvLister := informerFactory.Core().V1().PersistentVolumes().Lister()
 	pvcLister := informerFactory.Core().V1().PersistentVolumeClaims().Lister()
 	scLister := informerFactory.Storage().V1().StorageClasses().Lister()
 
 	return &GCEPDLimits{
-		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.GCEPDVolumeFilterType, csiNodeLister, scLister, pvLister, pvcLister),
+		predicate: predicates.NewMaxPDVolumeCountPredicate(predicates.GCEPDVolumeFilterType, getCSINodeListerIfEnabled(informerFactory), scLister, pvLister, pvcLister),
 	}, nil
 }

pkg/scheduler/scheduler.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	policyv1beta1informers "k8s.io/client-go/informers/policy/v1beta1"
+	storagev1beta1informers "k8s.io/client-go/informers/storage/v1beta1"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/events"
@@ -287,6 +288,11 @@ func New(client clientset.Interface,
 		pdbInformer = informerFactory.Policy().V1beta1().PodDisruptionBudgets()
 	}
 
+	var csiNodeInformer storagev1beta1informers.CSINodeInformer
+	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CSINodeInfo) {
+		csiNodeInformer = informerFactory.Storage().V1beta1().CSINodes()
+	}
+
 	// Set up the configurator which can create schedulers from configs.
 	configurator := NewConfigFactory(&ConfigFactoryArgs{
 		Client:                         client,
@@ -301,7 +307,7 @@ func New(client clientset.Interface,
 		ServiceInformer:                informerFactory.Core().V1().Services(),
 		PdbInformer:                    pdbInformer,
 		StorageClassInformer:           informerFactory.Storage().V1().StorageClasses(),
-		CSINodeInformer:                informerFactory.Storage().V1beta1().CSINodes(),
+		CSINodeInformer:                csiNodeInformer,
 		VolumeBinder:                   volumeBinder,
 		SchedulerCache:                 schedulerCache,
 		HardPodAffinitySymmetricWeight: options.hardPodAffinitySymmetricWeight,