Enable concurrency in BMO controllers

controllers/metal3.io/baremetalhost_controller.go

@@ -19,10 +19,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
 	"reflect"
-	"runtime"
-	"strconv"
 	"strings"
 	"time"
 
@@ -1732,41 +1729,15 @@ 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 {
-
-	maxConcurrentReconciles := runtime.NumCPU()
-	if maxConcurrentReconciles > 8 {
-		maxConcurrentReconciles = 8
-	}
-	if maxConcurrentReconciles < 2 {
-		maxConcurrentReconciles = 2
-	}
-	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))
-		}
-		if mcr > 0 {
-			ctrl.Log.Info(fmt.Sprintf("BMO_CONCURRENCY of %d is set via an environment variable", mcr))
-			maxConcurrentReconciles = mcr
-		} else {
-			ctrl.Log.Info(fmt.Sprintf("Invalid BMO_CONCURRENCY value. Operator Concurrency will be set to a default value of %d", maxConcurrentReconciles))
-		}
-	} else {
-		ctrl.Log.Info(fmt.Sprintf("Operator Concurrency will be set to a default value of %d", maxConcurrentReconciles))
-	}
-
-	opts := controller.Options{
-		MaxConcurrentReconciles: maxConcurrentReconciles,
-	}
+func (r *BareMetalHostReconciler) SetupWithManager(mgr ctrl.Manager, preprovImgEnable bool, options controller.Options) error {
 
 	controller := ctrl.NewControllerManagedBy(mgr).
 		For(&metal3api.BareMetalHost{}).
 		WithEventFilter(
 			predicate.Funcs{
 				UpdateFunc: r.updateEventHandler,
 			}).
-		WithOptions(opts).
+		WithOptions(options).
 		Owns(&corev1.Secret{}, builder.MatchEveryOwner)
 
 	if preprovImgEnable {

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"
@@ -289,9 +290,11 @@ 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 {
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3api.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,11 @@ 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 {
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3api.HostFirmwareSettings{}).
+		WithOptions(options).
 		WithEventFilter(
 			predicate.Funcs{
 				UpdateFunc: r.updateEventHandler,

controllers/metal3.io/preprovisioningimage_controller.go

@@ -31,6 +31,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"
 
 	metal3 "github.com/metal3-io/baremetal-operator/apis/metal3.io/v1alpha1"
 	"github.com/metal3-io/baremetal-operator/pkg/imageprovider"
@@ -351,9 +352,10 @@ 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 {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&metal3.PreprovisioningImage{}).
+		WithOptions(options).
 		Owns(&corev1.Secret{}, builder.MatchEveryOwner).
 		Complete(r)
 }

main.go

@@ -21,15 +21,18 @@ import (
 	"fmt"
 	"os"
 	"runtime"
+	"strconv"
 	"strings"
 
+	"github.com/pkg/errors"
 	"go.uber.org/zap/zapcore"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	cliflag "k8s.io/component-base/cli/flag"
 	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"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -123,6 +126,7 @@ func main() {
 	var webhookPort int
 	var restConfigQPS float64
 	var restConfigBurst int
+	var bmoConcurrency 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
@@ -164,6 +168,8 @@ func main() {
 			"If omitted, the default Go cipher suites will be used. \n"+
 			"Preferred values: "+strings.Join(tlsCipherPreferredValues, ", ")+". \n"+
 			"Insecure values: "+strings.Join(tlsCipherInsecureValues, ", ")+".")
+	flag.IntVar(&bmoConcurrency, "bmo-concurrency", 0,
+		"Number of baremetalhosts to process simultaneously")
 	flag.Parse()
 
 	logOpts := zap.Options{}
@@ -227,12 +233,18 @@ func main() {
 		provisionerFactory = ironic.NewProvisionerFactory(provLog, preprovImgEnable)
 	}
 
+	maxConcurrency, err := getMaxConcurrentReconciles(bmoConcurrency)
+	if err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "BareMetalHost")
+		os.Exit(1)
+	}
+
 	if err = (&metal3iocontroller.BareMetalHostReconciler{
 		Client:             mgr.GetClient(),
 		Log:                ctrl.Log.WithName("controllers").WithName("BareMetalHost"),
 		ProvisionerFactory: provisionerFactory,
 		APIReader:          mgr.GetAPIReader(),
-	}).SetupWithManager(mgr, preprovImgEnable); err != nil {
+	}).SetupWithManager(mgr, preprovImgEnable, concurrency(maxConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "BareMetalHost")
 		os.Exit(1)
 	}
@@ -246,7 +258,7 @@ func main() {
 			ImageProvider: imageprovider.NewDefaultImageProvider(),
 		}
 		if imgReconciler.CanStart() {
-			if err = (&imgReconciler).SetupWithManager(mgr); err != nil {
+			if err = (&imgReconciler).SetupWithManager(mgr, concurrency(maxConcurrency)); err != nil {
 				setupLog.Error(err, "unable to create controller", "controller", "PreprovisioningImage")
 				os.Exit(1)
 			}
@@ -258,7 +270,7 @@ func main() {
 		Client:             mgr.GetClient(),
 		Log:                ctrl.Log.WithName("controllers").WithName("HostFirmwareSettings"),
 		ProvisionerFactory: provisionerFactory,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrency(maxConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "HostFirmwareSettings")
 		os.Exit(1)
 	}
@@ -267,7 +279,7 @@ func main() {
 		Client:             mgr.GetClient(),
 		Log:                ctrl.Log.WithName("controllers").WithName("BMCEventSubscription"),
 		ProvisionerFactory: provisionerFactory,
-	}).SetupWithManager(mgr); err != nil {
+	}).SetupWithManager(mgr, concurrency(maxConcurrency)); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "BMCEventSubscription")
 		os.Exit(1)
 	}
@@ -356,3 +368,41 @@ func GetTLSVersion(version string) (uint16, error) {
 	}
 	return v, nil
 }
+
+func concurrency(c int) controller.Options {
+	return controller.Options{MaxConcurrentReconciles: c}
+}
+
+func getMaxConcurrentReconciles(bmoConcurrency int) (int, error) {
+	if bmoConcurrency > 0 {
+		ctrl.Log.Info(fmt.Sprintf("Controller Concurrency will be set to %d according to command line flag", bmoConcurrency))
+		return bmoConcurrency, nil
+	} else if bmoConcurrency < 0 {
+		return 0, fmt.Errorf("bmo-concurrency value: %d is invalid", bmoConcurrency)
+	}
+
+	// bmo-concurrency value is 0 i.e. no values passed via the flag
+	// maxConcurrentReconcile value would be set based on env var or number of CPUs.
+	maxConcurrentReconciles := runtime.NumCPU()
+	if maxConcurrentReconciles > 8 {
+		maxConcurrentReconciles = 8
+	}
+	if maxConcurrentReconciles < 2 {
+		maxConcurrentReconciles = 2
+	}
+	if mcrEnv, ok := os.LookupEnv("BMO_CONCURRENCY"); ok {
+		mcr, err := strconv.Atoi(mcrEnv)
+		if err != nil {
+			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))
+			maxConcurrentReconciles = mcr
+		} else {
+			ctrl.Log.Info(fmt.Sprintf("Invalid BMO_CONCURRENCY value. Operator Concurrency will be set to a default value of %d", maxConcurrentReconciles))
+		}
+	} else {
+		ctrl.Log.Info(fmt.Sprintf("Operator Concurrency will be set to a default value of %d", maxConcurrentReconciles))
+	}
+	return maxConcurrentReconciles, nil
+}