feat: Make affinity group and failure domain controller run with multiple workers

Upstream PR kubernetes-sigs#341

Author: hrak

controllers/cloudstackaffinitygroup_controller.go

@@ -21,6 +21,7 @@ import (
 
 	"sigs.k8s.io/cluster-api/util/predicates"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	infrav1 "sigs.k8s.io/cluster-api-provider-cloudstack/api/v1beta3"
@@ -90,8 +91,9 @@ func (r *CloudStackAGReconciliationRunner) ReconcileDelete() (ctrl.Result, error
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (reconciler *CloudStackAffinityGroupReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+func (reconciler *CloudStackAffinityGroupReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrav1.CloudStackAffinityGroup{}).
 		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), reconciler.WatchFilterValue)).
 		Complete(reconciler)

controllers/cloudstackaffinitygroup_controller_test.go

@@ -23,13 +23,14 @@ import (
 	infrav1 "sigs.k8s.io/cluster-api-provider-cloudstack/api/v1beta3"
 	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("CloudStackAffinityGroupReconciler", func() {
 	BeforeEach(func() {
 		SetupTestEnvironment() // Must happen before setting up managers/reconcilers.
 		dummies.SetDummyVars()
-		Ω(AffinityGReconciler.SetupWithManager(ctx, k8sManager)).Should(Succeed()) // Register CloudStack AffinityGReconciler.
+		Ω(AffinityGReconciler.SetupWithManager(ctx, k8sManager, controller.Options{})).Should(Succeed()) // Register CloudStack AffinityGReconciler.
 	})
 
 	It("Should patch back the affinity group as ready after calling GetOrCreateAffinityGroup.", func() {

controllers/cloudstackcluster_controller_test.go

@@ -30,9 +30,9 @@ import (
 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, controller.Options{})).Should(Succeed()) // Register CloudStack ClusterReconciler.
-			Ω(FailureDomainReconciler.SetupWithManager(ctx, 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(ctx, k8sManager, controller.Options{})).Should(Succeed()) // Register CloudStack FailureDomainReconciler.
 		})
 
 		It("Should create a CloudStackFailureDomain.", func() {

controllers/cloudstackfailuredomain_controller.go

@@ -18,6 +18,7 @@ package controllers
 
 import (
 	"context"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sort"
 
 	"github.com/pkg/errors"
@@ -265,8 +266,9 @@ func (r *CloudStackFailureDomainReconciliationRunner) RemoveFinalizer() (ctrl.Re
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (reconciler *CloudStackFailureDomainReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
+func (reconciler *CloudStackFailureDomainReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opts controller.Options) error {
 	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrav1.CloudStackFailureDomain{}).
 		WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(ctrl.LoggerFrom(ctx), reconciler.WatchFilterValue)).
 		Complete(reconciler)

controllers/cloudstackfailuredomain_controller_test.go

@@ -28,14 +28,15 @@ import (
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/util/patch"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 )
 
 var _ = Describe("CloudStackFailureDomainReconciler", func() {
 	Context("With k8s like test environment.", func() {
 		BeforeEach(func() {
 			dummies.SetDummyVars()
-			SetupTestEnvironment()                                                         // Must happen before setting up managers/reconcilers.
-			Ω(FailureDomainReconciler.SetupWithManager(ctx, k8sManager)).Should(Succeed()) // Register CloudStack FailureDomainReconciler.
+			SetupTestEnvironment()                                                                               // Must happen before setting up managers/reconcilers.
+			Ω(FailureDomainReconciler.SetupWithManager(ctx, k8sManager, controller.Options{})).Should(Succeed()) // Register CloudStack FailureDomainReconciler.
 			// Modify failure domain name the same way the cluster controller would.
 			dummies.CSFailureDomain1.Name = dummies.CSFailureDomain1.Name + "-" + dummies.CSCluster.Name
 

main.go

@@ -90,8 +90,10 @@ type managerOpts struct {
 	WebhookCertDir       string
 	WebhookPort          int
 
-	CloudStackClusterConcurrency int
-	CloudStackMachineConcurrency int
+	CloudStackClusterConcurrency       int
+	CloudStackMachineConcurrency       int
+	CloudStackAffinityGroupConcurrency int
+	CloudStackFailureDomainConcurrency int
 }
 
 func setFlags() *managerOpts {
@@ -158,6 +160,18 @@ func setFlags() *managerOpts {
 		10,
 		"Maximum concurrent reconciles for CloudStackMachine resources",
 	)
+	flag.IntVar(
+		&opts.CloudStackAffinityGroupConcurrency,
+		"cloudstackaffinitygroup-concurrency",
+		5,
+		"Maximum concurrent reconciles for CloudStackAffinityGroup resources",
+	)
+	flag.IntVar(
+		&opts.CloudStackFailureDomainConcurrency,
+		"cloudstackfailuredomain-concurrency",
+		5,
+		"Maximum concurrent reconciles for CloudStackFailureDomain resources",
+	)
 
 	return opts
 }
@@ -292,11 +306,11 @@ func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, opts manag
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackIsoNetReconciler")
 		os.Exit(1)
 	}
-	if err := (&controllers.CloudStackAffinityGroupReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr); err != nil {
+	if err := (&controllers.CloudStackAffinityGroupReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr, controller.Options{MaxConcurrentReconciles: opts.CloudStackAffinityGroupConcurrency}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackAffinityGroup")
 		os.Exit(1)
 	}
-	if err := (&controllers.CloudStackFailureDomainReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr); err != nil {
+	if err := (&controllers.CloudStackFailureDomainReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr, controller.Options{MaxConcurrentReconciles: opts.CloudStackFailureDomainConcurrency}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackFailureDomain")
 		os.Exit(1)
 	}