feat: configurable per-controller concurrency

CONTRIBUTING.md

@@ -26,28 +26,34 @@ the manager first. Using kind, create a cluster with:
 kind create cluster --name controller-test
 ```
 
-If in-cluster is the preferred way to test the controller, create a kind registry that kind can pull images from.
-To do that follow [this](https://kind.sigs.k8s.io/docs/user/local-registry/) guide.
+The controller watches Flux source/kustomize/helm resources, so install Flux into the cluster first:
 
-Once a local registry exists, build the controller image with:
+```
+flux install --components=source-controller,helm-controller,kustomize-controller
+```
+
+If in-cluster is the preferred way to test the controller, build the image and load it into kind directly
+(no local registry needed):
 
 ```
-docker build -t localhost:5001/ocm-controller/manager:v0.0.1 .
+make docker-build TAG=dev
+kind load docker-image ghcr.io/open-component-model/ocm-controller:dev --name controller-test
 ```
 
-When it's done, push it to the local registry:
+Install the controller with the Helm chart in `deploy/`, pointing the manager image at the tag you just loaded.
+`pullPolicy=Never` forces kind to use the loaded image instead of pulling from the registry:
 
 ```
-docker push localhost:5001/ocm-controller/manager:v0.0.1
+helm upgrade --install ocm-controller ./deploy -n ocm-system --create-namespace \
+  --set manager.image.tag=dev \
+  --set manager.image.pullPolicy=Never
 ```
 
-Now the controller can be installed onto the cluster. Before that, install all crds and rbacs with `make install`.
-There is a deployment manifest located [here](config/manager/deployment.yaml) which can be used to install the
-controller.
+The chart also deploys an in-cluster OCI registry. Its TLS certificate is provisioned by cert-manager when
+`tlsCert.generateTlsCert=true`; if cert-manager is not installed, disable registry TLS with
+`--set registry.tls.enabled=false`.
 
-_Note_: If a local kind cluster with local registry is used, change the `image: ` value from 
-`image: open-component-model/ocm-controller` to `localhost:5001/ocm-controller/manager:v0.0.1` so the Kind workers can
-pull to image from the local registry.
+Per-controller concurrency can be set via `manager.concurrency`, e.g. `--set manager.concurrency.resource=3`.
 
 If it's enough to run the manager locally, you don't need a local registry. Simply run:
 

README.md

@@ -441,6 +441,23 @@ helm install ocm-controller oci://ghcr.io/open-component-model/helm/ocm-controll
 This does require that certain items, like certificate secrets are pre-created by the user.
 For details, checkout the prime-test-cluster scripts under the `hack` folder.
 
+### Controller concurrency
+
+By default each controller reconciles one object at a time. To let busier
+controllers process more objects in parallel, set per-controller worker counts
+via the `manager.concurrency` chart value:
+
+```yaml
+manager:
+  concurrency:
+    componentVersion: 4
+    resource: 2
+```
+
+Valid controller names: `componentVersion`, `resource`, `snapshot`,
+`localization`, `configuration`, `fluxDeployer`. Controllers you omit keep the
+default of one worker.
+
 ## Uninstall
 
 Again, there are two ways to uninstall the controller. Using the ocm cli run `ocm controller uninstall`.

deploy/templates/deployment_manager.yaml

@@ -32,6 +32,26 @@ spec:
         args:
         - --zap-log-level=4
         - --leader-elect
+        {{- with .Values.manager.concurrency }}
+        {{- if .componentVersion }}
+        - {{ printf "--concurrency-component-version=%v" .componentVersion | quote }}
+        {{- end }}
+        {{- if .resource }}
+        - {{ printf "--concurrency-resource=%v" .resource | quote }}
+        {{- end }}
+        {{- if .snapshot }}
+        - {{ printf "--concurrency-snapshot=%v" .snapshot | quote }}
+        {{- end }}
+        {{- if .localization }}
+        - {{ printf "--concurrency-localization=%v" .localization | quote }}
+        {{- end }}
+        {{- if .configuration }}
+        - {{ printf "--concurrency-configuration=%v" .configuration | quote }}
+        {{- end }}
+        {{- if .fluxDeployer }}
+        - {{ printf "--concurrency-flux-deployer=%v" .fluxDeployer | quote }}
+        {{- end }}
+        {{- end }}
         - --oci-registry-addr={{ .Values.registry.address }}
         {{- if not .Values.registry.tls.enabled }}
         - --oci-registry-insecure-skip-verify

deploy/values.yaml

@@ -64,6 +64,15 @@ registry:
 
 manager:
   replicaCount: 1
+  # Optional per-controller worker counts. Unspecified controllers default to 1.
+  concurrency: {}
+  # concurrency:
+  #   componentVersion: 4
+  #   resource: 2
+  #   snapshot: 2
+  #   localization: 1
+  #   configuration: 1
+  #   fluxDeployer: 1
   image:
     repository: ghcr.io/open-component-model/ocm-controller
     pullPolicy: IfNotPresent

main.go

@@ -17,6 +17,7 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
+	controllerconfig "sigs.k8s.io/controller-runtime/pkg/config"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -51,94 +52,122 @@ func init() {
 	//+kubebuilder:scaffold:scheme
 }
 
-func main() {
-	var (
-		metricsAddr                   string
-		eventsAddr                    string
-		enableLeaderElection          bool
-		probeAddr                     string
-		pprofAddr                     string
-		ociRegistryAddr               string
-		ociRegistryCertSecretName     string
-		ociRegistryInsecureSkipVerify bool
-		ociRegistryNamespace          string
-		kubeAPIQPS                    float64
-		kubeAPIBurst                  int
-		kubeAPIRateLimiterDisabled    bool
-	)
+type flags struct {
+	metricsAddr                   string
+	eventsAddr                    string
+	enableLeaderElection          bool
+	probeAddr                     string
+	pprofAddr                     string
+	ociRegistryAddr               string
+	ociRegistryCertSecretName     string
+	ociRegistryInsecureSkipVerify bool
+	ociRegistryNamespace          string
+	kubeAPIQPS                    float64
+	kubeAPIBurst                  int
+	kubeAPIRateLimiterDisabled    bool
+	concurrencyComponentVersion   int
+	concurrencyResource           int
+	concurrencySnapshot           int
+	concurrencyLocalization       int
+	concurrencyConfiguration      int
+	concurrencyFluxDeployer       int
+}
 
+func bindFlags() (flags, zap.Options) {
+	var f flags
 	flag.StringVar(
-		&metricsAddr,
+		&f.metricsAddr,
 		"metrics-bind-address",
 		":9443",
 		"The address the metric endpoint binds to.",
 	)
-	flag.StringVar(&eventsAddr, "events-addr", "", "The address of the events receiver.")
+	flag.StringVar(&f.eventsAddr, "events-addr", "", "The address of the events receiver.")
 	flag.StringVar(
-		&probeAddr,
+		&f.probeAddr,
 		"health-probe-bind-address",
 		":8081",
 		"The address the probe endpoint binds to.",
 	)
 	flag.StringVar(
-		&pprofAddr,
+		&f.pprofAddr,
 		"pprof-bind-address",
 		":6060",
 		"The address the pprof endpoint binds to (only used in builds with the pprof tag).",
 	)
 	flag.StringVar(
-		&ociRegistryAddr,
+		&f.ociRegistryAddr,
 		"oci-registry-addr",
 		":5000",
 		"The address of the OCI registry.",
 	)
 	flag.StringVar(
-		&ociRegistryCertSecretName,
+		&f.ociRegistryCertSecretName,
 		"certificate-secret-name",
 		v1alpha1.DefaultRegistryCertificateSecretName,
 		"",
 	)
 	flag.StringVar(
-		&ociRegistryNamespace,
+		&f.ociRegistryNamespace,
 		"oci-registry-namespace",
 		"ocm-system",
 		"The namespace in which the registry is running in.",
 	)
 	flag.BoolVar(
-		&ociRegistryInsecureSkipVerify,
+		&f.ociRegistryInsecureSkipVerify,
 		"oci-registry-insecure-skip-verify",
 		false,
 		"Skip verification of the certificate that the registry is using.",
 	)
-	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
+	flag.BoolVar(&f.enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
 	flag.Float64Var(
-		&kubeAPIQPS,
+		&f.kubeAPIQPS,
 		"kube-api-qps",
 		defaultKubeAPIQPS,
 		"Maximum queries per second to the Kubernetes API. Ignored when --kube-api-rate-limiter-disabled is set.",
 	)
 	flag.IntVar(
-		&kubeAPIBurst,
+		&f.kubeAPIBurst,
 		"kube-api-burst",
 		defaultKubeAPIBurst,
 		"Maximum burst for Kubernetes API client throttling. Ignored when --kube-api-rate-limiter-disabled is set.",
 	)
 	flag.BoolVar(
-		&kubeAPIRateLimiterDisabled,
+		&f.kubeAPIRateLimiterDisabled,
 		"kube-api-rate-limiter-disabled",
 		false,
 		"Disable the client-side Kubernetes API rate limiter and rely on API priority and fairness instead.",
 	)
+	flag.IntVar(&f.concurrencyComponentVersion, "concurrency-component-version", 1, "Number of workers for the ComponentVersion controller.")
+	flag.IntVar(&f.concurrencyResource, "concurrency-resource", 1, "Number of workers for the Resource controller.")
+	flag.IntVar(&f.concurrencySnapshot, "concurrency-snapshot", 1, "Number of workers for the Snapshot controller.")
+	flag.IntVar(&f.concurrencyLocalization, "concurrency-localization", 1, "Number of workers for the Localization controller.")
+	flag.IntVar(&f.concurrencyConfiguration, "concurrency-configuration", 1, "Number of workers for the Configuration controller.")
+	flag.IntVar(&f.concurrencyFluxDeployer, "concurrency-flux-deployer", 1, "Number of workers for the FluxDeployer controller.")
 
 	opts := zap.Options{
 		Development: true,
 	}
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
+	return f, opts
+}
+
+func main() {
+	f, opts := bindFlags()
+
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
+	gk := func(kind string) string { return v1alpha1.GroupVersion.WithKind(kind).GroupKind().String() }
+	groupKindConcurrency := map[string]int{
+		gk("ComponentVersion"): f.concurrencyComponentVersion,
+		gk("Resource"):         f.concurrencyResource,
+		gk("Snapshot"):         f.concurrencySnapshot,
+		gk("Localization"):     f.concurrencyLocalization,
+		gk("Configuration"):    f.concurrencyConfiguration,
+		gk("FluxDeployer"):     f.concurrencyFluxDeployer,
+	}
 	// 2024-05-11 :
 	// We're setting the go-log leve because it is used by yqlib
 	// If we ever make it possible to set the log level for the service
@@ -147,38 +176,42 @@ func main() {
 	glog.SetLevel(glog.WARNING, "yq-lib")
 
 	restConfig := ctrl.GetConfigOrDie()
-	if kubeAPIRateLimiterDisabled {
+	if f.kubeAPIRateLimiterDisabled {
 		restConfig.QPS = -1
 		restConfig.Burst = 0
 	} else {
-		restConfig.QPS = float32(kubeAPIQPS)
-		restConfig.Burst = kubeAPIBurst
+		restConfig.QPS = float32(f.kubeAPIQPS)
+		restConfig.Burst = f.kubeAPIBurst
 	}
 
 	mgr, err := ctrl.NewManager(restConfig, ctrl.Options{
 		Scheme: scheme,
 		Metrics: metricsserver.Options{
-			BindAddress: metricsAddr,
+			BindAddress: f.metricsAddr,
 		},
-		HealthProbeBindAddress: probeAddr,
-		LeaderElection:         enableLeaderElection,
+		HealthProbeBindAddress: f.probeAddr,
+		LeaderElection:         f.enableLeaderElection,
 		LeaderElectionID:       "f8b21459.ocm.software",
+		Controller: controllerconfig.Controller{
+			GroupKindConcurrency: groupKindConcurrency,
+		},
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")
 		os.Exit(1)
 	}
 
+	ociRegistryAddr := f.ociRegistryAddr
 	if v, found := os.LookupEnv("OCI_REGISTRY_LOCALHOST"); found {
 		ociRegistryAddr = v
 	}
 
 	if pprofEnabled {
-		setupLog.Info("starting pprof server", "address", pprofAddr)
-		startPprof(pprofAddr)
+		setupLog.Info("starting pprof server", "address", f.pprofAddr)
+		startPprof(f.pprofAddr)
 	}
 
-	setupManagers(ociRegistryAddr, mgr, ociRegistryNamespace, ociRegistryCertSecretName, ociRegistryInsecureSkipVerify, restConfig, eventsAddr)
+	setupManagers(ociRegistryAddr, mgr, f.ociRegistryNamespace, f.ociRegistryCertSecretName, f.ociRegistryInsecureSkipVerify, restConfig, f.eventsAddr)
 
 	//+kubebuilder:scaffold:builder
 

pkg/ocm/ocm_test.go

@@ -10,11 +10,11 @@ import (
 	"strings"
 	"testing"
 
-	"go.podman.io/image/v5/pkg/compression"
 	_ "github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
 	"github.com/fluxcd/pkg/apis/meta"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.podman.io/image/v5/pkg/compression"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"ocm.software/ocm/api/ocm"