Concurrently reconcile CloudStackMachine resources

Author: chrisdoherty4

config/manager/manager.yaml

@@ -22,6 +22,7 @@ spec:
         - /manager
         args:
         - --leader-elect
+        - --cloudstackmachine-concurrency=${CAPC_CLOUDSTACKMACHINE_CONCURRENCY:=10}
         image: controller:latest
         name: manager
         securityContext:
@@ -48,7 +49,7 @@ spec:
       serviceAccountName: controller-manager
       terminationGracePeriodSeconds: 10
       tolerations:
-        - effect: NoSchedule
-          key: node-role.kubernetes.io/master
-        - effect: NoSchedule
-          key: node-role.kubernetes.io/control-plane
+      - effect: NoSchedule
+        key: node-role.kubernetes.io/master
+      - effect: NoSchedule
+        key: node-role.kubernetes.io/control-plane

controllers/cloudstackmachine_controller.go

@@ -32,6 +32,7 @@ import (
 	"sigs.k8s.io/cluster-api/util"
 	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/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -233,7 +234,6 @@ func (r *CloudStackMachineReconciliationRunner) GetOrCreateVMInstance() (retRes
 
 	userData := processCustomMetadata(data, r)
 	err := r.CSUser.GetOrCreateVMInstance(r.ReconciliationSubject, r.CAPIMachine, r.CSCluster, r.FailureDomain, r.AffinityGroup, userData)
-
 	if err != nil {
 		r.Recorder.Eventf(r.ReconciliationSubject, "Warning", "Creating", CSMachineCreationFailed, err.Error())
 	}
@@ -345,8 +345,9 @@ func (r *CloudStackMachineReconciliationRunner) ReconcileDelete() (retRes ctrl.R
 }
 
 // SetupWithManager registers the machine reconciler to the CAPI controller manager.
-func (reconciler *CloudStackMachineReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (reconciler *CloudStackMachineReconciler) SetupWithManager(mgr ctrl.Manager, opts controller.Options) error {
 	controller, err := ctrl.NewControllerManagedBy(mgr).
+		WithOptions(opts).
 		For(&infrav1.CloudStackMachine{}).
 		WithEventFilter(
 			predicate.Funcs{

controllers/cloudstackmachine_controller_test.go

@@ -18,6 +18,8 @@ package controllers_test
 
 import (
 	"fmt"
+	"strings"
+
 	"github.com/golang/mock/gomock"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -31,8 +33,8 @@ import (
 	"sigs.k8s.io/cluster-api/util/patch"
 	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/controller/controllerutil"
-	"strings"
 )
 
 var _ = Describe("CloudStackMachineReconciler", func() {
@@ -43,7 +45,7 @@ var _ = Describe("CloudStackMachineReconciler", func() {
 			dummies.CSCluster.Spec.FailureDomains[0].Name = dummies.CSFailureDomain1.Spec.Name
 
 			SetupTestEnvironment()                                              // Must happen before setting up managers/reconcilers.
-			Ω(MachineReconciler.SetupWithManager(k8sManager)).Should(Succeed()) // Register the CloudStack MachineReconciler.
+			Ω(MachineReconciler.SetupWithManager(k8sManager, controller.Options{})).Should(Succeed()) // Register the CloudStack MachineReconciler.
 
 			// Point CAPI machine Bootstrap secret ref to dummy bootstrap secret.
 			dummies.CAPIMachine.Spec.Bootstrap.DataSecretName = &dummies.BootstrapSecret.Name

main.go

@@ -37,6 +37,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -73,6 +74,8 @@ type managerOpts struct {
 	WatchingNamespace    string
 	WatchFilterValue     string
 	CertDir              string
+
+	CloudStackMachineConcurrency int
 }
 
 func setFlags() *managerOpts {
@@ -117,6 +120,13 @@ func setFlags() *managerOpts {
 		"webhook-cert-dir",
 		"/tmp/k8s-webhook-server/serving-certs/",
 		"Specify the directory where webhooks will get tls certificates.")
+	flag.IntVar(
+		&opts.CloudStackMachineConcurrency,
+		"cloudstackmachine-concurrency",
+		10,
+		"Maximum concurrent reconciles for CloudStackMachine resources",
+	)
+
 	return opts
 }
 
@@ -152,10 +162,11 @@ func main() {
 		K8sClient:  mgr.GetClient(),
 		BaseLogger: ctrl.Log.WithName("controllers"),
 		Recorder:   mgr.GetEventRecorderFor("capc-controller-manager"),
-		Scheme:     mgr.GetScheme()}
+		Scheme:     mgr.GetScheme(),
+	}
 
 	ctx := ctrl.SetupSignalHandler()
-	setupReconcilers(ctx, base, mgr)
+	setupReconcilers(ctx, base, *opts, mgr)
 	infrav1b3.K8sClient = base.K8sClient
 
 	// +kubebuilder:scaffold:builder
@@ -191,12 +202,12 @@ func main() {
 	}
 }
 
-func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, mgr manager.Manager) {
+func setupReconcilers(ctx context.Context, base utils.ReconcilerBase, opts managerOpts, mgr manager.Manager) {
 	if err := (&controllers.CloudStackClusterReconciler{ReconcilerBase: base}).SetupWithManager(ctx, mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackCluster")
 		os.Exit(1)
 	}
-	if err := (&controllers.CloudStackMachineReconciler{ReconcilerBase: base}).SetupWithManager(mgr); err != nil {
+	if err := (&controllers.CloudStackMachineReconciler{ReconcilerBase: base}).SetupWithManager(mgr, controller.Options{MaxConcurrentReconciles: opts.CloudStackMachineConcurrency}); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "CloudStackMachine")
 		os.Exit(1)
 	}

pkg/cloud/instance.go

@@ -291,7 +291,6 @@ func (c *client) CheckLimits(
 	fd *infrav1.CloudStackFailureDomain,
 	offering *cloudstack.ServiceOffering,
 ) error {
-
 	err := c.CheckAccountLimits(fd, offering)
 	if err != nil {
 		return err
@@ -316,7 +315,6 @@ func (c *client) DeployVM(
 	offering *cloudstack.ServiceOffering,
 	userData string,
 ) error {
-
 	templateID, err := c.ResolveTemplate(csCluster, csMachine, fd.Spec.Zone.ID)
 	if err != nil {
 		return err
@@ -391,7 +389,6 @@ func (c *client) GetOrCreateVMInstance(
 	affinity *infrav1.CloudStackAffinityGroup,
 	userData string,
 ) error {
-
 	// Check if VM instance already exists.
 	if err := c.ResolveVMInstanceDetails(csMachine); err == nil ||
 		!strings.Contains(strings.ToLower(err.Error()), "no match") {
@@ -499,5 +496,4 @@ func (c *client) listVMInstanceDatadiskVolumeIDs(instanceID string) ([]string, e
 	}
 
 	return ret, nil
-
 }