Repository: kubernetes-sigs/kubefed (go)

Issue:
Make worker count configurable for different controllers
**What would you like to be added**:
Since #1379 has already fixed the reconciliation conflicts of same object, a worker count option should be imported to `ReconcileWorker` for better performance on different reconciliations.

**Why is this needed**:
As the comments above, a controller-friendly option for better performance.

<!-- DO NOT EDIT BELOW THIS LINE -->
/kind feature


Pull request:
feat: make concurrency of the worker configurable
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, read our contributor guidelines https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
3. Add an entry to CHANGELOG.md if the PR represents a user-visible change.
4. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What this PR does / why we need it**:
As to gain better performance on different reconciliations, the concurrency of worker should be made configurable for different controllers.

In short, this PR does two things below:
* Introduce a new `MaxConcurrentReconciles` option to worker for inner controller concurrency settings.
* Introduce a new `Concurrency` section to KubeFedConfg to expose the concurrency settings to users.

**Which issue(s) this PR fixes**:
Fixes #1392

Patch:
diff --git a/.gitignore b/.gitignore
index de1ad0e7..475be01b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,3 +6,10 @@
 # Ignore binaries built by the makefile to avoid accidentally committing them
 /images/kubefed/hyperfed
 /images/kubefed/controller-manager
+
+# editor and IDE paraphernalia
+.idea
+.vscode
+
+# macOS paraphernalia
+.DS_Store
diff --git a/charts/kubefed/README.md b/charts/kubefed/README.md
index 237dc096..7e771bf5 100644
--- a/charts/kubefed/README.md
+++ b/charts/kubefed/README.md
@@ -123,7 +123,9 @@ chart and their default values.
 | controllermanager.clusterHealthCheckFailureThreshold | Minimum consecutive failures for the cluster health to be considered failed after having succeeded.                                                                          | 3                               |
 | controllermanager.clusterHealthCheckSuccessThreshold | Minimum consecutive successes for the cluster health to be considered successful after having failed.                                                                        | 1                               |
 | controllermanager.clusterHealthCheckTimeout          | Duration after which the cluster health check times out.                                                                                                                     | 3s                              |
-| controllermanager.syncController.adoptResources      | Whether to adopt pre-existing resource in member clusters.                                                                                                        		        | Enabled                         |
+| controllermanager.syncController.maxConcurrentReconciles | The maximum number of concurrent Reconciles of sync controller which can be run.                                                                                         | 1                               |
+| controllermanager.syncController.adoptResources          | Whether to adopt pre-existing resource in member clusters.                                                                                                        		  | Enabled                         |
+| controllermanager.statusController.maxConcurrentReconciles | The maximum number of concurrent Reconciles of status controller which can be run.                                                                                     | 1                               |
 | controllermanager.service.labels                     | Kubernetes labels attached to the controller manager's services                                                                                                       		    | {}                              |
 | controllermanager.certManager.enabled             | Specifies whether to enable the usage of the cert-manager for the certificates generation.                                                                                      | false                           |
 | controllermanager.certManager.rootCertificate.organizations       | Specifies the list of organizations to include in the cert-manager generated root certificate.                                                                  | []                              |
diff --git a/charts/kubefed/charts/controllermanager/crds/crds.yaml b/charts/kubefed/charts/controllermanager/crds/crds.yaml
index 98858fdd..30ff31d9 100644
--- a/charts/kubefed/charts/controllermanager/crds/crds.yaml
+++ b/charts/kubefed/charts/controllermanager/crds/crds.yaml
@@ -692,12 +692,25 @@ spec:
                   `Namespaced` or `Cluster`. `Namespaced` indicates that the KubeFed
                   namespace will be the only target of the control plane.
                 type: string
+              statusController:
+                properties:
+                  maxConcurrentReconciles:
+                    description: The maximum number of concurrent Reconciles of status
+                      controller which can be run. Defaults to 1.
+                    format: int64
+                    type: integer
+                type: object
               syncController:
                 properties:
                   adoptResources:
                     description: Whether to adopt pre-existing resources in member
                       clusters. Defaults to "Enabled".
                     type: string
+                  maxConcurrentReconciles:
+                    description: The maximum number of concurrent Reconciles of sync
+                      controller which can be run. Defaults to 1.
+                    format: int64
+                    type: integer
                 type: object
             required:
             - scope
