Repository: kubernetes-csi/external-provisioner (go)

Issue:
Consider using Node & CSINodeInfo informers
Currently the topology code makes individual API calls to fetch nodes and CSINodeInfo objects, and in many cases a selector is used to list a subset of objects rather than all of them. Should we use informers instead?

My take is to use an informer for CSINodeInfo objects but not Node objects. CSINodeInfo objects are only updated on CSI driver installation, which is less frequent than provision calls. Node objects are updated constantly, much more frequent than provision calls.

https://github.com/kubernetes-csi/external-provisioner/blob/393375a3af0b8fd4ea3cb9491586670c55a6eb25/pkg/controller/topology.go#L102
https://github.com/kubernetes-csi/external-provisioner/blob/393375a3af0b8fd4ea3cb9491586670c55a6eb25/pkg/controller/topology.go#L138
https://github.com/kubernetes-csi/external-provisioner/blob/393375a3af0b8fd4ea3cb9491586670c55a6eb25/pkg/controller/topology.go#L158
https://github.com/kubernetes-csi/external-provisioner/blob/393375a3af0b8fd4ea3cb9491586670c55a6eb25/pkg/controller/topology.go#L179

/cc @ddebroy

Pull request description:
Use csiNode informer to prevent hitting the API server for all time
**What type of PR is this?**
/kind cleanup

**What this PR does / why we need it**:
Use csiNode shared index informer instead of hitting the API server for all csiNode resource request.

**Which issue(s) this PR fixes**:
Fixes #144

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
```release-note
NONE
```

cc @msau42 

Patch:
diff --git a/cmd/csi-provisioner/csi-provisioner.go b/cmd/csi-provisioner/csi-provisioner.go
index 296491557..4b3866b2c 100644
--- a/cmd/csi-provisioner/csi-provisioner.go
+++ b/cmd/csi-provisioner/csi-provisioner.go
@@ -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)
 	}
 
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index be92ad8ae..50a142e11 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.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) {
diff --git a/pkg/controller/topology.go b/pkg/controller/topology.go
index b6f59e170..2ba6d6eec 100644
--- a/pkg/controller/topology.go
+++ b/pkg/controller/topology.go
@@ -25,11 +25,16 @@ import (
 	"strings"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
-	v1 "k8s.io/api/core/v1"
+	"github.com/kubernetes-csi/csi-lib-utils/connection"
+	"github.com/kubernetes-csi/external-provisioner/pkg/features"
+	"k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/version"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/kubernetes"
+	storagelisters "k8s.io/client-go/listers/storage/v1beta1"
 	"k8s.io/klog"
 )
 
@@ -69,12 +74,18 @@ func GenerateVolumeNodeAffinity(accessibleTopology []*csi.Topology) *v1.VolumeNo
 	}
 }
 
+// SupportsTopology returns whether topology is supported both for plugin and external provisioner
+func SupportsTopology(pluginCapabilities connection.PluginCapabilitySet) bool {
+	return pluginCapabilities[csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS] &&
+		utilfeature.DefaultFeatureGate.Enabled(features.Topology)
+}
+
 // GenerateAccessibilityRequirements returns the CSI TopologyRequirement
 // to pass into the CSI CreateVolume request.
 //
 // This function is only called if the topology feature is enabled
 // in the external-provisioner and the CSI driver implements the
-// CSI accessbility capability. It is disabled by default.
+// CSI accessibility capability. It is disabled by default.
 //
 // If enabled, we require that the K8s API server is on at least
 // K8s 1.14, and that the K8s CSINode feature gate is enabled. In
