feat: add configurable max-concurrent-reconciles flag

Add --max-concurrent-reconciles CLI flag (default 1) to control the
number of concurrent reconciles per controller. This is exposed via
Helm values as features.maxConcurrentReconciles.

Applied to PulsarConnection and RoleBinding controllers. Customers
with many RoleBindings (e.g. 85+) can increase this to speed up
reconciliation significantly.

Author: mattisonchao

charts/pulsar-resources-operator/templates/deployment.yaml

@@ -62,6 +62,7 @@ spec:
         - --leader-elect
         - --resync-period={{ .Values.features.resyncPeriod | default 10 }}
         - --retry-count={{ .Values.features.retryCount | default 5 }}
+        - --max-concurrent-reconciles={{ .Values.features.maxConcurrentReconciles | default 1 }}
         command:
         - /manager
         env:

charts/pulsar-resources-operator/values.yaml

@@ -47,6 +47,8 @@ features:
   resyncPeriod: 10
   # The number of retries in case of error.
   retryCount: 5
+  # The maximum number of concurrent reconciles per controller.
+  maxConcurrentReconciles: 1
 
 # -- It will override the value of label `app.kubernetes.io/name` on pod
 nameOverride: ""

controllers/rolebinding_controller.go

@@ -29,6 +29,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -292,8 +293,9 @@ func (r *RoleBindingReconciler) findResourcesForConnection(ctx context.Context,
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *RoleBindingReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *RoleBindingReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(options).
 		For(&resourcev1alpha1.RoleBinding{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		Watches(&resourcev1alpha1.StreamNativeCloudConnection{},
 			handler.EnqueueRequestsFromMapFunc(r.findResourcesForConnection)).

main.go

@@ -64,13 +64,15 @@ func main() {
 	var probeAddr string
 	var resyncPeriod int
 	var retryCount int
+	var maxConcurrentReconciles int
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8443", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
 	flag.IntVar(&resyncPeriod, "resync-period", 10, "resyncPeriod is the base frequency the informers are resynced.")
 	flag.IntVar(&retryCount, "retry-count", 5, "The number of retries in case of error.")
+	flag.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 1, "The maximum number of concurrent reconciles per controller.")
 	opts := k8szap.Options{}
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
@@ -116,15 +118,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	// TODO get MaxConcurrentReconciles from cmd params
 	if err = (&controllers.PulsarConnectionReconciler{
 		Client:             mgr.GetClient(),
 		Scheme:             mgr.GetScheme(),
 		Log:                ctrl.Log.WithName("controllers").WithName("PulsarConnection"),
 		Recorder:           mgr.GetEventRecorderFor("pulsarconnection-controller"),
 		PulsarAdminCreator: admin.NewPulsarAdmin,
 		Retryer:            utils.NewReconcileRetryer(retryCount, utils.NewEventSource(ctrl.Log.WithName("eventSource"))),
-	}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: 1}); err != nil {
+	}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "PulsarConnection")
 		os.Exit(1)
 	}
@@ -206,7 +207,7 @@ func main() {
 		Client:            mgr.GetClient(),
 		Scheme:            mgr.GetScheme(),
 		ConnectionManager: connectionManager,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RoleBinding")
 		os.Exit(1)
 	}