add rate limiting workqueue for node watcher

kubernetes-csi · Feb 6, 2022 · d2eaf25 · d2eaf25
1 parent fba209e
commit d2eaf25
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 5 deletions.
diff --git a/pkg/controller/node_watcher.go b/pkg/controller/node_watcher.go
@@ -71,9 +71,17 @@ type NodeWatcher struct {
 }
 
 // NewNodeWatcher creates a node watcher object that will watch the nodes
-func NewNodeWatcher(driverName string, client kubernetes.Interface, volumeLister corelisters.PersistentVolumeLister,
-	pvcLister corelisters.PersistentVolumeClaimLister, nodeInformer coreinformers.NodeInformer,
-	recorder record.EventRecorder, pvcToPodsCache *util.PVCToPodsCache, nodeWorkerExecuteInterval time.Duration, nodeListAndAddInterval time.Duration) *NodeWatcher {
+func NewNodeWatcher(driverName string,
+	client kubernetes.Interface,
+	volumeLister corelisters.PersistentVolumeLister,
+	pvcLister corelisters.PersistentVolumeClaimLister,
+	nodeInformer coreinformers.NodeInformer,
+	recorder record.EventRecorder,
+	pvcToPodsCache *util.PVCToPodsCache,
+	nodeWorkerExecuteInterval time.Duration,
+	nodeListAndAddInterval time.Duration,
+	contentRateLimiter workqueue.RateLimiter,
+) *NodeWatcher {
 
 	watcher := &NodeWatcher{
 		driverName:                driverName,
@@ -83,7 +91,7 @@ func NewNodeWatcher(driverName string, client kubernetes.Interface, volumeLister
 		recorder:                  recorder,
 		volumeLister:              volumeLister,
 		pvcLister:                 pvcLister,
-		nodeQueue:                 workqueue.NewNamed("nodes"),
+		nodeQueue:                 workqueue.NewNamedRateLimitingQueue(contentRateLimiter, "nodes"),
 		nodeFirstBrokenMap:        make(map[string]time.Time),
 		nodeEverMarkedDown:        make(map[string]bool),
 		pvcToPodsCache:            pvcToPodsCache,

diff --git a/pkg/controller/pv_monitor_controller.go b/pkg/controller/pv_monitor_controller.go
@@ -161,7 +161,18 @@ func NewPVMonitorController(
 	})
 
 	if ctrl.enableNodeWatcher {
-		ctrl.nodeWatcher = NewNodeWatcher(ctrl.driverName, ctrl.client, ctrl.pvLister, ctrl.pvcLister, nodeInformer, ctrl.eventRecorder, ctrl.pvcToPodsCache, option.NodeWorkerExecuteInterval, option.NodeListAndAddInterval)
+		ctrl.nodeWatcher = NewNodeWatcher(
+			ctrl.driverName,
+			ctrl.client,
+			ctrl.pvLister,
+			ctrl.pvcLister,
+			nodeInformer,
+			ctrl.eventRecorder,
+			ctrl.pvcToPodsCache,
+			option.NodeWorkerExecuteInterval,
+			option.NodeListAndAddInterval,
+			contentRateLimiter,
+		)
 	}
 
 	ctrl.pvChecker = handler.NewPVHealthConditionChecker(option.DriverName, conn, client, option.ContextTimeout, ctrl.pvcLister, ctrl.pvLister, eventInformer, ctrl.eventRecorder)