Concurrency level configuration (#1269)

Signed-off-by: Alexey Makhov <[email protected]>
Signed-off-by: makhov <[email protected]>

Author: makhov

cmd/main.go

@@ -39,6 +39,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	capictrl "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
@@ -90,12 +91,15 @@ func init() {
 }
 
 func main() {
-	var metricsAddr string // deprecated, use capi's diagnostics-address instead
-	var enableLeaderElection bool
-	var secureMetrics bool // deprecated, use capi's insecure-diagnostics instead
-	var enableHTTP2 bool   // deprecated
-	var probeAddr string
-	var enabledController string
+	var (
+		metricsAddr          string // deprecated, use capi's diagnostics-address instead
+		enableLeaderElection bool
+		secureMetrics        bool // deprecated, use capi's insecure-diagnostics instead
+		enableHTTP2          bool // deprecated
+		probeAddr            string
+		enabledController    string
+		concurrency          int
+	)
 
 	pflag.CommandLine.StringVar(&metricsAddr, "metrics-bind-address", ":8443", "[Deprecated, use --diagnostics-address instead] The address the metric endpoint binds to. "+
 		"Use :8080 for http and :8443 for https. Setting to 0 will disable the metrics endpoint.")
@@ -108,6 +112,7 @@ func main() {
 	pflag.CommandLine.BoolVar(&enableHTTP2, "enable-http2", false,
 		"[Deprecated] If set, HTTP/2 will be enabled for the metrics and webhook servers")
 	pflag.CommandLine.StringVar(&featureGates, "feature-gates", "", "feature gates to enable (comma separated list of key=value pairs)")
+	pflag.CommandLine.IntVar(&concurrency, "concurrency", 5, "controller concurrency, default: 5")
 
 	pflag.CommandLine.StringVar(&enabledController, "enable-controller", "", "The controller to enable. Default: all")
 	opts := zap.Options{
@@ -240,6 +245,9 @@ func main() {
 	}
 
 	//+kubebuilder:scaffold:builder
+	ctrlOptions := capictrl.Options{
+		MaxConcurrentReconciles: concurrency,
+	}
 
 	if isControllerEnabled(bootstrapController) && runCAPIControllers {
 		if err = (&bootstrap.Controller{
@@ -248,7 +256,7 @@ func main() {
 			Scheme:              mgr.GetScheme(),
 			ClientSet:           clientSet,
 			RESTConfig:          restConfig,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Bootstrap")
 			os.Exit(1)
 		}
@@ -258,7 +266,7 @@ func main() {
 			Scheme:              mgr.GetScheme(),
 			ClientSet:           clientSet,
 			RESTConfig:          restConfig,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Bootstrap")
 			os.Exit(1)
 		}
@@ -272,7 +280,7 @@ func main() {
 			ClientSet:  clientSet,
 			RESTConfig: restConfig,
 			Recorder:   mgr.GetEventRecorderFor("cluster-reconciler"),
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "K0smotronCluster")
 			os.Exit(1)
 		}
@@ -287,7 +295,7 @@ func main() {
 			Scheme:     mgr.GetScheme(),
 			ClientSet:  clientSet,
 			RESTConfig: restConfig,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "JoinTokenRequest")
 			os.Exit(1)
 		}
@@ -299,7 +307,7 @@ func main() {
 				Scheme:              mgr.GetScheme(),
 				ClientSet:           clientSet,
 				RESTConfig:          restConfig,
-			}).SetupWithManager(mgr); err != nil {
+			}).SetupWithManager(mgr, ctrlOptions); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "K0smotronControlPlane")
 				os.Exit(1)
 			}
@@ -309,7 +317,7 @@ func main() {
 				SecretCachingClient: secretCachingClient,
 				ClientSet:           clientSet,
 				RESTConfig:          restConfig,
-			}).SetupWithManager(mgr); err != nil {
+			}).SetupWithManager(mgr, ctrlOptions); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "K0sController")
 				os.Exit(1)
 			}
@@ -333,15 +341,15 @@ func main() {
 			Scheme:              mgr.GetScheme(),
 			ClientSet:           clientSet,
 			RESTConfig:          restConfig,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "RemoteMachine")
 			os.Exit(1)
 		}
 
 		if err = (&infrastructure.ClusterController{
 			Client: mgr.GetClient(),
 			Scheme: mgr.GetScheme(),
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "RemoteCluster")
 			os.Exit(1)
 		}
@@ -354,7 +362,7 @@ func main() {
 			Client:    mgr.GetClient(),
 			Scheme:    mgr.GetScheme(),
 			ClientSet: clientSet,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManager(mgr, ctrlOptions); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "ProviderID")
 			os.Exit(1)
 		}

