Enable concurrency in BMO controllers

controllers/metal3.io/baremetalhost_controller.go

@@ -1707,8 +1707,37 @@ func (r *BareMetalHostReconciler) updateEventHandler(e event.UpdateEvent) bool {
 }
 
 // SetupWithManager registers the reconciler to be run by the manager
-func (r *BareMetalHostReconciler) SetupWithManager(mgr ctrl.Manager, preprovImgEnable bool) error {
+func (r *BareMetalHostReconciler) SetupWithManager(mgr ctrl.Manager, preprovImgEnable bool, options controller.Options) error {
 
+	maxConcurrentReconciles := 0
+	if options.MaxConcurrentReconciles > 0 {
+		ctrl.Log.Info(fmt.Sprintf("Controller Concurrency will be set to %d according to command line flag", options.MaxConcurrentReconciles))
+	} else {
+		mCR, err := getMaxConcurrentReconciles()
+		if err != nil {
+			return err
+		}
+		maxConcurrentReconciles = mCR
+	}
+	options.MaxConcurrentReconciles = maxConcurrentReconciles
+
+	controller := ctrl.NewControllerManagedBy(mgr).
+		For(&metal3v1alpha1.BareMetalHost{}).
+		WithEventFilter(
+			predicate.Funcs{
+				UpdateFunc: r.updateEventHandler,
+			}).
+		WithOptions(options).
+		Owns(&corev1.Secret{})
+
+	if preprovImgEnable {
+		controller.Owns(&metal3v1alpha1.PreprovisioningImage{})
+	}
+
+	return controller.Complete(r)
+}
+
+func getMaxConcurrentReconciles() (int, error) {
 	maxConcurrentReconciles := runtime.NumCPU()
 	if maxConcurrentReconciles > 8 {
 		maxConcurrentReconciles = 8
@@ -1719,7 +1748,7 @@ func (r *BareMetalHostReconciler) SetupWithManager(mgr ctrl.Manager, preprovImgE
 	if mcrEnv, ok := os.LookupEnv("BMO_CONCURRENCY"); ok {
 		mcr, err := strconv.Atoi(mcrEnv)
 		if err != nil {
-			return errors.Wrap(err, fmt.Sprintf("BMO_CONCURRENCY value: %s is invalid", mcrEnv))
+			return 0, errors.Wrap(err, fmt.Sprintf("BMO_CONCURRENCY value: %s is invalid", mcrEnv))
 		}
 		if mcr > 0 {
 			ctrl.Log.Info(fmt.Sprintf("BMO_CONCURRENCY of %d is set via an environment variable", mcr))
@@ -1730,25 +1759,7 @@ func (r *BareMetalHostReconciler) SetupWithManager(mgr ctrl.Manager, preprovImgE
 	} else {
 		ctrl.Log.Info(fmt.Sprintf("Operator Concurrency will be set to a default value of %d", maxConcurrentReconciles))
 	}
-
-	opts := controller.Options{
-		MaxConcurrentReconciles: maxConcurrentReconciles,
-	}
-
-	controller := ctrl.NewControllerManagedBy(mgr).
-		For(&metal3v1alpha1.BareMetalHost{}).
-		WithEventFilter(
-			predicate.Funcs{
-				UpdateFunc: r.updateEventHandler,
-			}).
-		WithOptions(opts).
-		Owns(&corev1.Secret{})
-
-	if preprovImgEnable {
-		controller.Owns(&metal3v1alpha1.PreprovisioningImage{})
-	}
-
-	return controller.Complete(r)
+	return maxConcurrentReconciles, nil
 }
 
 func (r *BareMetalHostReconciler) reconciletHostData(ctx context.Context, host *metal3v1alpha1.BareMetalHost, request ctrl.Request) (result ctrl.Result, err error) {

controllers/metal3.io/bmceventsubscription_controller.go

@@ -30,6 +30,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -290,9 +291,23 @@ func (r *BMCEventSubscriptionReconciler) updateEventHandler(e event.UpdateEvent)
 }
 
 // SetupWithManager registers the reconciler to be run by the manager
-func (r *BMCEventSubscriptionReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *BMCEventSubscriptionReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
+
+	maxConcurrentReconciles := 0
+	if options.MaxConcurrentReconciles > 0 {
+		ctrl.Log.Info(fmt.Sprintf("Controller Concurrency will be set to %d according to command line flag", options.MaxConcurrentReconciles))
+	} else {
+		mCR, err := getMaxConcurrentReconciles()
+		if err != nil {
+			return err
+		}
+		maxConcurrentReconciles = mCR
+	}
+	options.MaxConcurrentReconciles = maxConcurrentReconciles
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3v1alpha1.BMCEventSubscription{}).
+		WithOptions(options).
 		WithEventFilter(predicate.Funcs{
 			UpdateFunc: r.updateEventHandler,
 		}).

controllers/metal3.io/hostfirmwaresettings_controller.go

@@ -35,6 +35,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	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/predicate"
@@ -364,10 +365,23 @@ func (r *HostFirmwareSettingsReconciler) getOrCreateFirmwareSchema(info *rInfo,
 }
 
 // SetupWithManager sets up the controller with the Manager.
-func (r *HostFirmwareSettingsReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *HostFirmwareSettingsReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
+
+	maxConcurrentReconciles := 0
+	if options.MaxConcurrentReconciles > 0 {
+		ctrl.Log.Info(fmt.Sprintf("Controller Concurrency will be set to %d according to command line flag", options.MaxConcurrentReconciles))
+	} else {
+		mCR, err := getMaxConcurrentReconciles()
+		if err != nil {
+			return err
+		}
+		maxConcurrentReconciles = mCR
+	}
+	options.MaxConcurrentReconciles = maxConcurrentReconciles
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3v1alpha1.HostFirmwareSettings{}).
+		WithOptions(options).
 		WithEventFilter(
 			predicate.Funcs{
 				UpdateFunc: r.updateEventHandler,

controllers/metal3.io/preprovisioningimage_controller.go

@@ -18,6 +18,7 @@ package controllers
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -30,6 +31,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	metal3 "github.com/metal3-io/baremetal-operator/apis/metal3.io/v1alpha1"
 	"github.com/metal3-io/baremetal-operator/pkg/imageprovider"
@@ -350,9 +352,22 @@ func (r *PreprovisioningImageReconciler) CanStart() bool {
 	return false
 }
 
-func (r *PreprovisioningImageReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func (r *PreprovisioningImageReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
+	maxConcurrentReconciles := 0
+	if options.MaxConcurrentReconciles > 0 {
+		ctrl.Log.Info(fmt.Sprintf("Controller Concurrency will be set to %d according to command line flag", options.MaxConcurrentReconciles))
+	} else {
+		mCR, err := getMaxConcurrentReconciles()
+		if err != nil {
+			return err
+		}
+		maxConcurrentReconciles = mCR
+	}
+	options.MaxConcurrentReconciles = maxConcurrentReconciles
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3.PreprovisioningImage{}).
+		WithOptions(options).
 		Owns(&corev1.Secret{}).
 		Complete(r)
 }

main.go

@@ -27,6 +27,7 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
@@ -95,6 +96,10 @@ func main() {
 	var runInTestMode bool
 	var runInDemoMode bool
 	var webhookPort int
+	var baremetalhostConcurrency int
+	var preprovisioningimageConcurrency int
+	var hostfirmwaresettingsConcurrency int
+	var bmceventsubscriptionConcurrency int
 
 	// From CAPI point of view, BMO should be able to watch all namespaces
 	// in case of a deployment that is not multi-tenant. If the deployment
@@ -116,6 +121,14 @@ func main() {
 		"The address the health endpoint binds to.")
 	flag.IntVar(&webhookPort, "webhook-port", 9443,
 		"Webhook Server port (set to 0 to disable)")
+	flag.IntVar(&baremetalhostConcurrency, "baremetalhost-concurrency", 10,
+		"Number of baremetalhosts to process simultaneously")
+	flag.IntVar(&preprovisioningimageConcurrency, "preprovisioningimage-concurrency", 10,
+		"Number of preprovisioningimages to process simultaneously")
+	flag.IntVar(&hostfirmwaresettingsConcurrency, "hostfirmwaresettings-concurrency", 10,
+		"Number of hostfirmwaresettings to process simultaneously")
+	flag.IntVar(&bmceventsubscriptionConcurrency, "bmceventsubscription-concurrency", 10,
+		"Number of bmceventsubscriptions to process simultaneously")
 	flag.Parse()
 
 	logOpts := zap.Options{}
@@ -170,7 +183,7 @@ func main() {
 		Log:                ctrl.Log.WithName("controllers").WithName("BareMetalHost"),
 		ProvisionerFactory: provisionerFactory,
 		APIReader:          mgr.GetAPIReader(),
-	}).SetupWithManager(mgr, preprovImgEnable); err != nil {
+	}).SetupWithManager(mgr, preprovImgEnable, concurrency(baremetalhostConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "BareMetalHost")
 		os.Exit(1)
 	}
@@ -184,7 +197,7 @@ func main() {
 			ImageProvider: imageprovider.NewDefaultImageProvider(),
 		}
 		if imgReconciler.CanStart() {
-			if err = (&imgReconciler).SetupWithManager(mgr); err != nil {
+			if err = (&imgReconciler).SetupWithManager(mgr, concurrency(preprovisioningimageConcurrency)); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "PreprovisioningImage")
 				os.Exit(1)
 			}
@@ -196,7 +209,7 @@ func main() {
 		Client:             mgr.GetClient(),
 		Log:                ctrl.Log.WithName("controllers").WithName("HostFirmwareSettings"),
 		ProvisionerFactory: provisionerFactory,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrency(hostfirmwaresettingsConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "HostFirmwareSettings")
 		os.Exit(1)
 	}
@@ -205,7 +218,7 @@ func main() {
 		Client:             mgr.GetClient(),
 		Log:                ctrl.Log.WithName("controllers").WithName("BMCEventSubscription"),
 		ProvisionerFactory: provisionerFactory,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrency(bmceventsubscriptionConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "BMCEventSubscription")
 		os.Exit(1)
 	}
@@ -222,3 +235,7 @@ func main() {
 		os.Exit(1)
 	}
 }
+
+func concurrency(c int) controller.Options {
+	return controller.Options{MaxConcurrentReconciles: c}
+}