Merge pull request #1302 from ncdc/0.1/concurrency

⚠️ [0.1] Cluster, machine controller: support concurrency
kubernetes-sigs · Aug 21, 2019 · c80f6e5 · c80f6e5
2 parents 961efcc + 28cc750
commit c80f6e5
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 12 deletions.
diff --git a/cmd/example-provider/main.go b/cmd/example-provider/main.go
@@ -33,8 +33,15 @@ import (
 )
 
 func main() {
+	var (
+		clusterConcurrency int
+		machineConcurrency int
+	)
+
 	klog.InitFlags(nil)
 	flag.Set("logtostderr", "true")
+	flag.IntVar(&clusterConcurrency, "cluster-concurrency", 1, "Number of clusters to process simultaneously")
+	flag.IntVar(&machineConcurrency, "machine-concurrency", 1, "Number of machines to process simultaneously")
 	flag.Parse()
 
 	cfg := config.GetConfigOrDie()
@@ -65,8 +72,8 @@ func main() {
 		klog.Fatal(err)
 	}
 
-	capimachine.AddWithActuator(mgr, machineActuator)
-	capicluster.AddWithActuator(mgr, clusterActuator)
+	capimachine.AddWithActuator(mgr, machineActuator, machineConcurrency)
+	capicluster.AddWithActuator(mgr, clusterActuator, clusterConcurrency)
 
 	if err := mgr.Start(signals.SetupSignalHandler()); err != nil {
 		klog.Fatalf("Failed to run manager: %v", err)

diff --git a/pkg/controller/cluster/cluster_controller.go b/pkg/controller/cluster/cluster_controller.go
@@ -47,13 +47,13 @@ const deleteRequeueAfter = 5 * time.Second
 
 var DefaultActuator Actuator
 
-func AddWithActuator(mgr manager.Manager, actuator Actuator) error {
+func AddWithActuator(mgr manager.Manager, actuator Actuator, concurrency int) error {
 	reconciler, err := newReconciler(mgr, actuator)
 	if err != nil {
 		return err
 	}
 
-	return add(mgr, reconciler)
+	return add(mgr, reconciler, concurrency)
 }
 
 // newReconciler returns a new reconcile.Reconciler
@@ -70,9 +70,9 @@ func newReconciler(mgr manager.Manager, actuator Actuator) (reconcile.Reconciler
 }
 
 // add adds a new Controller to mgr with r as the reconcile.Reconciler
-func add(mgr manager.Manager, r reconcile.Reconciler) error {
+func add(mgr manager.Manager, r reconcile.Reconciler, concurrency int) error {
 	// Create a new controller
-	c, err := controller.New("cluster-controller", mgr, controller.Options{Reconciler: r})
+	c, err := controller.New("cluster-controller", mgr, controller.Options{Reconciler: r, MaxConcurrentReconciles: concurrency})
 	if err != nil {
 		return err
 	}

diff --git a/pkg/controller/cluster/cluster_reconciler_test.go b/pkg/controller/cluster/cluster_reconciler_test.go
@@ -60,7 +60,7 @@ func TestReconcile(t *testing.T) {
 		t.Fatalf("Couldn't create controller: %v", err)
 	}
 	recFn, requests := SetupTestReconcile(r)
-	if err := add(mgr, recFn); err != nil {
+	if err := add(mgr, recFn, 1); err != nil {
 		t.Fatalf("error adding controller to manager: %v", err)
 	}
 	defer close(StartTestManager(mgr, t))

diff --git a/pkg/controller/machine/machine_controller.go b/pkg/controller/machine/machine_controller.go
@@ -45,8 +45,8 @@ const (
 
 var DefaultActuator Actuator
 
-func AddWithActuator(mgr manager.Manager, actuator Actuator) error {
-	return add(mgr, newReconciler(mgr, actuator))
+func AddWithActuator(mgr manager.Manager, actuator Actuator, concurrency int) error {
+	return add(mgr, newReconciler(mgr, actuator), concurrency)
 }
 
 // newReconciler returns a new reconcile.Reconciler
@@ -66,9 +66,9 @@ func newReconciler(mgr manager.Manager, actuator Actuator) reconcile.Reconciler
 }
 
 // add adds a new Controller to mgr with r as the reconcile.Reconciler
-func add(mgr manager.Manager, r reconcile.Reconciler) error {
+func add(mgr manager.Manager, r reconcile.Reconciler, concurrency int) error {
 	// Create a new controller
-	c, err := controller.New("machine-controller", mgr, controller.Options{Reconciler: r})
+	c, err := controller.New("machine-controller", mgr, controller.Options{Reconciler: r, MaxConcurrentReconciles: concurrency})
 	if err != nil {
 		return err
 	}

diff --git a/pkg/controller/machine/machine_reconciler_test.go b/pkg/controller/machine/machine_reconciler_test.go
@@ -53,7 +53,7 @@ func TestReconcile(t *testing.T) {
 
 	a := newTestActuator()
 	recFn, requests := SetupTestReconcile(newReconciler(mgr, a))
-	if err := add(mgr, recFn); err != nil {
+	if err := add(mgr, recFn, 1); err != nil {
 		t.Fatalf("error adding controller to manager: %v", err)
 	}
 	defer close(StartTestManager(mgr, t))