Add configurable concurrent reconciliation to controllers

The controller-runtime defaults MaxConcurrentReconciles to 1, which creates
a bottleneck when multiple BMC operations need to be processed simultaneously.
This is particularly noticeable during cluster provisioning when many nodes
are being set up in parallel.

This change adds a CLI flag (--max-concurrent-reconciles) and corresponding
environment variable (RUFIO_MAX_CONCURRENT_RECONCILES) to configure the
number of concurrent reconciliations across all controllers.

Signed-off-by: Rahul Ganesh <rahulgab@amazon.com>

Author: rahulbabu95

controller/job.go

@@ -26,6 +26,7 @@ import (
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 
 	"github.com/tinkerbell/rufio/api/v1alpha1"
@@ -221,7 +222,7 @@ func (r *JobReconciler) patchStatus(ctx context.Context, job *v1alpha1.Job, patc
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *JobReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+func (r *JobReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts ctrlcontroller.Options) error {
 	if err := mgr.GetFieldIndexer().IndexField(
 		ctx,
 		&v1alpha1.Task{},
@@ -233,6 +234,7 @@ func (r *JobReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager)
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.Job{}).
+		WithOptions(opts).
 		Watches(
 			&v1alpha1.Task{},
 			handler.EnqueueRequestForOwner(

controller/machine.go

@@ -29,6 +29,8 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/record"
+	ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
+
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -210,8 +212,9 @@ func retrieveHMACSecrets(ctx context.Context, c client.Client, hmacSecrets v1alp
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *MachineReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *MachineReconciler) SetupWithManager(mgr ctrl.Manager, opts ctrlcontroller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.Machine{}).
+		WithOptions(opts).
 		Complete(r)
 }

controller/task.go

@@ -25,6 +25,7 @@ import (
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
 
 	"github.com/tinkerbell/rufio/api/v1alpha1"
 )
@@ -274,8 +275,9 @@ func (r *TaskReconciler) patchStatus(ctx context.Context, task *v1alpha1.Task, p
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *TaskReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *TaskReconciler) SetupWithManager(mgr ctrl.Manager, opts ctrlcontroller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.Task{}).
+		WithOptions(opts).
 		Complete(r)
 }

main.go

@@ -33,16 +33,16 @@ import (
 	"github.com/peterbourgon/ff/v3"
 	"github.com/peterbourgon/ff/v3/ffcli"
 	"github.com/rs/zerolog"
+	"github.com/tinkerbell/rufio/api/v1alpha1"
+	"github.com/tinkerbell/rufio/controller"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
-
-	"github.com/tinkerbell/rufio/api/v1alpha1"
-	"github.com/tinkerbell/rufio/controller"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -86,6 +86,7 @@ func main() {
 	var kubeconfig string
 	var kubeNamespace string
 	var bmcConnectTimeout time.Duration
+	var maxConcurrentReconciles int
 	fs := flag.NewFlagSet(appName, flag.ExitOnError)
 	fs.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	fs.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
@@ -96,6 +97,7 @@ func main() {
 	fs.StringVar(&kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig file.")
 	fs.StringVar(&kubeNamespace, "kube-namespace", "", "Namespace that the controller watches to reconcile objects.")
 	fs.DurationVar(&bmcConnectTimeout, "bmc-connect-timeout", 60*time.Second, "Timeout for establishing a connection to BMCs.")
+	fs.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 1, "Maximum number of concurrent reconciles per controller.")
 	cli := &ffcli.Command{
 		Name:    appName,
 		FlagSet: fs,
@@ -144,7 +146,7 @@ func main() {
 	bmcClientFactory := controller.NewClientFunc(bmcConnectTimeout)
 
 	// Setup controller reconcilers
-	setupReconcilers(ctx, mgr, bmcClientFactory)
+	setupReconcilers(ctx, mgr, bmcClientFactory, maxConcurrentReconciles)
 
 	//+kubebuilder:scaffold:builder
 
@@ -175,20 +177,20 @@ func newClientConfig(kubeAPIServer, kubeconfig string) clientcmd.ClientConfig {
 }
 
 // setupReconcilers initializes the controllers with the Manager.
-func setupReconcilers(ctx context.Context, mgr ctrl.Manager, bmcClientFactory controller.ClientFunc) {
+func setupReconcilers(ctx context.Context, mgr ctrl.Manager, bmcClientFactory controller.ClientFunc, maxConcurrentReconciles int) {
 	err := (controller.NewMachineReconciler(
 		mgr.GetClient(),
 		mgr.GetEventRecorderFor("machine-controller"),
 		bmcClientFactory,
-	)).SetupWithManager(mgr)
+	)).SetupWithManager(mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})
 	if err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Machine")
 		os.Exit(1)
 	}
 
 	err = (controller.NewJobReconciler(
 		mgr.GetClient(),
-	)).SetupWithManager(ctx, mgr)
+	)).SetupWithManager(ctx, mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})
 	if err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Job")
 		os.Exit(1)
@@ -197,7 +199,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager, bmcClientFactory co
 	err = (controller.NewTaskReconciler(
 		mgr.GetClient(),
 		bmcClientFactory,
-	)).SetupWithManager(mgr)
+	)).SetupWithManager(mgr, ctrlcontroller.Options{MaxConcurrentReconciles: maxConcurrentReconciles})
 	if err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Task")
 		os.Exit(1)