add cluster concurrent reconciles opts (#2601)

controllers/cluster/controllers/cluster_controller.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/go-logr/logr"
 
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
@@ -74,6 +75,10 @@ type ClusterReconciler struct {
 	recorder record.EventRecorder
 }
 
+type ClusterReconcilerOptions struct {
+	MaxConcurrentReconciles int
+}
+
 //+kubebuilder:rbac:groups=cluster.sealos.io,resources=clusters,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=cluster.sealos.io,resources=clusters/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=cluster.sealos.io,resources=clusters/finalizers,verbs=update
@@ -360,7 +365,7 @@ func (r *ClusterReconciler) updateStatus(ctx context.Context, nn types.Namespace
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager, opts ClusterReconcilerOptions) error {
 	const controllerName = "cluster_controller"
 	r.Logger = ctrl.Log.WithName(controllerName)
 	r.recorder = mgr.GetEventRecorderFor("sealos-cluster-controller")
@@ -368,5 +373,8 @@ func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1.Cluster{}).
 		Watches(&source.Kind{Type: &infrav1.Infra{}}, &handler.EnqueueRequestForObject{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: opts.MaxConcurrentReconciles,
+		}).
 		Complete(r)
 }

controllers/cluster/main.go

@@ -52,11 +52,16 @@ func init() {
 }
 
 func main() {
-	var metricsAddr string
-	var enableLeaderElection bool
-	var probeAddr string
+	var (
+		metricsAddr          string
+		enableLeaderElection bool
+		probeAddr            string
+		concurrent           int
+	)
+
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8082", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8083", "The address the probe endpoint binds to.")
+	flag.IntVar(&concurrent, "concurrent", 5, "The number of concurrent cluster reconciles.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
@@ -95,7 +100,8 @@ func main() {
 	if err = (&controllers.ClusterReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, controllers.ClusterReconcilerOptions{
+		MaxConcurrentReconciles: concurrent}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Cluster")
 		os.Exit(1)
 	}