diff --git a/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml b/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
index 34110b38..f0c30716 100644
--- a/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
+++ b/charts/kubefed/charts/controllermanager/templates/kubefedconfig.yaml
@@ -19,7 +19,10 @@ spec:
     successThreshold: {{ .Values.clusterHealthCheckSuccessThreshold | default 1 }}
     timeout: {{ .Values.clusterHealthCheckTimeout | default "3s" | quote }}
   syncController:
+    maxConcurrentReconciles: {{ .Values.syncController.maxConcurrentReconciles | default 1 }}
     adoptResources: {{ .Values.syncController.adoptResources | default "Enabled" | quote }}
+  statusController:
+    maxConcurrentReconciles: {{ .Values.statusController.maxConcurrentReconciles | default 1 }}
   featureGates:
 {{- if .Values.featureGates }}
   - name: PushReconciler
diff --git a/charts/kubefed/values.yaml b/charts/kubefed/values.yaml
index 813c3ec2..c37715cf 100644
--- a/charts/kubefed/values.yaml
+++ b/charts/kubefed/values.yaml
@@ -18,7 +18,10 @@ controllermanager:
   ## Supported options are `configmaps` and `endpoints`
   leaderElectResourceLock:
   syncController:
+    maxConcurrentReconciles:
     adoptResources:
+  statusController:
+    maxConcurrentReconciles:
   ## Value of feature gates item should be either `Enabled` or `Disabled`
   featureGates:
     PushReconciler:
