Merge pull request #151 from jokuniew/concurrency

feat: concurrency

Author: richardcase

bootstrap/controllers/kthreesconfig_controller.go

@@ -41,6 +41,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/predicates"
 	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/reconcile"
 	kubeyaml "sigs.k8s.io/yaml"
 
@@ -535,13 +536,16 @@ func (r *KThreesConfigReconciler) handleClusterNotInitialized(ctx context.Contex
 	return ctrl.Result{}, nil
 }
 
-func (r *KThreesConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *KThreesConfigReconciler) SetupWithManager(mgr ctrl.Manager, concurrency int) error {
 	if r.KThreesInitLock == nil {
 		r.KThreesInitLock = locking.NewControlPlaneInitMutex(mgr.GetClient())
 	}
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&bootstrapv1.KThreesConfig{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
 		WithEventFilter(predicates.ResourceNotPaused(r.Log)).
 		Complete(r)
 }

bootstrap/main.go

@@ -55,6 +55,7 @@ func main() {
 	var metricsAddr string
 	var enableLeaderElection bool
 	var syncPeriod time.Duration
+	var concurrencyNumber int
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
@@ -64,6 +65,9 @@ func main() {
 	flag.DurationVar(&syncPeriod, "sync-period", 10*time.Minute,
 		"The minimum interval at which watched resources are reconciled (e.g. 15m)")
 
+	flag.IntVar(&concurrencyNumber, "concurrency", 1,
+		"Number of core resources to process simultaneously")
+
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
@@ -91,7 +95,7 @@ func main() {
 		Client: mgr.GetClient(),
 		Log:    ctrl.Log.WithName("controllers").WithName("KThreesConfig"),
 		Scheme: mgr.GetScheme(),
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrencyNumber); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "KThreesConfig")
 		os.Exit(1)
 	}
@@ -108,7 +112,7 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
-	setupLog.Info("starting manager")
+	setupLog.Info("Starting manager", "concurrency", concurrencyNumber)
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)

controlplane/controllers/kthreescontrolplane_controller.go

@@ -283,11 +283,13 @@ func patchKThreesControlPlane(ctx context.Context, patchHelper *patch.Helper, kc
 	)
 }
 
-func (r *KThreesControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, log *logr.Logger) error {
+func (r *KThreesControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, log *logr.Logger, concurrency int) error {
 	c, err := ctrl.NewControllerManagedBy(mgr).
 		For(&controlplanev1.KThreesControlPlane{}).
 		Owns(&clusterv1.Machine{}).
-		//	WithOptions(options).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
 		WithEventFilter(predicates.ResourceNotPaused(r.Log)).
 		Watches(
 			&clusterv1.Cluster{},

controlplane/controllers/machine_controller.go

@@ -19,6 +19,7 @@ import (
 	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	k3s "github.com/k3s-io/cluster-api-k3s/pkg/k3s"
 )
@@ -43,9 +44,12 @@ type MachineReconciler struct {
 	managementClusterUncached k3s.ManagementCluster
 }
 
-func (r *MachineReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, log *logr.Logger) error {
+func (r *MachineReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, log *logr.Logger, concurrency int) error {
 	_, err := ctrl.NewControllerManagedBy(mgr).
 		For(&clusterv1.Machine{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
 		WithEventFilter(predicates.ResourceNotPaused(r.Log)).
 		Build(r)
 

controlplane/main.go

@@ -63,6 +63,7 @@ func main() {
 	var syncPeriod time.Duration
 	var etcdDialTimeout time.Duration
 	var etcdCallTimeout time.Duration
+	var concurrencyNumber int
 
 	flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "enable-leader-election", false,
@@ -78,6 +79,9 @@ func main() {
 	flag.DurationVar(&etcdCallTimeout, "etcd-call-timeout-duration", etcd.DefaultCallTimeout,
 		"Duration that the etcd client waits at most for read and write operations to etcd.")
 
+	flag.IntVar(&concurrencyNumber, "concurrency", 1,
+		"Number of core resources to process simultaneously")
+
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
@@ -110,7 +114,7 @@ func main() {
 		Scheme:          mgr.GetScheme(),
 		EtcdDialTimeout: etcdDialTimeout,
 		EtcdCallTimeout: etcdCallTimeout,
-	}).SetupWithManager(ctx, mgr, &ctrPlaneLogger); err != nil {
+	}).SetupWithManager(ctx, mgr, &ctrPlaneLogger, concurrencyNumber); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "KThreesControlPlane")
 		os.Exit(1)
 	}
@@ -122,7 +126,7 @@ func main() {
 		Scheme:          mgr.GetScheme(),
 		EtcdDialTimeout: etcdDialTimeout,
 		EtcdCallTimeout: etcdCallTimeout,
-	}).SetupWithManager(ctx, mgr, &ctrMachineLogger); err != nil {
+	}).SetupWithManager(ctx, mgr, &ctrMachineLogger, concurrencyNumber); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Machine")
 		os.Exit(1)
 	}
@@ -135,7 +139,7 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
-	setupLog.Info("starting manager")
+	setupLog.Info("Starting manager", "concurrency", concurrencyNumber)
 	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)