Merge pull request #296 from chrisdoherty4/feature/concurrent-cluster-reconciliation

Concurrently reconcile CloudStackCluster resources

Author: k8s-ci-robot

config/manager/manager.yaml

@@ -22,6 +22,7 @@ spec:
         - /manager
         args:
         - --leader-elect
+        - --cloudstackcluster-concurrency=${CAPC_CLOUDSTACKCLUSTER_CONCURRENCY:=10}
         - --cloudstackmachine-concurrency=${CAPC_CLOUDSTACKMACHINE_CONCURRENCY:=10}
         image: controller:latest
         name: manager

controllers/cloudstackcluster_controller.go

@@ -22,6 +22,7 @@ import (
 	"reflect"
 
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -129,7 +130,8 @@ func (r *CloudStackClusterReconciliationRunner) SetFailureDomainsStatusMap() (ct
 	for _, fdSpec := range r.ReconciliationSubject.Spec.FailureDomains {
 		metaHashName := infrav1.FailureDomainHashedMetaName(fdSpec.Name, r.CAPICluster.Name)
 		r.ReconciliationSubject.Status.FailureDomains[fdSpec.Name] = clusterv1.FailureDomainSpec{
-			ControlPlane: true, Attributes: map[string]string{"MetaHashName": metaHashName}}
+			ControlPlane: true, Attributes: map[string]string{"MetaHashName": metaHashName},
+		}
 	}
 	return ctrl.Result{}, nil
 }
@@ -153,8 +155,9 @@ func (r *CloudStackClusterReconciliationRunner) ReconcileDelete() (ctrl.Result,
 }
 
 // Called in main, this registers the cluster reconciler to the CAPI controller manager.
-func (reconciler *CloudStackClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+func (reconciler *CloudStackClusterReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts controller.Options) error {
 	controller, err := ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrav1.CloudStackCluster{}).
 		WithEventFilter(
 			predicate.Funcs{
@@ -194,6 +197,7 @@ func (reconciler *CloudStackClusterReconciler) SetupWithManager(ctx context.Cont
 				return oldCluster.Spec.Paused && !newCluster.Spec.Paused
 			},
 			DeleteFunc: func(e event.DeleteEvent) bool { return false },
-			CreateFunc: func(e event.CreateEvent) bool { return false }})
+			CreateFunc: func(e event.CreateEvent) bool { return false },
+		})
 	return errors.Wrap(err, "building CloudStackCluster controller")
 }

controllers/cloudstackcluster_controller_test.go

@@ -24,14 +24,15 @@ import (
 	"sigs.k8s.io/cluster-api-provider-cloudstack/controllers"
 	dummies "sigs.k8s.io/cluster-api-provider-cloudstack/test/dummies/v1beta3"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 )
 
 var _ = Describe("CloudStackClusterReconciler", func() {
 	Context("With k8s like test environment.", func() {
 		BeforeEach(func() {
-			SetupTestEnvironment()                                                    // Must happen before setting up managers/reconcilers.
-			Ω(ClusterReconciler.SetupWithManager(ctx, k8sManager)).Should(Succeed())  // Register CloudStack ClusterReconciler.
-			Ω(FailureDomainReconciler.SetupWithManager(k8sManager)).Should(Succeed()) // Register CloudStack FailureDomainReconciler.
+			SetupTestEnvironment()                                                                         // Must happen before setting up managers/reconcilers.
+			Ω(ClusterReconciler.SetupWithManager(ctx, k8sManager, controller.Options{})).Should(Succeed()) // Register CloudStack ClusterReconciler.
+			Ω(FailureDomainReconciler.SetupWithManager(k8sManager)).Should(Succeed())                      // Register CloudStack FailureDomainReconciler.
 		})
 
 		It("Should create a CloudStackFailureDomain.", func() {

main.go

@@ -75,6 +75,7 @@ type managerOpts struct {
 	WatchFilterValue     string
 	CertDir              string
 
+	CloudStackClusterConcurrency int
 	CloudStackMachineConcurrency int
 }
 
@@ -120,6 +121,12 @@ func setFlags() *managerOpts {
 		"webhook-cert-dir",
 		"/tmp/k8s-webhook-server/serving-certs/",
 		"Specify the directory where webhooks will get tls certificates.")
+	flag.IntVar(
+		&opts.CloudStackClusterConcurrency,
+		"cloudstackcluster-concurrency",
+		10,
+		"Maximum concurrent reconciles for CloudStackCluster resources",
+	)
 	flag.IntVar(
 		&opts.CloudStackMachineConcurrency,
 		"cloudstackmachine-concurrency",
@@ -203,7 +210,7 @@ func main() {
 }
 
 func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, opts managerOpts, mgr manager.Manager) {
-	if err := (&controllers.CloudStackClusterReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr); err != nil {
+	if err := (&controllers.CloudStackClusterReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr, controller.Options{MaxConcurrentReconciles: opts.CloudStackClusterConcurrency}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackCluster")
 		os.Exit(1)
 	}