add conccurrency flags

Author: Daniecki, Jozef

cmd/connect-controller/main.go

@@ -61,6 +61,9 @@ func main() {
 	var connectionProbeTimeout time.Duration
 	var profilerAddress string
 	var enableContentionProfiling bool
+	var concurrency int
+	var kubeApiQPS float64
+	var kubeApiBurst int
 	flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
 		"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
@@ -81,6 +84,10 @@ func main() {
 	flag.DurationVar(&connectionProbeTimeout, "connection-probe-timeout", 5*time.Minute, "The timeout duration for connection probes.")
 	flag.StringVar(&profilerAddress, "profiler-address", "", "Bind address to expose the pprof profiler (e.g. localhost:6060)")
 	flag.BoolVar(&enableContentionProfiling, "contention-profiling", false, "Enable block profiling")
+	flag.IntVar(&concurrency, "concurrency", 1, "Maximum number of concurrent workers processing ClusterConnect resources")
+	flag.Float64Var(&kubeApiQPS, "kube-api-qps", 20, "Maximum queries per second from the controller client to the Kubernetes API server.")
+	flag.IntVar(&kubeApiBurst, "kube-api-burst", 30, "Maximum number of queries that should be allowed in one burst from the controller client to the Kubernetes API server.")
+
 	opts := zap.Options{
 		Development: true,
 	}
@@ -181,8 +188,11 @@ func main() {
 			config.GetCertificate = metricsCertWatcher.GetCertificate
 		})
 	}
+	restConfig := ctrl.GetConfigOrDie()
+	restConfig.QPS = float32(kubeApiQPS)
+	restConfig.Burst = kubeApiBurst
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	mgr, err := ctrl.NewManager(restConfig, ctrl.Options{
 		Scheme:                 scheme,
 		Metrics:                metricsServerOptions,
 		WebhookServer:          webhookServer,
@@ -220,7 +230,7 @@ func main() {
 	if err = (&controller.ClusterConnectReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
-	}).SetupWithManager(ctx, mgr, connectionProbeTimeout); err != nil {
+	}).SetupWithManager(ctx, mgr, connectionProbeTimeout, concurrency); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ClusterConnect")
 		os.Exit(1)
 	}

internal/controller/clusterconnect_controller.go

@@ -25,6 +25,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/patch"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	cutil "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -126,7 +127,7 @@ func clusterRefIdxFunc(o client.Object) []string {
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *ClusterConnectReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, connectionTimeout time.Duration) error {
+func (r *ClusterConnectReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, connectionTimeout time.Duration, concurrency int) error {
 	if r.Client == nil {
 		return errors.New("Client must not be nil")
 	}
@@ -136,6 +137,7 @@ func (r *ClusterConnectReconciler) SetupWithManager(ctx context.Context, mgr ctr
 	c, err := ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.ClusterConnect{}).
 		Named("cluster/clusterconnect").
+		WithOptions(controller.Options{MaxConcurrentReconciles: concurrency}).
 		Build(r)
 
 	if err != nil {

internal/controller/suite_test.go

@@ -105,7 +105,7 @@ var _ = BeforeSuite(func() {
 	err = (&ClusterConnectReconciler{
 		Client: k8sManager.GetClient(),
 		Scheme: k8sManager.GetScheme(),
-	}).SetupWithManager(ctx, k8sManager, 1*time.Second)
+	}).SetupWithManager(ctx, k8sManager, 1*time.Second, 1)
 	Expect(err).ToNot(HaveOccurred())
 
 	go func() {