set concurrency for controllers

Signed-off-by: roc <[email protected]>

Author: imroc

cmd/app/setup_manager.go

@@ -72,7 +72,7 @@ func SetupManager(mgr ctrl.Manager) {
 		Client:   mgr.GetClient(),
 		Scheme:   mgr.GetScheme(),
 		Recorder: mgr.GetEventRecorderFor("clbportpool-controller"),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, workers); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CLBPortPool")
 		os.Exit(1)
 	}
@@ -86,7 +86,7 @@ func SetupManager(mgr ctrl.Manager) {
 		Client:   mgr.GetClient(),
 		Scheme:   mgr.GetScheme(),
 		Recorder: mgr.GetEventRecorderFor("clbpodbinding-controller"),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, workers); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CLBPodBinding")
 		os.Exit(1)
 	}
@@ -100,7 +100,7 @@ func SetupManager(mgr ctrl.Manager) {
 		Client:   mgr.GetClient(),
 		Scheme:   mgr.GetScheme(),
 		Recorder: mgr.GetEventRecorderFor("pod-controller"),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, workers); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Pod")
 		os.Exit(1)
 	}

internal/controller/clbpodbinding_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	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/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
@@ -515,13 +516,16 @@ func (r *CLBPodBindingReconciler) cleanup(ctx context.Context, pb *networkingv1a
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *CLBPodBindingReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *CLBPodBindingReconciler) SetupWithManager(mgr ctrl.Manager, workers int) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&networkingv1alpha1.CLBPodBinding{}).
 		Watches(
 			&corev1.Pod{},
 			handler.EnqueueRequestsFromMapFunc(r.findObjectsForPod),
 		).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: workers,
+		}).
 		Named("clbpodbinding").
 		Complete(r)
 }

internal/controller/clbportpool_controller.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	networkingv1alpha1 "github.com/imroc/tke-extend-network-controller/api/v1alpha1"
 	"github.com/imroc/tke-extend-network-controller/internal/portpool"
@@ -272,9 +273,12 @@ func (r *CLBPortPoolReconciler) sync(ctx context.Context, pool *networkingv1alph
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *CLBPortPoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *CLBPortPoolReconciler) SetupWithManager(mgr ctrl.Manager, workers int) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&networkingv1alpha1.CLBPortPool{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: workers,
+		}).
 		Named("clbportpool").
 		Complete(r)
 }

internal/controller/pod_controller.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 	"time"
 
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	networkingv1alpha1 "github.com/imroc/tke-extend-network-controller/api/v1alpha1"
@@ -161,13 +162,16 @@ func (r *PodReconciler) sync(ctx context.Context, pod *corev1.Pod) (result ctrl.
 var podEventSource = make(chan event.TypedGenericEvent[client.Object])
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager, workers int) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		Watches(
 			&corev1.Pod{},
 			handler.EnqueueRequestsFromMapFunc(r.findObjectsForPod),
 		).
 		WatchesRawSource(source.Channel(podEventSource, &handler.EnqueueRequestForObject{})).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: workers,
+		}).
 		Named("pod").
 		Complete(r)
 }