@@ -140,7 +151,8 @@ func GenerateAccessibilityRequirements(
 	pvcName string,
 	allowedTopologies []v1.TopologySelectorTerm,
 	selectedNode *v1.Node,
-	strictTopology bool) (*csi.TopologyRequirement, error) {
+	strictTopology bool,
+	csiNodeLister storagelisters.CSINodeLister) (*csi.TopologyRequirement, error) {
 	requirement := &csi.TopologyRequirement{}
 
 	var (
@@ -152,7 +164,7 @@ func GenerateAccessibilityRequirements(
 
 	// 1. Get CSINode for the selected node
 	if selectedNode != nil {
-		selectedCSINode, err = getSelectedCSINode(kubeClient, selectedNode)
+		selectedCSINode, err = getSelectedCSINode(csiNodeLister, selectedNode)
 		if err != nil {
 			return nil, err
 		}
@@ -207,7 +219,7 @@ func GenerateAccessibilityRequirements(
 			requisiteTerms = flatten(allowedTopologies)
 		} else {
 			// Aggregate existing topologies in nodes across the entire cluster.
-			requisiteTerms, err = aggregateTopologies(kubeClient, driverName, selectedCSINode)
+			requisiteTerms, err = aggregateTopologies(kubeClient, driverName, selectedCSINode, csiNodeLister)
 			if err != nil {
 				return nil, err
 			}
@@ -263,11 +275,10 @@ func GenerateAccessibilityRequirements(
 
 // getSelectedCSINode returns the CSINode object for the given selectedNode.
 func getSelectedCSINode(
-	kubeClient kubernetes.Interface,
+	csiNodeLister storagelisters.CSINodeLister,
 	selectedNode *v1.Node) (*storage.CSINode, error) {
 
-	// TODO (#144): use informers
-	selectedNodeInfo, err := kubeClient.StorageV1beta1().CSINodes().Get(selectedNode.Name, metav1.GetOptions{})
+	selectedCSINode, err := csiNodeLister.Get(selectedNode.Name)
 	if err != nil {
 		// If the Node is before 1.14, then we fallback to "topology disabled" behavior
 		// to retain backwards compatibility in a single-topology environment with
@@ -289,7 +300,7 @@ func getSelectedCSINode(
 		// error with the API server.
 		return nil, fmt.Errorf("error getting CSINode for selected node %q: %v", selectedNode.Name, err)
 	}
-	return selectedNodeInfo, nil
+	return selectedCSINode, nil
 }
 
 // aggregateTopologies returns all the supported topology values in the cluster that
@@ -297,29 +308,27 @@ func getSelectedCSINode(
 func aggregateTopologies(
 	kubeClient kubernetes.Interface,
 	driverName string,
-	selectedCSINode *storage.CSINode) ([]topologyTerm, error) {
+	selectedCSINode *storage.CSINode,
+	csiNodeLister storagelisters.CSINodeLister) ([]topologyTerm, error) {
 
 	// 1. Determine topologyKeys to use for aggregation
 	var topologyKeys []string
 	if selectedCSINode == nil {
 		// Immediate binding
-
-		// TODO (#144): use informers
-		nodeInfos, err := kubeClient.StorageV1beta1().CSINodes().List(metav1.ListOptions{})
+		csiNodes, err := csiNodeLister.List(labels.Everything())
 		if err != nil {
 			// Require CSINode beta feature on K8s apiserver to be enabled.
 			// We don't want to fallback and provision in the wrong topology if there's some temporary
 			// error with the API server.
 			return nil, fmt.Errorf("error listing CSINodes: %v", err)
 		}
-
-		rand.Shuffle(len(nodeInfos.Items), func(i, j int) {
-			nodeInfos.Items[i], nodeInfos.Items[j] = nodeInfos.Items[j], nodeInfos.Items[i]
+		rand.Shuffle(len(csiNodes), func(i, j int) {
+			csiNodes[i], csiNodes[j] = csiNodes[j], csiNodes[i]
 		})
 
 		// Pick the first node with topology keys
-		for _, nodeInfo := range nodeInfos.Items {
-			topologyKeys = getTopologyKeys(&nodeInfo, driverName)
+		for _, csiNode := range csiNodes {
+			topologyKeys = getTopologyKeys(csiNode, driverName)
 			if topologyKeys != nil {
 				break
 			}
@@ -478,8 +487,8 @@ func sortAndShift(terms []topologyTerm, primary topologyTerm, shiftIndex uint32)
 	return preferredTerms
 }
 
-func getTopologyKeys(nodeInfo *storage.CSINode, driverName string) []string {
-	for _, driver := range nodeInfo.Spec.Drivers {
+func getTopologyKeys(csiNode *storage.CSINode, driverName string) []string {
+	for _, driver := range csiNode.Spec.Drivers {
 		if driver.Name == driverName {
 			return driver.TopologyKeys
 		}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 3ad75a771..dd0c5ca90 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -168,10 +168,6 @@ k8s.io/api/apps/v1
 k8s.io/api/apps/v1beta1
 k8s.io/api/apps/v1beta2
 k8s.io/api/auditregistration/v1alpha1
-k8s.io/api/authentication/v1
-k8s.io/api/authentication/v1beta1
-k8s.io/api/authorization/v1
-k8s.io/api/authorization/v1beta1
 k8s.io/api/autoscaling/v1
 k8s.io/api/autoscaling/v2beta1
 k8s.io/api/autoscaling/v2beta2
@@ -196,18 +192,23 @@ k8s.io/api/scheduling/v1alpha1
 k8s.io/api/scheduling/v1beta1
 k8s.io/api/settings/v1alpha1
 k8s.io/api/storage/v1alpha1
+k8s.io/api/authentication/v1
+k8s.io/api/authentication/v1beta1
+k8s.io/api/authorization/v1
+k8s.io/api/authorization/v1beta1
 # k8s.io/apimachinery v0.0.0-20191006235458-f9f2f3f8ab02 => k8s.io/apimachinery v0.0.0-20190817020851-f2f3a405f61d
 k8s.io/apimachinery/pkg/util/wait
 k8s.io/apimachinery/pkg/api/resource
 k8s.io/apimachinery/pkg/apis/meta/v1
+k8s.io/apimachinery/pkg/labels
 k8s.io/apimachinery/pkg/util/json
 k8s.io/apimachinery/pkg/util/sets
 k8s.io/apimachinery/pkg/util/validation
 k8s.io/apimachinery/pkg/util/version
 k8s.io/apimachinery/pkg/util/runtime
-k8s.io/apimachinery/pkg/api/errors
 k8s.io/apimachinery/pkg/runtime
 k8s.io/apimachinery/pkg/runtime/schema
+k8s.io/apimachinery/pkg/api/errors
 k8s.io/apimachinery/pkg/runtime/serializer/streaming
 k8s.io/apimachinery/pkg/types
 k8s.io/apimachinery/pkg/util/net
@@ -218,7 +219,6 @@ k8s.io/apimachinery/pkg/util/uuid
 k8s.io/apimachinery/pkg/util/intstr
 k8s.io/apimachinery/pkg/conversion
 k8s.io/apimachinery/pkg/fields
-k8s.io/apimachinery/pkg/labels
 k8s.io/apimachinery/pkg/selection
 k8s.io/apimachinery/pkg/util/validation/field
 k8s.io/apimachinery/pkg/runtime/serializer
@@ -226,25 +226,27 @@ k8s.io/apimachinery/pkg/util/strategicpatch
 k8s.io/apimachinery/pkg/version
 k8s.io/apimachinery/pkg/conversion/queryparams
 k8s.io/apimachinery/pkg/util/naming
-k8s.io/apimachinery/pkg/runtime/serializer/json
-k8s.io/apimachinery/pkg/runtime/serializer/versioning
 k8s.io/apimachinery/pkg/api/meta
 k8s.io/apimachinery/pkg/util/cache
 k8s.io/apimachinery/pkg/util/diff
+k8s.io/apimachinery/pkg/runtime/serializer/json
+k8s.io/apimachinery/pkg/runtime/serializer/versioning
 k8s.io/apimachinery/third_party/forked/golang/reflect
 k8s.io/apimachinery/pkg/runtime/serializer/protobuf
 k8s.io/apimachinery/pkg/runtime/serializer/recognizer
 k8s.io/apimachinery/pkg/apis/meta/v1/unstructured
 k8s.io/apimachinery/pkg/util/mergepatch
 k8s.io/apimachinery/third_party/forked/golang/json
+k8s.io/apimachinery/pkg/apis/meta/internalversion
 k8s.io/apimachinery/pkg/util/framer
 k8s.io/apimachinery/pkg/util/yaml
-k8s.io/apimachinery/pkg/apis/meta/internalversion
 k8s.io/apimachinery/pkg/apis/meta/v1beta1
 # k8s.io/apiserver v0.0.0-20190918200908-1e17798da8c1 => k8s.io/apiserver v0.0.0-20190918200908-1e17798da8c1
 k8s.io/apiserver/pkg/util/feature
 # k8s.io/client-go v0.0.0-20190918200256-06eb1244587a => k8s.io/client-go v0.0.0-20190918200256-06eb1244587a
+k8s.io/client-go/informers
 k8s.io/client-go/kubernetes
+k8s.io/client-go/listers/storage/v1beta1
 k8s.io/client-go/rest
 k8s.io/client-go/tools/clientcmd
 k8s.io/client-go/util/workqueue
@@ -255,6 +257,25 @@ k8s.io/client-go/tools/leaderelection/resourcelock
 k8s.io/client-go/tools/record
 k8s.io/client-go/discovery
 k8s.io/client-go/util/flowcontrol
+k8s.io/client-go/informers/admissionregistration
+k8s.io/client-go/informers/apps
+k8s.io/client-go/informers/auditregistration
+k8s.io/client-go/informers/autoscaling
+k8s.io/client-go/informers/batch
+k8s.io/client-go/informers/certificates
+k8s.io/client-go/informers/coordination
+k8s.io/client-go/informers/core
+k8s.io/client-go/informers/events
+k8s.io/client-go/informers/extensions
+k8s.io/client-go/informers/internalinterfaces
+k8s.io/client-go/informers/networking
+k8s.io/client-go/informers/node
+k8s.io/client-go/informers/policy
+k8s.io/client-go/informers/rbac
+k8s.io/client-go/informers/scheduling
+k8s.io/client-go/informers/settings
+k8s.io/client-go/informers/storage
+k8s.io/client-go/tools/cache
 k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1
 k8s.io/client-go/kubernetes/typed/apps/v1
 k8s.io/client-go/kubernetes/typed/apps/v1beta1
@@ -300,38 +321,50 @@ k8s.io/client-go/util/cert
 k8s.io/client-go/tools/auth
 k8s.io/client-go/tools/clientcmd/api/latest
 k8s.io/client-go/util/homedir
-k8s.io/client-go/informers
-k8s.io/client-go/tools/cache
 k8s.io/client-go/tools/reference
 k8s.io/client-go/kubernetes/fake
 k8s.io/client-go/testing
 k8s.io/client-go/tools/record/util
+k8s.io/client-go/informers/admissionregistration/v1beta1
+k8s.io/client-go/informers/apps/v1
+k8s.io/client-go/informers/apps/v1beta1
+k8s.io/client-go/informers/apps/v1beta2
+k8s.io/client-go/informers/auditregistration/v1alpha1
+k8s.io/client-go/informers/autoscaling/v1
+k8s.io/client-go/informers/autoscaling/v2beta1
+k8s.io/client-go/informers/autoscaling/v2beta2
+k8s.io/client-go/informers/batch/v1
+k8s.io/client-go/informers/batch/v1beta1
+k8s.io/client-go/informers/batch/v2alpha1
+k8s.io/client-go/informers/certificates/v1beta1
+k8s.io/client-go/informers/coordination/v1
+k8s.io/client-go/informers/coordination/v1beta1
+k8s.io/client-go/informers/core/v1
+k8s.io/client-go/informers/events/v1beta1
+k8s.io/client-go/informers/extensions/v1beta1
+k8s.io/client-go/informers/networking/v1
+k8s.io/client-go/informers/networking/v1beta1
+k8s.io/client-go/informers/node/v1alpha1
+k8s.io/client-go/informers/node/v1beta1
+k8s.io/client-go/informers/policy/v1beta1
+k8s.io/client-go/informers/rbac/v1
+k8s.io/client-go/informers/rbac/v1alpha1
+k8s.io/client-go/informers/rbac/v1beta1
+k8s.io/client-go/informers/scheduling/v1
+k8s.io/client-go/informers/scheduling/v1alpha1
+k8s.io/client-go/informers/scheduling/v1beta1
+k8s.io/client-go/informers/settings/v1alpha1
+k8s.io/client-go/informers/storage/v1
+k8s.io/client-go/informers/storage/v1alpha1
+k8s.io/client-go/informers/storage/v1beta1
+k8s.io/client-go/tools/pager
+k8s.io/client-go/util/retry
 k8s.io/client-go/pkg/apis/clientauthentication
 k8s.io/client-go/pkg/apis/clientauthentication/v1alpha1
 k8s.io/client-go/pkg/apis/clientauthentication/v1beta1
 k8s.io/client-go/util/connrotation
 k8s.io/client-go/util/keyutil
 k8s.io/client-go/tools/clientcmd/api/v1
-k8s.io/client-go/informers/admissionregistration
-k8s.io/client-go/informers/apps
-k8s.io/client-go/informers/auditregistration
-k8s.io/client-go/informers/autoscaling
-k8s.io/client-go/informers/batch
-k8s.io/client-go/informers/certificates
-k8s.io/client-go/informers/coordination
-k8s.io/client-go/informers/core
-k8s.io/client-go/informers/events
-k8s.io/client-go/informers/extensions
-k8s.io/client-go/informers/internalinterfaces
-k8s.io/client-go/informers/networking
-k8s.io/client-go/informers/node
-k8s.io/client-go/informers/policy
-k8s.io/client-go/informers/rbac
-k8s.io/client-go/informers/scheduling
-k8s.io/client-go/informers/settings
-k8s.io/client-go/informers/storage
-k8s.io/client-go/tools/pager
-k8s.io/client-go/util/retry
 k8s.io/client-go/discovery/fake
 k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1/fake
 k8s.io/client-go/kubernetes/typed/apps/v1/fake
@@ -369,38 +402,6 @@ k8s.io/client-go/kubernetes/typed/settings/v1alpha1/fake
 k8s.io/client-go/kubernetes/typed/storage/v1/fake
 k8s.io/client-go/kubernetes/typed/storage/v1alpha1/fake
 k8s.io/client-go/kubernetes/typed/storage/v1beta1/fake
-k8s.io/client-go/informers/admissionregistration/v1beta1
-k8s.io/client-go/informers/apps/v1
-k8s.io/client-go/informers/apps/v1beta1
-k8s.io/client-go/informers/apps/v1beta2
-k8s.io/client-go/informers/auditregistration/v1alpha1
-k8s.io/client-go/informers/autoscaling/v1
-k8s.io/client-go/informers/autoscaling/v2beta1
-k8s.io/client-go/informers/autoscaling/v2beta2
-k8s.io/client-go/informers/batch/v1
-k8s.io/client-go/informers/batch/v1beta1
-k8s.io/client-go/informers/batch/v2alpha1
-k8s.io/client-go/informers/certificates/v1beta1
-k8s.io/client-go/informers/coordination/v1
-k8s.io/client-go/informers/coordination/v1beta1
-k8s.io/client-go/informers/core/v1
-k8s.io/client-go/informers/events/v1beta1
-k8s.io/client-go/informers/extensions/v1beta1
-k8s.io/client-go/informers/networking/v1
-k8s.io/client-go/informers/networking/v1beta1
-k8s.io/client-go/informers/node/v1alpha1
-k8s.io/client-go/informers/node/v1beta1
-k8s.io/client-go/informers/policy/v1beta1
-k8s.io/client-go/informers/rbac/v1
-k8s.io/client-go/informers/rbac/v1alpha1
-k8s.io/client-go/informers/rbac/v1beta1
-k8s.io/client-go/informers/scheduling/v1
-k8s.io/client-go/informers/scheduling/v1alpha1
-k8s.io/client-go/informers/scheduling/v1beta1
-k8s.io/client-go/informers/settings/v1alpha1
-k8s.io/client-go/informers/storage/v1
-k8s.io/client-go/informers/storage/v1alpha1
-k8s.io/client-go/informers/storage/v1beta1
 k8s.io/client-go/listers/admissionregistration/v1beta1
 k8s.io/client-go/listers/apps/v1
 k8s.io/client-go/listers/apps/v1beta1
@@ -432,7 +433,6 @@ k8s.io/client-go/listers/scheduling/v1beta1
 k8s.io/client-go/listers/settings/v1alpha1
 k8s.io/client-go/listers/storage/v1
 k8s.io/client-go/listers/storage/v1alpha1
-k8s.io/client-go/listers/storage/v1beta1
 # k8s.io/cloud-provider v0.0.0-20191005121959-0a96ff5a97f9 => k8s.io/cloud-provider v0.0.0-20190918203125-ae665f80358a
 k8s.io/cloud-provider/volume
 # k8s.io/component-base v0.0.0-20190918200425-ed2f0867c778 => k8s.io/component-base v0.0.0-20190918200425-ed2f0867c778


Test patch:
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index f4b298565..94f46af1b 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.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()
diff --git a/pkg/controller/topology_test.go b/pkg/controller/topology_test.go
index 80d50b14d..052a06b8b 100644
--- a/pkg/controller/topology_test.go
+++ b/pkg/controller/topology_test.go
@@ -21,11 +21,13 @@ import (
 	"testing"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
-	v1 "k8s.io/api/core/v1"
+	"k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/informers"
 	fakeclientset "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/listers/storage/v1beta1"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
 )
 
@@ -386,9 +388,12 @@ func TestStatefulSetSpreading(t *testing.T) {
 	}
 
 	nodes := buildNodes(nodeLabels, k8sTopologyBetaVersion.String())
-	nodeInfos := buildNodeInfos(topologyKeys)
+	csiNodes := buildCSINodes(topologyKeys)
 
-	kubeClient := fakeclientset.NewSimpleClientset(nodes, nodeInfos)
+	kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
+
+	csiNodeLister, stopChan := csiNodeLister(kubeClient, t)
+	defer close(stopChan)
 
 	for name, tc := range testcases {
 		for _, strictTopology := range []bool{false, true} {
@@ -404,6 +409,7 @@ func TestStatefulSetSpreading(t *testing.T) {
 				tc.allowedTopologies,
 				nil,
 				strictTopology,
+				csiNodeLister,
 			)
 
 			if err != nil {
@@ -797,6 +803,7 @@ func TestAllowedTopologies(t *testing.T) {
 				tc.allowedTopologies,
 				nil, /* selectedNode */
 				strictTopology,
+				nil,
 			)
 
 			if err != nil {
@@ -1069,9 +1076,13 @@ func TestTopologyAggregation(t *testing.T) {
 				nodeVersion = preBetaNodeVersion
 			}
 			nodes := buildNodes(tc.nodeLabels, nodeVersion)
-			nodeInfos := buildNodeInfos(tc.topologyKeys)
+			csiNodes := buildCSINodes(tc.topologyKeys)
+
+			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
+
+			csiNodeLister, stopChan := csiNodeLister(kubeClient, t)
+			defer close(stopChan)
 
-			kubeClient := fakeclientset.NewSimpleClientset(nodes, nodeInfos)
 			var selectedNode *v1.Node
 			if tc.hasSelectedNode {
 				selectedNode = &nodes.Items[0]
@@ -1083,6 +1094,7 @@ func TestTopologyAggregation(t *testing.T) {
 				nil, /* allowedTopologies */
 				selectedNode,
 				strictTopology,
+				csiNodeLister,
 			)
 
 			if tc.expectError {
@@ -1315,11 +1327,14 @@ func TestPreferredTopologies(t *testing.T) {
 			t.Logf("test: %s", name)
 
 			nodes := buildNodes(tc.nodeLabels, k8sTopologyBetaVersion.String())
-			nodeInfos := buildNodeInfos(tc.topologyKeys)
+			csiNodes := buildCSINodes(tc.topologyKeys)
 
-			kubeClient := fakeclientset.NewSimpleClientset(nodes, nodeInfos)
+			kubeClient := fakeclientset.NewSimpleClientset(nodes, csiNodes)
 			selectedNode := &nodes.Items[0]
 
+			csiNodeLister, stopChan := csiNodeLister(kubeClient, t)
+			defer close(stopChan)
+
 			requirements, err := GenerateAccessibilityRequirements(
 				kubeClient,
 				testDriverName,
@@ -1327,6 +1342,7 @@ func TestPreferredTopologies(t *testing.T) {
 				tc.allowedTopologies,
 				selectedNode,
 				strictTopology,
+				csiNodeLister,
 			)
 
 			if tc.expectError {
@@ -1383,10 +1399,10 @@ func buildNodes(nodeLabels []map[string]string, nodeVersion string) *v1.NodeList
 	return list
 }
 
-func buildNodeInfos(nodeInfos []map[string][]string) *storage.CSINodeList {
+func buildCSINodes(csiNodes []map[string][]string) *storage.CSINodeList {
 	list := &storage.CSINodeList{}
 	i := 0
-	for _, nodeInfo := range nodeInfos {
+	for _, csiNode := range csiNodes {
 		nodeName := fmt.Sprintf("node-%d", i)
 		n := storage.CSINode{
 			ObjectMeta: metav1.ObjectMeta{
@@ -1394,7 +1410,7 @@ func buildNodeInfos(nodeInfos []map[string][]string) *storage.CSINodeList {
 			},
 		}
 		var csiDrivers []storage.CSINodeDriver
-		for driver, topologyKeys := range nodeInfo {
+		for driver, topologyKeys := range csiNode {
 			driverInfos := []storage.CSINodeDriver{
 				{
 					Name:         driver,
@@ -1528,3 +1544,12 @@ func requisiteEqual(t1, t2 []*csi.Topology) bool {
 
 	return unchecked.Len() == 0
 }
+
+func csiNodeLister(kubeClient *fakeclientset.Clientset, t *testing.T) (v1beta1.CSINodeLister, chan struct{}) {
+	factory := informers.NewSharedInformerFactory(kubeClient, ResyncPeriodOfCsiNodeInformer)
+	stopChan := make(chan struct{})
+	csiNodeLister := factory.Storage().V1beta1().CSINodes().Lister()
+	factory.Start(stopChan)
+	factory.WaitForCacheSync(stopChan)
+	return csiNodeLister, stopChan
+}