Concurrently reconcile CloudStackMachine resources

controllers/cloudstackmachine_controller.go

@@ -32,6 +32,7 @@ import (
 	"sigs.k8s.io/cluster-api/util"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"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/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -233,7 +234,6 @@ func (r *CloudStackMachineReconciliationRunner) GetOrCreateVMInstance() (retRes
 
 	userData := processCustomMetadata(data, r)
 	err := r.CSUser.GetOrCreateVMInstance(r.ReconciliationSubject, r.CAPIMachine, r.CSCluster, r.FailureDomain, r.AffinityGroup, userData)
-
 	if err != nil {
 		r.Recorder.Eventf(r.ReconciliationSubject, "Warning", "Creating", CSMachineCreationFailed, err.Error())
 	}
@@ -345,8 +345,9 @@ func (r *CloudStackMachineReconciliationRunner) ReconcileDelete() (retRes ctrl.R
 }
 
 // SetupWithManager registers the machine reconciler to the CAPI controller manager.
-func (reconciler *CloudStackMachineReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (reconciler *CloudStackMachineReconciler) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	controller, err := ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrav1.CloudStackMachine{}).
 		WithEventFilter(
 			predicate.Funcs{

main.go

@@ -37,6 +37,7 @@ import (
 	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/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -73,6 +74,8 @@ type managerOpts struct {
 	WatchingNamespace    string
 	WatchFilterValue     string
 	CertDir              string
+
+	MachineConcurrency int
 }
 
 func setFlags() *managerOpts {
@@ -117,6 +120,14 @@ func setFlags() *managerOpts {
 		"webhook-cert-dir",
 		"/tmp/k8s-webhook-server/serving-certs/",
 		"Specify the directory where webhooks will get tls certificates.")
+
+	flag.IntVar(
+		&opts.MachineConcurrency,
+		"machine-concurrency",
+		1,
+		"Concurrent reconciles for the CloudStackMachine resource",
+	)
+
 	return opts
 }
 
@@ -152,10 +163,11 @@ func main() {
 		K8sClient:  mgr.GetClient(),
 		BaseLogger: ctrl.Log.WithName("controllers"),
 		Recorder:   mgr.GetEventRecorderFor("capc-controller-manager"),
-		Scheme:     mgr.GetScheme()}
+		Scheme:     mgr.GetScheme(),
+	}
 
 	ctx := ctrl.SetupSignalHandler()
-	setupReconcilers(ctx, base, mgr)
+	setupReconcilers(ctx, base, *opts, mgr)
 	infrav1b3.K8sClient = base.K8sClient
 
 	// +kubebuilder:scaffold:builder
@@ -191,12 +203,12 @@ func main() {
 	}
 }
 
-func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, mgr manager.Manager) {
+func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, opts managerOpts, mgr manager.Manager) {
 	if err := (&controllers.CloudStackClusterReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackCluster")
 		os.Exit(1)
 	}
-	if err := (&controllers.CloudStackMachineReconciler{ReconcilerBase: base}).SetupWithManager(mgr); err != nil {
+	if err := (&controllers.CloudStackMachineReconciler{ReconcilerBase: base}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: opts.MachineConcurrency}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackMachine")
 		os.Exit(1)
 	}