diff --git a/cmd/controller-manager/app/controller-manager.go b/cmd/controller-manager/app/controller-manager.go
index bef051d5..f28c4c37 100644
--- a/cmd/controller-manager/app/controller-manager.go
+++ b/cmd/controller-manager/app/controller-manager.go
@@ -365,6 +365,9 @@ func setOptionsByKubeFedConfig(opts *options.Options) {
 	opts.ClusterHealthCheckConfig.FailureThreshold = *spec.ClusterHealthCheck.FailureThreshold
 	opts.ClusterHealthCheckConfig.SuccessThreshold = *spec.ClusterHealthCheck.SuccessThreshold
 
+	opts.Config.MaxConcurrentSyncReconciles = *spec.SyncController.MaxConcurrentReconciles
+	opts.Config.MaxConcurrentStatusReconciles = *spec.StatusController.MaxConcurrentReconciles
+
 	opts.Config.SkipAdoptingResources = *spec.SyncController.AdoptResources == corev1b1.AdoptResourcesDisabled
 
 	var featureGates = make(map[string]bool)
diff --git a/config/kubefedconfig.yaml b/config/kubefedconfig.yaml
index 1fcb77df..777022dc 100644
--- a/config/kubefedconfig.yaml
+++ b/config/kubefedconfig.yaml
@@ -26,4 +26,7 @@ spec:
     successThreshold: 1
     timeout: 3s
   syncController:
+    maxConcurrentReconciles: 1
     adoptResources: Enabled
+  statusController:
+    maxConcurrentReconciles: 1
diff --git a/pkg/apis/core/v1beta1/defaults/defaults.go b/pkg/apis/core/v1beta1/defaults/defaults.go
index 31ce87d0..1abb5fc1 100644
--- a/pkg/apis/core/v1beta1/defaults/defaults.go
+++ b/pkg/apis/core/v1beta1/defaults/defaults.go
@@ -39,6 +39,9 @@ const (
 	DefaultClusterHealthCheckFailureThreshold = 3
 	DefaultClusterHealthCheckSuccessThreshold = 1
 	DefaultClusterHealthCheckTimeout          = 3 * time.Second
+
+	DefaultSyncControllerMaxConcurrentReconciles   = 1
+	DefaultStatusControllerMaxConcurrentReconciles = 1
 )
 
 func SetDefaultKubeFedConfig(fedConfig *v1beta1.KubeFedConfig) {
@@ -87,10 +90,18 @@ func SetDefaultKubeFedConfig(fedConfig *v1beta1.KubeFedConfig) {
 		spec.SyncController = &v1beta1.SyncControllerConfig{}
 	}
 
+	setInt64(&spec.SyncController.MaxConcurrentReconciles, DefaultSyncControllerMaxConcurrentReconciles)
+
 	if spec.SyncController.AdoptResources == nil {
 		spec.SyncController.AdoptResources = new(v1beta1.ResourceAdoption)
 		*spec.SyncController.AdoptResources = v1beta1.AdoptResourcesEnabled
 	}
+
+	if spec.StatusController == nil {
+		spec.StatusController = &v1beta1.StatusControllerConfig{}
+	}
+
+	setInt64(&spec.StatusController.MaxConcurrentReconciles, DefaultStatusControllerMaxConcurrentReconciles)
 }
 
 func setDefaultKubeFedFeatureGates(fgc []v1beta1.FeatureGatesConfig) []v1beta1.FeatureGatesConfig {
diff --git a/pkg/apis/core/v1beta1/kubefedconfig_types.go b/pkg/apis/core/v1beta1/kubefedconfig_types.go
index b86930b9..73a201b9 100644
--- a/pkg/apis/core/v1beta1/kubefedconfig_types.go
+++ b/pkg/apis/core/v1beta1/kubefedconfig_types.go
@@ -37,6 +37,8 @@ type KubeFedConfigSpec struct {
 	ClusterHealthCheck *ClusterHealthCheckConfig `json:"clusterHealthCheck,omitempty"`
 	// +optional
 	SyncController *SyncControllerConfig `json:"syncController,omitempty"`
+	// +optional
+	StatusController *StatusControllerConfig `json:"statusController,omitempty"`
 }
 
 type DurationConfig struct {
@@ -105,6 +107,10 @@ type ClusterHealthCheckConfig struct {
 }
 
 type SyncControllerConfig struct {
+	// The maximum number of concurrent Reconciles of sync controller which can be run.
+	// Defaults to 1.
+	// +optional
+	MaxConcurrentReconciles *int64 `json:"maxConcurrentReconciles,omitempty"`
 	// Whether to adopt pre-existing resources in member clusters. Defaults to
 	// "Enabled".
 	// +optional
@@ -118,6 +124,13 @@ const (
 	AdoptResourcesDisabled ResourceAdoption = "Disabled"
 )
 
+type StatusControllerConfig struct {
+	// The maximum number of concurrent Reconciles of status controller which can be run.
+	// Defaults to 1.
+	// +optional
+	MaxConcurrentReconciles *int64 `json:"maxConcurrentReconciles,omitempty"`
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:resource:path=kubefedconfigs
 
diff --git a/pkg/apis/core/v1beta1/validation/validation.go b/pkg/apis/core/v1beta1/validation/validation.go
index 45ef4b19..b63145f9 100644
--- a/pkg/apis/core/v1beta1/validation/validation.go
+++ b/pkg/apis/core/v1beta1/validation/validation.go
@@ -356,10 +356,19 @@ func ValidateKubeFedConfig(kubeFedConfig, oldKubeFedConfig *v1beta1.KubeFedConfi
 	case sync.AdoptResources == nil:
 		allErrs = append(allErrs, field.Required(adoptPath, ""))
 	default:
+		allErrs = append(allErrs, validateIntPtrGreaterThan0(syncPath.Child("maxConcurrentReconciles"), sync.MaxConcurrentReconciles)...)
 		allErrs = append(allErrs, validateEnumStrings(adoptPath, string(*sync.AdoptResources),
 			[]string{string(v1beta1.AdoptResourcesEnabled), string(v1beta1.AdoptResourcesDisabled)})...)
 	}
 
+	statusController := spec.StatusController
+	statusControllerPath := specPath.Child("statusController")
+	if statusController == nil {
+		allErrs = append(allErrs, field.Required(statusControllerPath, ""))
+	} else {
+		allErrs = append(allErrs, validateIntPtrGreaterThan0(statusControllerPath.Child("maxConcurrentReconciles"), statusController.MaxConcurrentReconciles)...)
+	}
+
 	return allErrs
 }
 
diff --git a/pkg/apis/core/v1beta1/zz_generated.deepcopy.go b/pkg/apis/core/v1beta1/zz_generated.deepcopy.go
index 694be4fa..013f3a6c 100644
--- a/pkg/apis/core/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/apis/core/v1beta1/zz_generated.deepcopy.go
@@ -454,6 +454,11 @@ func (in *KubeFedConfigSpec) DeepCopyInto(out *KubeFedConfigSpec) {
 		*out = new(SyncControllerConfig)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.StatusController != nil {
+		in, out := &in.StatusController, &out.StatusController
+		*out = new(StatusControllerConfig)
+		(*in).DeepCopyInto(*out)
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KubeFedConfigSpec.
@@ -516,9 +521,34 @@ func (in *LocalSecretReference) DeepCopy() *LocalSecretReference {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *StatusControllerConfig) DeepCopyInto(out *StatusControllerConfig) {
+	*out = *in
+	if in.MaxConcurrentReconciles != nil {
+		in, out := &in.MaxConcurrentReconciles, &out.MaxConcurrentReconciles
+		*out = new(int64)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StatusControllerConfig.
+func (in *StatusControllerConfig) DeepCopy() *StatusControllerConfig {
+	if in == nil {
+		return nil
+	}
+	out := new(StatusControllerConfig)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *SyncControllerConfig) DeepCopyInto(out *SyncControllerConfig) {
 	*out = *in
+	if in.MaxConcurrentReconciles != nil {
+		in, out := &in.MaxConcurrentReconciles, &out.MaxConcurrentReconciles
+		*out = new(int64)
+		**out = **in
+	}
 	if in.AdoptResources != nil {
 		in, out := &in.AdoptResources, &out.AdoptResources
 		*out = new(ResourceAdoption)
diff --git a/pkg/controller/federatedtypeconfig/controller.go b/pkg/controller/federatedtypeconfig/controller.go
index 0e6cade6..7d6cc5f5 100644
--- a/pkg/controller/federatedtypeconfig/controller.go
+++ b/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
diff --git a/pkg/controller/kubefedcluster/controller.go b/pkg/controller/kubefedcluster/controller.go
index 03e5c395..1f581dcc 100644
--- a/pkg/controller/kubefedcluster/controller.go
+++ b/pkg/controller/kubefedcluster/controller.go
@@ -185,7 +185,6 @@ func (cc *ClusterController) addToClusterSet(obj *fedv1b1.KubeFedCluster) {
 	restClient, err := NewClusterClientSet(obj, cc.client, cc.fedNamespace, cc.clusterHealthCheckConfig.Timeout)
 	if err != nil || restClient == nil {
 		cc.RecordError(obj, "MalformedClusterConfig", errors.Wrap(err, "The configuration for this cluster may be malformed"))
-		klog.Errorf("The configuration for cluster %s may be malformed", obj.Name)
 		return
 	}
 	cc.clusterDataMap[obj.Name] = &ClusterData{clusterKubeClient: restClient, cachedObj: obj.DeepCopy()}
@@ -246,7 +245,6 @@ func (cc *ClusterController) updateIndividualClusterStatus(cluster *fedv1b1.Kube
 	currentClusterStatus, err := clusterClient.GetClusterHealthStatus()
 	if err != nil {
 		cc.RecordError(cluster, "RetrievingClusterHealthFailed", errors.Wrap(err, "Failed to retrieve health of the cluster"))
-		klog.Errorf("Failed to retrieve health of the cluster %s: %v", cluster.Name, err)
 	}
 
 	currentClusterStatus = thresholdAdjustedClusterStatus(currentClusterStatus, storedData, cc.clusterHealthCheckConfig)
diff --git a/pkg/controller/schedulingmanager/controller.go b/pkg/controller/schedulingmanager/controller.go
index 4c8881ae..afab17b2 100644
--- a/pkg/controller/schedulingmanager/controller.go
+++ b/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(
diff --git a/pkg/controller/schedulingpreference/controller.go b/pkg/controller/schedulingpreference/controller.go
index 78dfce1e..cb119c33 100644
--- a/pkg/controller/schedulingpreference/controller.go
+++ b/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{
diff --git a/pkg/controller/status/controller.go b/pkg/controller/status/controller.go
index a38252b7..ada8dcd2 100644
--- a/pkg/controller/status/controller.go
+++ b/pkg/controller/status/controller.go
@@ -123,8 +123,11 @@ 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,
+		},
+		MaxConcurrentReconciles: int(controllerConfig.MaxConcurrentStatusReconciles),
 	})
 
 	// Build deliverer for triggering cluster reconciliations.
diff --git a/pkg/controller/sync/controller.go b/pkg/controller/sync/controller.go
index f864d6e6..1d5fd923 100644
--- a/pkg/controller/sync/controller.go
+++ b/pkg/controller/sync/controller.go
@@ -132,8 +132,11 @@ 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,
+		},
+		MaxConcurrentReconciles: int(controllerConfig.MaxConcurrentSyncReconciles),
 	})
 
 	// Build deliverer for triggering cluster reconciliations.
@@ -281,7 +284,6 @@ func (s *KubeFedSyncController) reconcile(qualifiedName util.QualifiedName) util
 	err = s.ensureFinalizer(fedResource)
 	if err != nil {
 		fedResource.RecordError("EnsureFinalizerError", errors.Wrap(err, "Failed to ensure finalizer"))
-		runtime.HandleError(errors.Wrapf(err, "failed to ensure finalizer"))
 		return util.StatusError
 	}
 
@@ -298,14 +300,12 @@ func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) ut
 	clusters, err := s.informer.GetClusters()
 	if err != nil {
 		fedResource.RecordError(string(status.ClusterRetrievalFailed), errors.Wrap(err, "Failed to retrieve list of clusters"))
-		runtime.HandleError(errors.Wrapf(err, "failed to retrieve list of clusters"))
 		return s.setFederatedStatus(fedResource, status.ClusterRetrievalFailed, nil, nil, enableRawResourceStatusCollection)
 	}
 
 	selectedClusterNames, err := fedResource.ComputePlacement(clusters)
 	if err != nil {
 		fedResource.RecordError(string(status.ComputePlacementFailed), errors.Wrap(err, "Failed to compute placement"))
-		runtime.HandleError(errors.Wrapf(err, "failed to compute placement"))
 		return s.setFederatedStatus(fedResource, status.ComputePlacementFailed, nil, nil, enableRawResourceStatusCollection)
 	}
 
@@ -377,8 +377,8 @@ func (s *KubeFedSyncController) syncToClusters(fedResource FederatedResource) ut
 	_, timeoutErr := dispatcher.Wait()
 	if timeoutErr != nil {
 		fedResource.RecordError("OperationTimeoutError", timeoutErr)
-		runtime.HandleError(errors.Wrapf(timeoutErr, "operation timeout"))
 	}
+
 	// Write updated versions to the API.
 	updatedVersionMap := dispatcher.VersionMap()
 	err = fedResource.UpdateVersions(selectedClusterNames.List(), updatedVersionMap)
diff --git a/pkg/controller/util/controllerconfig.go b/pkg/controller/util/controllerconfig.go
index af09f469..17317119 100644
--- a/pkg/controller/util/controllerconfig.go
+++ b/pkg/controller/util/controllerconfig.go
@@ -68,12 +68,14 @@ type ClusterHealthCheckConfig struct {
 // controllers.
 type ControllerConfig struct {
 	KubeFedNamespaces
-	KubeConfig                  *restclient.Config
-	ClusterAvailableDelay       time.Duration
-	ClusterUnavailableDelay     time.Duration
-	MinimizeLatency             bool
-	SkipAdoptingResources       bool
-	RawResourceStatusCollection bool
+	KubeConfig                    *restclient.Config
+	ClusterAvailableDelay         time.Duration
+	ClusterUnavailableDelay       time.Duration
+	MinimizeLatency               bool
+	MaxConcurrentSyncReconciles   int64
+	MaxConcurrentStatusReconciles int64
+	SkipAdoptingResources         bool
+	RawResourceStatusCollection   bool
 }
 
 func (c *ControllerConfig) LimitedScope() bool {
diff --git a/pkg/controller/util/worker.go b/pkg/controller/util/worker.go
index d38ddc92..e972c931 100644
--- a/pkg/controller/util/worker.go
+++ b/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() {


Test patch:
diff --git a/pkg/apis/core/v1beta1/defaults/defaults_test.go b/pkg/apis/core/v1beta1/defaults/defaults_test.go
index 8456395d..e093be7b 100644
--- a/pkg/apis/core/v1beta1/defaults/defaults_test.go
+++ b/pkg/apis/core/v1beta1/defaults/defaults_test.go
@@ -120,11 +120,26 @@ func TestSetDefaultKubeFedConfig(t *testing.T) {
 	successCases["spec.clusterHealthCheck.timeout is preserved"] = KubeFedConfigComparison{timeoutKFC, modifiedTimeoutKFC}
 
 	// SyncController
+	syncControllerMaxConcurrentReconcilesKFC := defaultKubeFedConfig()
+	syncControllerMaxConcurrentReconciles := int64(DefaultSyncControllerMaxConcurrentReconciles + 3)
+	syncControllerMaxConcurrentReconcilesKFC.Spec.SyncController.MaxConcurrentReconciles = &syncControllerMaxConcurrentReconciles
+	modifiedSyncControllerMaxConcurrentReconcilesKFC := syncControllerMaxConcurrentReconcilesKFC.DeepCopyObject().(*v1beta1.KubeFedConfig)
+	SetDefaultKubeFedConfig(modifiedSyncControllerMaxConcurrentReconcilesKFC)
+	successCases["spec.syncController.maxConcurrentReconciles is preserved"] = KubeFedConfigComparison{syncControllerMaxConcurrentReconcilesKFC, modifiedSyncControllerMaxConcurrentReconcilesKFC}
+
 	adoptResourcesKFC := defaultKubeFedConfig()
 	*adoptResourcesKFC.Spec.SyncController.AdoptResources = v1beta1.AdoptResourcesDisabled
 	modifiedAdoptResourcesKFC := adoptResourcesKFC.DeepCopyObject().(*v1beta1.KubeFedConfig)
 	SetDefaultKubeFedConfig(modifiedAdoptResourcesKFC)
-	successCases["spec.leaderElect.adoptResources is preserved"] = KubeFedConfigComparison{adoptResourcesKFC, modifiedAdoptResourcesKFC}
+	successCases["spec.syncController.adoptResources is preserved"] = KubeFedConfigComparison{adoptResourcesKFC, modifiedAdoptResourcesKFC}
+
+	// StatusController
+	statusControllerMaxConcurrentReconcilesKFC := defaultKubeFedConfig()
+	statusControllerMaxConcurrentReconciles := int64(DefaultStatusControllerMaxConcurrentReconciles + 3)
+	statusControllerMaxConcurrentReconcilesKFC.Spec.StatusController.MaxConcurrentReconciles = &statusControllerMaxConcurrentReconciles
+	modifiedStatusControllerMaxConcurrentReconcilesKFC := statusControllerMaxConcurrentReconcilesKFC.DeepCopyObject().(*v1beta1.KubeFedConfig)
+	SetDefaultKubeFedConfig(modifiedStatusControllerMaxConcurrentReconcilesKFC)
+	successCases["spec.statusController.maxConcurrentReconciles is preserved"] = KubeFedConfigComparison{statusControllerMaxConcurrentReconcilesKFC, modifiedStatusControllerMaxConcurrentReconcilesKFC}
 
 	for k, v := range successCases {
 		if !reflect.DeepEqual(v.original, v.modified) {
diff --git a/pkg/apis/core/v1beta1/validation/validation_test.go b/pkg/apis/core/v1beta1/validation/validation_test.go
index a6e6a350..23fa73eb 100644
--- a/pkg/apis/core/v1beta1/validation/validation_test.go
+++ b/pkg/apis/core/v1beta1/validation/validation_test.go
@@ -844,6 +844,14 @@ func TestValidateKubeFedConfig(t *testing.T) {
 	invalidSyncControllerNil.Spec.SyncController = nil
 	errorCases["spec.syncController: Required value"] = invalidSyncControllerNil
 
+	invalidSyncControllerMaxConcurrentReconcilesNil := testcommon.ValidKubeFedConfig()
+	invalidSyncControllerMaxConcurrentReconcilesNil.Spec.SyncController.MaxConcurrentReconciles = nil
+	errorCases["spec.syncController.maxConcurrentReconciles: Required value"] = invalidSyncControllerMaxConcurrentReconcilesNil
+
+	invalidSyncControllerMaxConcurrentReconcilesGreaterThan0 := testcommon.ValidKubeFedConfig()
+	invalidSyncControllerMaxConcurrentReconcilesGreaterThan0.Spec.SyncController.MaxConcurrentReconciles = zeroIntPtr
+	errorCases["spec.syncController.maxConcurrentReconciles: Invalid value"] = invalidSyncControllerMaxConcurrentReconcilesGreaterThan0
+
 	invalidAdoptResourcesNil := testcommon.ValidKubeFedConfig()
 	invalidAdoptResourcesNil.Spec.SyncController.AdoptResources = nil
 	errorCases["spec.syncController.adoptResources: Required value"] = invalidAdoptResourcesNil
@@ -853,6 +861,18 @@ func TestValidateKubeFedConfig(t *testing.T) {
 	invalidAdoptResources.Spec.SyncController.AdoptResources = &invalidAdoptResourcesValue
 	errorCases["spec.syncController.adoptResources: Unsupported value"] = invalidAdoptResources
 
+	invalidStatusControllerNil := testcommon.ValidKubeFedConfig()
+	invalidStatusControllerNil.Spec.StatusController = nil
+	errorCases["spec.statusController: Required value"] = invalidStatusControllerNil
+
+	invalidStatusControllerMaxConcurrentReconcilesNil := testcommon.ValidKubeFedConfig()
+	invalidStatusControllerMaxConcurrentReconcilesNil.Spec.StatusController.MaxConcurrentReconciles = nil
+	errorCases["spec.statusController.maxConcurrentReconciles: Required value"] = invalidStatusControllerMaxConcurrentReconcilesNil
+
+	invalidStatusControllerMaxConcurrentReconcilesGreaterThan0 := testcommon.ValidKubeFedConfig()
+	invalidStatusControllerMaxConcurrentReconcilesGreaterThan0.Spec.StatusController.MaxConcurrentReconciles = zeroIntPtr
+	errorCases["spec.statusController.maxConcurrentReconciles: Invalid value"] = invalidStatusControllerMaxConcurrentReconcilesGreaterThan0
+
 	for k, v := range errorCases {
 		errs := ValidateKubeFedConfig(v, testcommon.ValidKubeFedConfig())
 		if len(errs) == 0 {
diff --git a/pkg/controller/testdata/fixtures/crds.yaml b/pkg/controller/testdata/fixtures/crds.yaml
index 99321a4d..2db7f5f2 100644
--- a/pkg/controller/testdata/fixtures/crds.yaml
+++ b/pkg/controller/testdata/fixtures/crds.yaml
@@ -774,12 +774,25 @@ spec:
                   `Namespaced` or `Cluster`. `Namespaced` indicates that the KubeFed
                   namespace will be the only target of the control plane.
                 type: string
+              statusController:
+                properties:
+                  maxConcurrentReconciles:
+                    description: The maximum number of concurrent Reconciles of status
+                      controller which can be run. Defaults to 1.
+                    format: int64
+                    type: integer
+                type: object
               syncController:
                 properties:
                   adoptResources:
                     description: Whether to adopt pre-existing resources in member
                       clusters. Defaults to "Enabled".
                     type: string
+                  maxConcurrentReconciles:
+                    description: The maximum number of concurrent Reconciles of sync
+                      controller which can be run. Defaults to 1.
+                    format: int64
+                    type: integer
                 type: object
             required:
             - scope
diff --git a/pkg/controller/util/worker_test.go b/pkg/controller/util/worker_test.go
index 1d2c1138..a0fe84ca 100644
--- a/pkg/controller/util/worker_test.go
+++ b/pkg/controller/util/worker_test.go
@@ -61,7 +61,7 @@ func TestDeduplicate(t *testing.T) {
 			})
 			return StatusAllOK
 		},
-		WorkerTiming{},
+		WorkerOptions{},
 	)
 
 	// run worker
diff --git a/test/common/bindata.go b/test/common/bindata.go
index 51e63f26..ef4385ee 100644
--- a/test/common/bindata.go
+++ b/test/common/bindata.go
@@ -400,7 +400,10 @@ spec:
     successThreshold: 1
     timeout: 3s
   syncController:
+    maxConcurrentReconciles: 1
     adoptResources: Enabled
+  statusController:
+    maxConcurrentReconciles: 1
 `)
 
 func configKubefedconfigYamlBytes() ([]byte, error) {