introduce concurrent option to worker

Author: zqzten

pkg/controller/federatedtypeconfig/controller.go

@@ -90,7 +90,7 @@ func newController(config *util.ControllerConfig) (*Controller, error) {
 		stopChannels:     make(map[string]chan struct{}),
 	}
 
-	c.worker = util.NewReconcileWorker("federatedtypeconfig", c.reconcile, util.WorkerTiming{})
+	c.worker = util.NewReconcileWorker("federatedtypeconfig", c.reconcile, util.WorkerOptions{})
 
 	// Only watch the KubeFed namespace to ensure
 	// restrictive authz can be applied to a namespaced

pkg/controller/schedulingmanager/controller.go

@@ -92,7 +92,7 @@ func newSchedulingManager(config *util.ControllerConfig) (*SchedulingManager, er
 		schedulers: util.NewSafeMap(),
 	}
 
-	c.worker = util.NewReconcileWorker("schedulingmanager", c.reconcile, util.WorkerTiming{})
+	c.worker = util.NewReconcileWorker("schedulingmanager", c.reconcile, util.WorkerOptions{})
 
 	var err error
 	c.store, c.controller, err = util.NewGenericInformer(

pkg/controller/schedulingpreference/controller.go

@@ -105,8 +105,10 @@ func newSchedulingPreferenceController(config *util.ControllerConfig, scheduling
 		eventRecorder:           recorder,
 	}
 
-	s.worker = util.NewReconcileWorker(strings.ToLower(schedulingType.Kind), s.reconcile, util.WorkerTiming{
-		ClusterSyncDelay: s.clusterAvailableDelay,
+	s.worker = util.NewReconcileWorker(strings.ToLower(schedulingType.Kind), s.reconcile, util.WorkerOptions{
+		WorkerTiming: util.WorkerTiming{
+			ClusterSyncDelay: s.clusterAvailableDelay,
+		},
 	})
 
 	eventHandlers := schedulingtypes.SchedulerEventHandlers{

pkg/controller/status/controller.go

@@ -123,8 +123,10 @@ func newKubeFedStatusController(controllerConfig *util.ControllerConfig, typeCon
 		fedNamespace:            controllerConfig.KubeFedNamespace,
 	}
 
-	s.worker = util.NewReconcileWorker(strings.ToLower(statusAPIResource.Kind), s.reconcile, util.WorkerTiming{
-		ClusterSyncDelay: s.clusterAvailableDelay,
+	s.worker = util.NewReconcileWorker(strings.ToLower(statusAPIResource.Kind), s.reconcile, util.WorkerOptions{
+		WorkerTiming: util.WorkerTiming{
+			ClusterSyncDelay: s.clusterAvailableDelay,
+		},
 	})
 
 	// Build deliverer for triggering cluster reconciliations.

pkg/controller/sync/controller.go

@@ -132,8 +132,10 @@ func newKubeFedSyncController(controllerConfig *util.ControllerConfig, typeConfi
 		rawResourceStatusCollection: controllerConfig.RawResourceStatusCollection,
 	}
 
-	s.worker = util.NewReconcileWorker(strings.ToLower(federatedTypeAPIResource.Kind), s.reconcile, util.WorkerTiming{
-		ClusterSyncDelay: s.clusterAvailableDelay,
+	s.worker = util.NewReconcileWorker(strings.ToLower(federatedTypeAPIResource.Kind), s.reconcile, util.WorkerOptions{
+		WorkerTiming: util.WorkerTiming{
+			ClusterSyncDelay: s.clusterAvailableDelay,
+		},
 	})
 
 	// Build deliverer for triggering cluster reconciliations.

pkg/controller/util/worker.go

@@ -38,6 +38,13 @@ type ReconcileWorker interface {
 	SetDelay(retryDelay, clusterSyncDelay time.Duration)
 }
 
+type WorkerOptions struct {
+	WorkerTiming
+
+	// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
+	MaxConcurrentReconciles int
+}
+
 type WorkerTiming struct {
 	Interval         time.Duration
 	RetryDelay       time.Duration
@@ -53,6 +60,8 @@ type asyncWorker struct {
 
 	timing WorkerTiming
 
+	maxConcurrentReconciles int
+
 	// For triggering reconciliation of a single resource. This is
 	// used when there is an add/update/delete operation on a resource
 	// in either the API of the cluster hosting KubeFed or in the API
@@ -66,26 +75,30 @@ type asyncWorker struct {
 	backoff *flowcontrol.Backoff
 }
 
-func NewReconcileWorker(name string, reconcile ReconcileFunc, timing WorkerTiming) ReconcileWorker {
-	if timing.Interval == 0 {
-		timing.Interval = time.Second * 1
+func NewReconcileWorker(name string, reconcile ReconcileFunc, options WorkerOptions) ReconcileWorker {
+	if options.Interval == 0 {
+		options.Interval = time.Second * 1
 	}
-	if timing.RetryDelay == 0 {
-		timing.RetryDelay = time.Second * 10
+	if options.RetryDelay == 0 {
+		options.RetryDelay = time.Second * 10
 	}
-	if timing.InitialBackoff == 0 {
-		timing.InitialBackoff = time.Second * 5
+	if options.InitialBackoff == 0 {
+		options.InitialBackoff = time.Second * 5
 	}
-	if timing.MaxBackoff == 0 {
-		timing.MaxBackoff = time.Minute
+	if options.MaxBackoff == 0 {
+		options.MaxBackoff = time.Minute
+	}
+	if options.MaxConcurrentReconciles == 0 {
+		options.MaxConcurrentReconciles = 1
 	}
 	return &asyncWorker{
-		name:      name,
-		reconcile: reconcile,
-		timing:    timing,
-		deliverer: NewDelayingDeliverer(),
-		queue:     workqueue.NewNamed(name),
-		backoff:   flowcontrol.NewBackOff(timing.InitialBackoff, timing.MaxBackoff),
+		name:                    name,
+		reconcile:               reconcile,
+		timing:                  options.WorkerTiming,
+		maxConcurrentReconciles: options.MaxConcurrentReconciles,
+		deliverer:               NewDelayingDeliverer(),
+		queue:                   workqueue.NewNamed(name),
+		backoff:                 flowcontrol.NewBackOff(options.InitialBackoff, options.MaxBackoff),
 	}
 }
 
@@ -122,7 +135,10 @@ func (w *asyncWorker) Run(stopChan <-chan struct{}) {
 			w.queue.Add(*qualifiedName)
 		}
 	})
-	go wait.Until(w.worker, w.timing.Interval, stopChan)
+
+	for i := 0; i < w.maxConcurrentReconciles; i++ {
+		go wait.Until(w.worker, w.timing.Interval, stopChan)
+	}
 
 	// Ensure all goroutines are cleaned up when the stop channel closes
 	go func() {

pkg/controller/util/worker_test.go

@@ -61,7 +61,7 @@ func TestDeduplicate(t *testing.T) {
 			})
 			return StatusAllOK
 		},
-		WorkerTiming{},
+		WorkerOptions{},
 	)
 
 	// run worker