From 5cbbfd1af167a94b0536c6e718c5b34a5461463f Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Thu, 15 Oct 2026 11:16:21 -0700 Subject: [PATCH 1/9] feat: configurable per-controller concurrency Add a --controller-concurrency flag taking comma-separated name=count pairs so each controller's worker count can be tuned independently. Values feed controller-runtime's GroupKindConcurrency; unspecified controllers keep their default. Invalid input fails fast at startup. The Helm chart exposes this via manager.concurrency with equivalent validation at template-render time. Signed-off-by: sleungcy-sap --- concurrency.go | 47 +++++++++ concurrency_test.go | 119 +++++++++++++++++++++++ deploy/templates/deployment_manager.yaml | 14 +++ deploy/values.yaml | 10 ++ main.go | 19 ++++ 5 files changed, 209 insertions(+) create mode 100644 concurrency.go create mode 100644 concurrency_test.go diff --git a/concurrency.go b/concurrency.go new file mode 100644 index 00000000..2c9880e1 --- /dev/null +++ b/concurrency.go @@ -0,0 +1,47 @@ +package main + +import ( + "fmt" + "strconv" + "strings" + + "github.com/open-component-model/ocm-controller/api/v1alpha1" +) + +func parseControllerConcurrency(value string) (map[string]int, error) { + if strings.TrimSpace(value) == "" { + return nil, nil + } + + kinds := map[string]string{ + "componentVersion": "ComponentVersion", + "resource": "Resource", + "snapshot": "Snapshot", + "localization": "Localization", + "configuration": "Configuration", + "fluxDeployer": "FluxDeployer", + } + concurrency := make(map[string]int) + for _, entry := range strings.Split(value, ",") { + name, count, ok := strings.Cut(strings.TrimSpace(entry), "=") + if !ok { + return nil, fmt.Errorf("expected controller=count, got %q", entry) + } + name = strings.TrimSpace(name) + kind, ok := kinds[name] + if !ok { + return nil, fmt.Errorf("unknown controller %q", name) + } + key := v1alpha1.GroupVersion.WithKind(kind).GroupKind().String() + if _, exists := concurrency[key]; exists { + return nil, fmt.Errorf("duplicate controller %q", name) + } + workers, err := strconv.Atoi(strings.TrimSpace(count)) + if err != nil || workers < 1 { + return nil, fmt.Errorf("concurrency for controller %q must be a positive integer, got %q", name, count) + } + concurrency[key] = workers + } + + return concurrency, nil +} diff --git a/concurrency_test.go b/concurrency_test.go new file mode 100644 index 00000000..60d924e0 --- /dev/null +++ b/concurrency_test.go @@ -0,0 +1,119 @@ +package main + +import ( + "os/exec" + "strings" + "testing" + + "github.com/stretchr/testify/require" + appsv1 "k8s.io/api/apps/v1" + "sigs.k8s.io/yaml" +) + +func TestParseControllerConcurrency(t *testing.T) { + tests := []struct { + name string + value string + want map[string]int + }{ + {name: "omitted"}, + {name: "empty", value: " "}, + { + name: "partial with whitespace", + value: " componentVersion = 4, resource=2 ", + want: map[string]int{ + "ComponentVersion.delivery.ocm.software": 4, + "Resource.delivery.ocm.software": 2, + }, + }, + { + name: "all controllers", + value: "componentVersion=2,resource=3,snapshot=4,localization=5,configuration=6,fluxDeployer=7", + want: map[string]int{ + "ComponentVersion.delivery.ocm.software": 2, + "Resource.delivery.ocm.software": 3, + "Snapshot.delivery.ocm.software": 4, + "Localization.delivery.ocm.software": 5, + "Configuration.delivery.ocm.software": 6, + "FluxDeployer.delivery.ocm.software": 7, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := parseControllerConcurrency(tt.value) + require.NoError(t, err) + require.Equal(t, tt.want, got) + }) + } +} + +func TestParseControllerConcurrencyInvalid(t *testing.T) { + for _, value := range []string{ + "unknown=2", "resource", "resource=", "=2", "resource=0", + "resource=-1", "resource=1.5", "resource=two", "resource=2=3", + "resource=2,resource=3", "resource=2,", "resource=999999999999999999999999", + } { + t.Run(value, func(t *testing.T) { + _, err := parseControllerConcurrency(value) + require.Error(t, err) + }) + } +} + +func TestHelmControllerConcurrency(t *testing.T) { + helm, err := exec.LookPath("helm") + if err != nil { + t.Skip("helm is required for chart rendering tests") + } + render := func(t *testing.T, values string) ([]byte, error) { + t.Helper() + cmd := exec.Command(helm, "template", "test", "./deploy", "--show-only", "templates/deployment_manager.yaml", "--values", "-") + cmd.Stdin = strings.NewReader(values) + return cmd.CombinedOutput() + } + baseline, err := render(t, "{}") + require.NoError(t, err, string(baseline)) + require.NotContains(t, string(baseline), "--controller-concurrency") + for _, values := range []string{"manager: {concurrency: {}}", "manager: {concurrency: null}"} { + output, err := render(t, values) + require.NoError(t, err, string(output)) + require.Equal(t, string(baseline), string(output)) + } + + for _, values := range []string{ + "manager: {concurrency: {resource: 2, componentVersion: 4}}", + "manager: {concurrency: {componentVersion: 2, resource: 3, snapshot: 4, localization: 5, configuration: 6, fluxDeployer: 7}}", + } { + output, err := render(t, values) + require.NoError(t, err, string(output)) + var deployment appsv1.Deployment + require.NoError(t, yaml.Unmarshal(output, &deployment)) + var overrides []string + for _, arg := range deployment.Spec.Template.Spec.Containers[0].Args { + if value, ok := strings.CutPrefix(arg, "--controller-concurrency="); ok { + overrides = append(overrides, value) + } + } + require.Len(t, overrides, 1) + parsed, err := parseControllerConcurrency(overrides[0]) + require.NoError(t, err) + if strings.Contains(values, "snapshot") { + require.Len(t, parsed, 6) + } else { + require.Equal(t, "componentVersion=4,resource=2", overrides[0]) + require.Len(t, parsed, 2) + } + } + + for _, values := range []string{ + "manager: {concurrency: {unknown: 2}}", + "manager: {concurrency: {resource: 0}}", + "manager: {concurrency: {resource: -1}}", + "manager: {concurrency: {resource: 1.5}}", + "manager: {concurrency: {resource: true}}", + } { + output, err := render(t, values) + require.Error(t, err, string(output)) + } +} diff --git a/deploy/templates/deployment_manager.yaml b/deploy/templates/deployment_manager.yaml index 5769eaca..b02a9d49 100644 --- a/deploy/templates/deployment_manager.yaml +++ b/deploy/templates/deployment_manager.yaml @@ -32,6 +32,20 @@ spec: args: - --zap-log-level=4 - --leader-elect + {{- with .Values.manager.concurrency }} + {{- $entries := list }} + {{- $controllers := list "componentVersion" "resource" "snapshot" "localization" "configuration" "fluxDeployer" }} + {{- range $name, $count := . }} + {{- if not (has $name $controllers) }} + {{- fail (printf "unknown controller in manager.concurrency: %s" $name) }} + {{- end }} + {{- if not (regexMatch "^[1-9][0-9]*$" (toString $count)) }} + {{- fail (printf "manager.concurrency.%s must be a positive integer" $name) }} + {{- end }} + {{- $entries = append $entries (printf "%s=%v" $name $count) }} + {{- end }} + - {{ printf "--controller-concurrency=%s" (join "," $entries) | quote }} + {{- end }} - --oci-registry-addr={{ .Values.registry.address }} {{- if not .Values.registry.tls.enabled }} - --oci-registry-insecure-skip-verify diff --git a/deploy/values.yaml b/deploy/values.yaml index f2a7d40b..3e9ff805 100644 --- a/deploy/values.yaml +++ b/deploy/values.yaml @@ -64,6 +64,16 @@ registry: manager: replicaCount: 1 + # Optional per-controller worker counts. Unspecified controllers keep their + # existing default of one worker. Counts must be positive integers. + 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 diff --git a/main.go b/main.go index 61f21b63..7e090652 100644 --- a/main.go +++ b/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" @@ -65,6 +66,7 @@ func main() { kubeAPIQPS float64 kubeAPIBurst int kubeAPIRateLimiterDisabled bool + controllerConcurrency string ) flag.StringVar( @@ -132,6 +134,15 @@ func main() { "Disable the client-side Kubernetes API rate limiter and rely on API priority and fairness instead.", ) + flag.StringVar( + &controllerConcurrency, + "controller-concurrency", + "", + "Per-controller worker counts as comma-separated name=count pairs. " + + "Names: componentVersion, resource, snapshot, localization, configuration, fluxDeployer. " + + "Unspecified controllers retain their default concurrency.", + ) + opts := zap.Options{ Development: true, } @@ -139,6 +150,11 @@ func main() { flag.Parse() ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts))) + groupKindConcurrency, err := parseControllerConcurrency(controllerConcurrency) + if err != nil { + setupLog.Error(err, "invalid --controller-concurrency") + os.Exit(1) + } // 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 @@ -163,6 +179,9 @@ func main() { HealthProbeBindAddress: probeAddr, LeaderElection: enableLeaderElection, LeaderElectionID: "f8b21459.ocm.software", + Controller: controllerconfig.Controller{ + GroupKindConcurrency: groupKindConcurrency, + }, }) if err != nil { setupLog.Error(err, "unable to start manager") From 347b9eb5993b93a5ae0f02976ea80656b11052b4 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Fri, 16 Oct 2026 13:24:51 -0700 Subject: [PATCH 2/9] linter errors Signed-off-by: sleungcy-sap --- concurrency.go | 2 +- main.go | 91 +++++++++++++++++++++++++++----------------------- 2 files changed, 50 insertions(+), 43 deletions(-) diff --git a/concurrency.go b/concurrency.go index 2c9880e1..1d69d2db 100644 --- a/concurrency.go +++ b/concurrency.go @@ -22,7 +22,7 @@ func parseControllerConcurrency(value string) (map[string]int, error) { "fluxDeployer": "FluxDeployer", } concurrency := make(map[string]int) - for _, entry := range strings.Split(value, ",") { + for entry := range strings.SplitSeq(value, ",") { name, count, ok := strings.Cut(strings.TrimSpace(entry), "=") if !ok { return nil, fmt.Errorf("expected controller=count, got %q", entry) diff --git a/main.go b/main.go index 7e090652..a1084cfe 100644 --- a/main.go +++ b/main.go @@ -52,94 +52,94 @@ 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 - controllerConcurrency string - ) +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 + controllerConcurrency string +} +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.StringVar( - &controllerConcurrency, + &f.controllerConcurrency, "controller-concurrency", "", - "Per-controller worker counts as comma-separated name=count pairs. " + - "Names: componentVersion, resource, snapshot, localization, configuration, fluxDeployer. " + + "Per-controller worker counts as comma-separated name=count pairs. "+ + "Names: componentVersion, resource, snapshot, localization, configuration, fluxDeployer. "+ "Unspecified controllers retain their default concurrency.", ) @@ -149,8 +149,14 @@ func main() { opts.BindFlags(flag.CommandLine) flag.Parse() + return f, opts +} + +func main() { + f, opts := bindFlags() + ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts))) - groupKindConcurrency, err := parseControllerConcurrency(controllerConcurrency) + groupKindConcurrency, err := parseControllerConcurrency(f.controllerConcurrency) if err != nil { setupLog.Error(err, "invalid --controller-concurrency") os.Exit(1) @@ -163,21 +169,21 @@ 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, @@ -188,16 +194,17 @@ func main() { 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 From e0f6a77f561022f1b58b95a9eba1f20e20c2ce5f Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Fri, 16 Oct 2026 13:25:07 -0700 Subject: [PATCH 3/9] dockerfile go version does not match go.mod go version Signed-off-by: sleungcy-sap --- Dockerfile | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 2b949fe6..743404ba 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # Build the manager binary -FROM docker.io/library/golang:1.26.4 as builder +FROM docker.io/library/golang:1.26.8 as builder WORKDIR /workspace # Copy the Go Modules manifests @@ -11,6 +11,7 @@ RUN go mod download # Copy the go source COPY main.go main.go +COPY concurrency.go concurrency.go COPY pprof.go pprof.go COPY pprof_disabled.go pprof_disabled.go COPY api/ api/ @@ -19,7 +20,8 @@ COPY pkg/ pkg/ # Build ARG GO_TAGS="" -RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -tags "${GO_TAGS}" -a -o manager . +ARG GOARCH=amd64 +RUN CGO_ENABLED=0 GOOS=linux GOARCH=${GOARCH} go build -tags "${GO_TAGS}" -a -o manager . # Use distroless as minimal base image to package the manager binary # Refer to https://github.com/GoogleContainerTools/distroless for more details From c9ce16e8f5a8b3144809eb856ae1d05b29de6993 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Fri, 16 Oct 2026 13:25:15 -0700 Subject: [PATCH 4/9] update stale contributing.md Signed-off-by: sleungcy-sap --- CONTRIBUTING.md | 30 ++++++++++++++++++------------ 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4c16da4a..0cbf52fc 100644 --- a/CONTRIBUTING.md +++ b/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: From 42e29e00bf3be8fbeed0470f9abce86ca8df88a0 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Fri, 16 Oct 2026 15:22:49 -0700 Subject: [PATCH 5/9] docs: document controller concurrency configuration Signed-off-by: sleungcy-sap --- README.md | 18 ++++++++++++++++++ pkg/ocm/ocm_test.go | 2 +- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 3e8e847e..4a2a6288 100644 --- a/README.md +++ b/README.md @@ -441,6 +441,24 @@ 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`. Counts must be positive +integers; controllers you omit keep the default of one worker. This renders the +manager's `--controller-concurrency` flag (comma-separated `name=count` pairs). + ## Uninstall Again, there are two ways to uninstall the controller. Using the ocm cli run `ocm controller uninstall`. diff --git a/pkg/ocm/ocm_test.go b/pkg/ocm/ocm_test.go index edc302ec..e6a02139 100644 --- a/pkg/ocm/ocm_test.go +++ b/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" From 2faefc869a5ed48305b202c7acd8191c3efdee43 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Thu, 22 Oct 2026 12:49:10 -0700 Subject: [PATCH 6/9] remove unused Dockerfile goreleaser.dockerfile is used for CI/release builds; this file is not referenced anywhere. Signed-off-by: sleungcy-sap --- Dockerfile | 34 ---------------------------------- 1 file changed, 34 deletions(-) delete mode 100644 Dockerfile diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 743404ba..00000000 --- a/Dockerfile +++ /dev/null @@ -1,34 +0,0 @@ -# Build the manager binary -FROM docker.io/library/golang:1.26.8 as builder - -WORKDIR /workspace -# Copy the Go Modules manifests -COPY go.mod go.mod -COPY go.sum go.sum -# cache deps before building and copying source so that we don't need to re-download as much -# and so that source changes don't invalidate our downloaded layer -RUN go mod download - -# Copy the go source -COPY main.go main.go -COPY concurrency.go concurrency.go -COPY pprof.go pprof.go -COPY pprof_disabled.go pprof_disabled.go -COPY api/ api/ -COPY controllers/ controllers/ -COPY pkg/ pkg/ - -# Build -ARG GO_TAGS="" -ARG GOARCH=amd64 -RUN CGO_ENABLED=0 GOOS=linux GOARCH=${GOARCH} go build -tags "${GO_TAGS}" -a -o manager . - -# Use distroless as minimal base image to package the manager binary -# Refer to https://github.com/GoogleContainerTools/distroless for more details -FROM gcr.io/distroless/static:nonroot -WORKDIR / -COPY --from=builder /workspace/manager . - -USER 65532:65532 - -ENTRYPOINT ["/manager"] From 8e66a55c8a3459d9888a519d1eb91a79e9a93bd5 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Thu, 22 Oct 2026 14:38:35 -0700 Subject: [PATCH 7/9] feat: replace --controller-concurrency flag with per-controller flags Each controller now has its own --concurrency- flag instead of a single comma-separated --controller-concurrency flag. Unknown flags are rejected at startup, preventing silent misconfiguration from typos. Signed-off-by: sleungcy-sap --- main.go | 33 ++++++++++++++++++++------------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/main.go b/main.go index a1084cfe..7d541307 100644 --- a/main.go +++ b/main.go @@ -65,7 +65,12 @@ type flags struct { kubeAPIQPS float64 kubeAPIBurst int kubeAPIRateLimiterDisabled bool - controllerConcurrency string + concurrencyComponentVersion int + concurrencyResource int + concurrencySnapshot int + concurrencyLocalization int + concurrencyConfiguration int + concurrencyFluxDeployer int } func bindFlags() (flags, zap.Options) { @@ -134,14 +139,12 @@ func bindFlags() (flags, zap.Options) { false, "Disable the client-side Kubernetes API rate limiter and rely on API priority and fairness instead.", ) - flag.StringVar( - &f.controllerConcurrency, - "controller-concurrency", - "", - "Per-controller worker counts as comma-separated name=count pairs. "+ - "Names: componentVersion, resource, snapshot, localization, configuration, fluxDeployer. "+ - "Unspecified controllers retain their default concurrency.", - ) + 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, @@ -156,10 +159,14 @@ func main() { f, opts := bindFlags() ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts))) - groupKindConcurrency, err := parseControllerConcurrency(f.controllerConcurrency) - if err != nil { - setupLog.Error(err, "invalid --controller-concurrency") - os.Exit(1) + 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 From 94f17c928e139acf72aa8a4df05aaf8295805f4c Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Thu, 22 Oct 2026 14:38:39 -0700 Subject: [PATCH 8/9] feat: update helm chart to use per-controller concurrency flags Replace the comma-list --controller-concurrency rendering with individual --concurrency- flags, one per controller. Signed-off-by: sleungcy-sap --- deploy/templates/deployment_manager.yaml | 24 +++++++++++++++--------- deploy/values.yaml | 3 +-- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/deploy/templates/deployment_manager.yaml b/deploy/templates/deployment_manager.yaml index b02a9d49..bcbdc0b1 100644 --- a/deploy/templates/deployment_manager.yaml +++ b/deploy/templates/deployment_manager.yaml @@ -33,18 +33,24 @@ spec: - --zap-log-level=4 - --leader-elect {{- with .Values.manager.concurrency }} - {{- $entries := list }} - {{- $controllers := list "componentVersion" "resource" "snapshot" "localization" "configuration" "fluxDeployer" }} - {{- range $name, $count := . }} - {{- if not (has $name $controllers) }} - {{- fail (printf "unknown controller in manager.concurrency: %s" $name) }} + {{- if .componentVersion }} + - {{ printf "--concurrency-component-version=%v" .componentVersion | quote }} {{- end }} - {{- if not (regexMatch "^[1-9][0-9]*$" (toString $count)) }} - {{- fail (printf "manager.concurrency.%s must be a positive integer" $name) }} + {{- if .resource }} + - {{ printf "--concurrency-resource=%v" .resource | quote }} {{- end }} - {{- $entries = append $entries (printf "%s=%v" $name $count) }} + {{- 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 }} - - {{ printf "--controller-concurrency=%s" (join "," $entries) | quote }} {{- end }} - --oci-registry-addr={{ .Values.registry.address }} {{- if not .Values.registry.tls.enabled }} diff --git a/deploy/values.yaml b/deploy/values.yaml index 3e9ff805..5f5d9a77 100644 --- a/deploy/values.yaml +++ b/deploy/values.yaml @@ -64,8 +64,7 @@ registry: manager: replicaCount: 1 - # Optional per-controller worker counts. Unspecified controllers keep their - # existing default of one worker. Counts must be positive integers. + # Optional per-controller worker counts. Unspecified controllers default to 1. concurrency: {} # concurrency: # componentVersion: 4 From e911a548a5f8f0515bfe178bc29517401eaa53d2 Mon Sep 17 00:00:00 2001 From: sleungcy-sap Date: Thu, 22 Oct 2026 14:38:42 -0700 Subject: [PATCH 9/9] chore: remove concurrency parser and update docs concurrency.go and its tests are no longer needed with per-controller flags. Updated README to remove reference to --controller-concurrency internals. Signed-off-by: sleungcy-sap --- README.md | 5 +- concurrency.go | 47 ----------------- concurrency_test.go | 119 -------------------------------------------- 3 files changed, 2 insertions(+), 169 deletions(-) delete mode 100644 concurrency.go delete mode 100644 concurrency_test.go diff --git a/README.md b/README.md index 4a2a6288..083e6dd0 100644 --- a/README.md +++ b/README.md @@ -455,9 +455,8 @@ manager: ``` Valid controller names: `componentVersion`, `resource`, `snapshot`, -`localization`, `configuration`, `fluxDeployer`. Counts must be positive -integers; controllers you omit keep the default of one worker. This renders the -manager's `--controller-concurrency` flag (comma-separated `name=count` pairs). +`localization`, `configuration`, `fluxDeployer`. Controllers you omit keep the +default of one worker. ## Uninstall diff --git a/concurrency.go b/concurrency.go deleted file mode 100644 index 1d69d2db..00000000 --- a/concurrency.go +++ /dev/null @@ -1,47 +0,0 @@ -package main - -import ( - "fmt" - "strconv" - "strings" - - "github.com/open-component-model/ocm-controller/api/v1alpha1" -) - -func parseControllerConcurrency(value string) (map[string]int, error) { - if strings.TrimSpace(value) == "" { - return nil, nil - } - - kinds := map[string]string{ - "componentVersion": "ComponentVersion", - "resource": "Resource", - "snapshot": "Snapshot", - "localization": "Localization", - "configuration": "Configuration", - "fluxDeployer": "FluxDeployer", - } - concurrency := make(map[string]int) - for entry := range strings.SplitSeq(value, ",") { - name, count, ok := strings.Cut(strings.TrimSpace(entry), "=") - if !ok { - return nil, fmt.Errorf("expected controller=count, got %q", entry) - } - name = strings.TrimSpace(name) - kind, ok := kinds[name] - if !ok { - return nil, fmt.Errorf("unknown controller %q", name) - } - key := v1alpha1.GroupVersion.WithKind(kind).GroupKind().String() - if _, exists := concurrency[key]; exists { - return nil, fmt.Errorf("duplicate controller %q", name) - } - workers, err := strconv.Atoi(strings.TrimSpace(count)) - if err != nil || workers < 1 { - return nil, fmt.Errorf("concurrency for controller %q must be a positive integer, got %q", name, count) - } - concurrency[key] = workers - } - - return concurrency, nil -} diff --git a/concurrency_test.go b/concurrency_test.go deleted file mode 100644 index 60d924e0..00000000 --- a/concurrency_test.go +++ /dev/null @@ -1,119 +0,0 @@ -package main - -import ( - "os/exec" - "strings" - "testing" - - "github.com/stretchr/testify/require" - appsv1 "k8s.io/api/apps/v1" - "sigs.k8s.io/yaml" -) - -func TestParseControllerConcurrency(t *testing.T) { - tests := []struct { - name string - value string - want map[string]int - }{ - {name: "omitted"}, - {name: "empty", value: " "}, - { - name: "partial with whitespace", - value: " componentVersion = 4, resource=2 ", - want: map[string]int{ - "ComponentVersion.delivery.ocm.software": 4, - "Resource.delivery.ocm.software": 2, - }, - }, - { - name: "all controllers", - value: "componentVersion=2,resource=3,snapshot=4,localization=5,configuration=6,fluxDeployer=7", - want: map[string]int{ - "ComponentVersion.delivery.ocm.software": 2, - "Resource.delivery.ocm.software": 3, - "Snapshot.delivery.ocm.software": 4, - "Localization.delivery.ocm.software": 5, - "Configuration.delivery.ocm.software": 6, - "FluxDeployer.delivery.ocm.software": 7, - }, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got, err := parseControllerConcurrency(tt.value) - require.NoError(t, err) - require.Equal(t, tt.want, got) - }) - } -} - -func TestParseControllerConcurrencyInvalid(t *testing.T) { - for _, value := range []string{ - "unknown=2", "resource", "resource=", "=2", "resource=0", - "resource=-1", "resource=1.5", "resource=two", "resource=2=3", - "resource=2,resource=3", "resource=2,", "resource=999999999999999999999999", - } { - t.Run(value, func(t *testing.T) { - _, err := parseControllerConcurrency(value) - require.Error(t, err) - }) - } -} - -func TestHelmControllerConcurrency(t *testing.T) { - helm, err := exec.LookPath("helm") - if err != nil { - t.Skip("helm is required for chart rendering tests") - } - render := func(t *testing.T, values string) ([]byte, error) { - t.Helper() - cmd := exec.Command(helm, "template", "test", "./deploy", "--show-only", "templates/deployment_manager.yaml", "--values", "-") - cmd.Stdin = strings.NewReader(values) - return cmd.CombinedOutput() - } - baseline, err := render(t, "{}") - require.NoError(t, err, string(baseline)) - require.NotContains(t, string(baseline), "--controller-concurrency") - for _, values := range []string{"manager: {concurrency: {}}", "manager: {concurrency: null}"} { - output, err := render(t, values) - require.NoError(t, err, string(output)) - require.Equal(t, string(baseline), string(output)) - } - - for _, values := range []string{ - "manager: {concurrency: {resource: 2, componentVersion: 4}}", - "manager: {concurrency: {componentVersion: 2, resource: 3, snapshot: 4, localization: 5, configuration: 6, fluxDeployer: 7}}", - } { - output, err := render(t, values) - require.NoError(t, err, string(output)) - var deployment appsv1.Deployment - require.NoError(t, yaml.Unmarshal(output, &deployment)) - var overrides []string - for _, arg := range deployment.Spec.Template.Spec.Containers[0].Args { - if value, ok := strings.CutPrefix(arg, "--controller-concurrency="); ok { - overrides = append(overrides, value) - } - } - require.Len(t, overrides, 1) - parsed, err := parseControllerConcurrency(overrides[0]) - require.NoError(t, err) - if strings.Contains(values, "snapshot") { - require.Len(t, parsed, 6) - } else { - require.Equal(t, "componentVersion=4,resource=2", overrides[0]) - require.Len(t, parsed, 2) - } - } - - for _, values := range []string{ - "manager: {concurrency: {unknown: 2}}", - "manager: {concurrency: {resource: 0}}", - "manager: {concurrency: {resource: -1}}", - "manager: {concurrency: {resource: 1.5}}", - "manager: {concurrency: {resource: true}}", - } { - output, err := render(t, values) - require.Error(t, err, string(output)) - } -}