Merge pull request #624 from andybavier/concurrency

Enable concurrency in controllers

Author: furkatgofurov7

bootstrap/internal/controllers/rke2config_controller.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/utils/ptr"
 	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/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	kubeyaml "sigs.k8s.io/yaml"
@@ -206,13 +207,16 @@ func (r *RKE2ConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *RKE2ConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *RKE2ConfigReconciler) SetupWithManager(mgr ctrl.Manager, concurrency int) error {
 	if r.RKE2InitLock == nil {
 		r.RKE2InitLock = locking.NewControlPlaneInitMutex(mgr.GetClient())
 	}
 
 	builder := ctrl.NewControllerManagedBy(mgr).
 		For(&bootstrapv1.RKE2Config{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
 		Watches(
 			&clusterv1.Machine{},
 			handler.EnqueueRequestsFromMapFunc(r.MachineToBootstrapMapFunc),

bootstrap/main.go

@@ -198,7 +198,7 @@ func main() {
 	setupReconcilers(mgr)
 	setupWebhooks(mgr)
 
-	setupLog.Info("Starting manager", "version", version.Get().String())
+	setupLog.Info("Starting manager", "version", version.Get().String(), "concurrency", concurrencyNumber)
 
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")
@@ -222,7 +222,7 @@ func setupReconcilers(mgr ctrl.Manager) {
 	if err := (&controllers.RKE2ConfigReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrencyNumber); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Rke2Config")
 		os.Exit(1)
 	}

controlplane/internal/controllers/rke2controlplane_controller.go

@@ -238,10 +238,13 @@ func patchRKE2ControlPlane(ctx context.Context, patchHelper *patch.Helper, rcp *
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *RKE2ControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, clientQPS float32, clientBurst int) error {
+func (r *RKE2ControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, clientQPS float32, clientBurst, concurrency int) error {
 	c, err := ctrl.NewControllerManagedBy(mgr).
 		For(&controlplanev1.RKE2ControlPlane{}).
 		Owns(&clusterv1.Machine{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
 		Build(r)
 	if err != nil {
 		return errors.Wrap(err, "failed setting up with a controller manager")

controlplane/main.go

@@ -203,7 +203,7 @@ func main() {
 	setupWebhooks(mgr)
 	setupReconcilers(ctx, mgr)
 
-	setupLog.Info("Starting manager", "version", version.Get().String())
+	setupLog.Info("Starting manager", "version", version.Get().String(), "concurrency", concurrencyNumber)
 
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")
@@ -240,7 +240,7 @@ func setupReconcilers(ctx context.Context, mgr ctrl.Manager) {
 		Scheme:              mgr.GetScheme(),
 		WatchFilterValue:    watchFilterValue,
 		SecretCachingClient: secretCachingClient,
-	}).SetupWithManager(ctx, mgr, clusterCacheTrackerClientQPS, clusterCacheTrackerClientBurst); err != nil {
+	}).SetupWithManager(ctx, mgr, clusterCacheTrackerClientQPS, clusterCacheTrackerClientBurst, concurrencyNumber); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RKE2ControlPlane")
 		os.Exit(1)
 	}