internal/controller/bootstrap/controlplane_bootstrap_controller.go

@@ -50,6 +50,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/secret"
 	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/log"
 
 	"github.com/go-logr/logr"
@@ -599,8 +600,10 @@ func createTokenSecret(tokenID, tokenSecret string) *corev1.Secret {
 	}
 }
 
-func (c *ControlPlaneController) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManager sets up the controller with the Manager.
+func (c *ControlPlaneController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&bootstrapv1.K0sControllerConfig{}).
 		Complete(c)
 }

internal/controller/bootstrap/providerid_controller.go

@@ -14,6 +14,7 @@ import (
 	capiutil "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/log"
 
 	k0smoutil "github.com/k0sproject/k0smotron/internal/controller/util"
@@ -91,7 +92,8 @@ func (p *ProviderIDController) Reconcile(ctx context.Context, req ctrl.Request)
 	return ctrl.Result{}, nil
 }
 
-func (p *ProviderIDController) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManager sets up the controller with the Manager.
+func (p *ProviderIDController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	apiResources, err := p.ClientSet.Discovery().ServerResourcesForGroupVersion(clusterv1.GroupVersion.String())
 	if err != nil && !apierrors.IsNotFound(err) {
 		return err
@@ -102,6 +104,7 @@ func (p *ProviderIDController) SetupWithManager(mgr ctrl.Manager) error {
 	}
 
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&clusterv1.Machine{}).
 		Complete(p)
 }

internal/controller/bootstrap/worker_bootstrap_controller.go

@@ -42,6 +42,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/secret"
 	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/log"
 
 	"github.com/go-logr/logr"
@@ -438,8 +439,10 @@ func createInstallCmd(scope *Scope) string {
 	return strings.Join(installCmd, " ")
 }
 
-func (r *Controller) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManager sets up the controller with the Manager.
+func (r *Controller) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&bootstrapv1.K0sWorkerConfig{}).
 		Complete(r)
 }

internal/controller/controlplane/k0s_controlplane_controller.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"strings"
 	"time"
 
@@ -1012,9 +1013,10 @@ func (c *K0sController) createFRPToken(ctx context.Context, cluster *clusterv1.C
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (c *K0sController) SetupWithManager(mgr ctrl.Manager) error {
+func (c *K0sController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	// Check if the cluster.x-k8s.io API is available and if not, don't try to watch for Machine objects
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&cpv1beta1.K0sControlPlane{}).
 		Owns(&clusterv1.Machine{}).
 		Complete(c)

internal/controller/controlplane/k0smotron_controlplane_controller.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/cluster-api/controllers/remote"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -697,8 +698,9 @@ func (c *K0smotronController) getKmcScope(ctx context.Context, kcp *cpv1beta1.K0
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (c *K0smotronController) SetupWithManager(mgr ctrl.Manager) error {
+func (c *K0smotronController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&cpv1beta1.K0smotronControlPlane{}).
 		Owns(&kapi.Cluster{}, builder.MatchEveryOwner).
 		Complete(c)

internal/controller/infrastructure/cluster_controller.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	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/log"
 
 	infrastructure "github.com/k0sproject/k0smotron/api/infrastructure/v1beta1"
@@ -62,8 +63,10 @@ func (r *ClusterController) Reconcile(ctx context.Context, req ctrl.Request) (re
 	return ctrl.Result{}, nil
 }
 
-func (r *ClusterController) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManager sets up the controller with the Manager.
+func (r *ClusterController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrastructure.RemoteCluster{}).
 		Complete(r)
 }

internal/controller/infrastructure/remote_machine_controller.go

@@ -36,6 +36,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"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
@@ -415,8 +416,10 @@ func (r *RemoteMachineController) getBootstrapData(ctx context.Context, machine
 	return secret.Data["value"], nil
 }
 
-func (r *RemoteMachineController) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManager sets up the controller with the Manager.
+func (r *RemoteMachineController) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrastructure.RemoteMachine{}).
 		Complete(r)
 }

internal/controller/k0smotron.io/jointokenrequest_controller.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	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/log"
 
@@ -201,8 +202,9 @@ func (r *JoinTokenRequestReconciler) updateStatus(ctx context.Context, jtr km.Jo
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *JoinTokenRequestReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *JoinTokenRequestReconciler) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&km.JoinTokenRequest{}).
 		Complete(r)
 }

internal/controller/k0smotron.io/k0smotroncluster_controller.go

@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/secret"
 	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/log"
 
@@ -345,8 +346,9 @@ func (r *ClusterReconciler) getKmcScope(ctx context.Context, kmc *km.Cluster) (*
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&km.Cluster{}).
 		Owns(&apps.StatefulSet{}).
 		Complete(r)