feat: add configurable stack concurrency limit

Add support for limiting concurrent stack reconciliations to prevent
cluster overload. This feature uses the native controller-runtime
MaxConcurrentReconciles mechanism.

Key changes:
- Add MaxConcurrentReconciles support in core reconciler
- Create GetStackConcurrency() to read STACK_MAX_CONCURRENT env var
- Configure Helm chart with stackMaxConcurrent parameter (default: 5)
- Add comprehensive Helm documentation in STACK_CONCURRENCY.md
- Minor formatting fixes in unrelated files

The default limit of 5 concurrent reconciliations provides a good
balance for most clusters. Can be configured via Helm values or set
to 0 for unlimited concurrency.

Author: flemzord

api/formance.com/v1beta1/benthos_types.go

@@ -37,7 +37,7 @@ type BenthosSpec struct {
 	//+optional
 	Batching *Batching `json:"batching,omitempty"`
 	//+optional
-	InitContainers   []corev1.Container `json:"initContainers"`
+	InitContainers   []corev1.Container            `json:"initContainers"`
 	ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty"`
 }
 

helm/operator/STACK_CONCURRENCY.md

@@ -0,0 +1,242 @@
+# Stack Concurrency Configuration
+
+## Overview
+
+Control the number of Stack reconciliations that run in parallel to prevent cluster overload.
+
+## Configuration
+
+### Via Helm Values
+
+Edit your `values.yaml` or use `--set`:
+
+```yaml
+operator:
+  stackMaxConcurrent: 5  # Max 5 concurrent stack reconciliations
+```
+
+Or with Helm command:
+
+```bash
+helm install operator ./helm/operator \
+  --set operator.stackMaxConcurrent=5
+```
+
+### Default Behavior
+
+- **Default value: `5`** (good balance for most clusters)
+- Set to `0` to disable the limit (unlimited concurrency)
+
+## Recommended Values
+
+| Cluster Size | Stacks | CPU | Recommended Value |
+|-------------|--------|-----|-------------------|
+| Small | 1-10 | 2-4 CPU | `2-3` |
+| Medium | 10-30 | 4-8 CPU | `5` |
+| Large | 30-100 | 8-16 CPU | `10` |
+| XL | 100+ | 16+ CPU | `20` |
+
+## Examples
+
+### Example 1: Small Cluster
+
+```yaml
+# values.yaml
+operator:
+  stackMaxConcurrent: 3
+```
+
+```bash
+helm upgrade operator ./helm/operator -f values.yaml
+```
+
+### Example 2: Production Cluster
+
+```yaml
+# values-prod.yaml
+operator:
+  stackMaxConcurrent: 10
+  enableLeaderElection: true
+  region: "eu-west-1"
+  env: "production"
+```
+
+```bash
+helm upgrade operator ./helm/operator -f values-prod.yaml
+```
+
+### Example 3: Override with --set
+
+```bash
+helm upgrade operator ./helm/operator \
+  --set operator.stackMaxConcurrent=5 \
+  --set operator.region=us-east-1
+```
+
+## How It Works
+
+1. The Helm chart sets the `STACK_MAX_CONCURRENT` environment variable
+2. The operator reads this value on startup
+3. Stack reconciliations are limited to N concurrent executions
+4. Additional stacks are queued automatically by Kubernetes
+
+### Behavior
+
+**Without limit (default):**
+```
+Stack A ──┐
+Stack B ──┤
+Stack C ──┼─> All processed in parallel
+Stack D ──┤
+Stack E ──┘
+```
+
+**With limit of 5:**
+```
+Stack A ──┐
+Stack B ──┤
+Stack C ──┼─> Max 5 in parallel
+Stack D ──┤
+Stack E ──┘
+Stack F ──> Queued (waiting)
+Stack G ──> Queued (waiting)
+```
+
+## Verification
+
+Check if the environment variable is set:
+
+```bash
+# Get the pod name
+POD=$(kubectl get pods -n formance-system -l control-plane=formance-controller-manager -o jsonpath='{.items[0].metadata.name}')
+
+# Check environment variables
+kubectl exec -n formance-system $POD -- env | grep STACK_MAX_CONCURRENT
+```
+
+Expected output:
+```
+STACK_MAX_CONCURRENT=5
+```
+
+## Troubleshooting
+
+### Value not applied
+
+1. **Check Helm values:**
+   ```bash
+   helm get values operator -n formance-system
+   ```
+
+2. **Verify deployment:**
+   ```bash
+   kubectl get deployment operator-manager -n formance-system -o yaml | grep -A 2 "STACK_MAX_CONCURRENT"
+   ```
+
+3. **Restart pods to apply changes:**
+   ```bash
+   kubectl rollout restart deployment operator-manager -n formance-system
+   ```
+
+### Performance Issues
+
+**Too many concurrent reconciliations:**
+- Symptoms: High CPU/memory, slow reconciliations
+- Solution: Lower the value (e.g., from 10 to 5)
+
+**Too few concurrent reconciliations:**
+- Symptoms: Long queue times, slow stack deployments
+- Solution: Increase the value (e.g., from 5 to 10)
+
+## Monitoring
+
+### Check current stack status
+
+```bash
+# Total stacks
+kubectl get stacks --all-namespaces --no-headers | wc -l
+
+# Ready stacks
+kubectl get stacks --all-namespaces -o json | jq '[.items[] | select(.status.ready==true)] | length'
+
+# Not ready stacks (in queue or processing)
+kubectl get stacks --all-namespaces -o json | jq '[.items[] | select(.status.ready==false)] | length'
+```
+
+### Operator logs
+
+```bash
+kubectl logs -n formance-system -l control-plane=formance-controller-manager -f
+```
+
+## Advanced Usage
+
+### Environment-specific values
+
+```yaml
+# values-dev.yaml
+operator:
+  stackMaxConcurrent: 2
+  env: "dev"
+
+# values-staging.yaml
+operator:
+  stackMaxConcurrent: 5
+  env: "staging"
+
+# values-prod.yaml
+operator:
+  stackMaxConcurrent: 10
+  env: "production"
+```
+
+Deploy:
+```bash
+helm upgrade operator ./helm/operator -f values-prod.yaml
+```
+
+### ArgoCD / GitOps
+
+```yaml
+# argocd-application.yaml
+apiVersion: argoproj.io/v1alpha1
+kind: Application
+metadata:
+  name: formance-operator
+spec:
+  source:
+    helm:
+      values: |
+        operator:
+          stackMaxConcurrent: 5
+          region: "eu-west-1"
+          env: "production"
+```
+
+## Technical Details
+
+### Implementation
+
+- **Environment Variable:** `STACK_MAX_CONCURRENT`
+- **Read by:** `internal/resources/stacks/config.go::GetStackConcurrency()`
+- **Applied in:** `internal/resources/stacks/init.go`
+- **Uses:** Native controller-runtime `MaxConcurrentReconciles`
+
+### Source Code
+
+```go
+// internal/resources/stacks/config.go
+func GetStackConcurrency() int {
+    if v := os.Getenv("STACK_MAX_CONCURRENT"); v != "" {
+        if n, err := strconv.Atoi(v); err == nil && n > 0 {
+            return n
+        }
+    }
+    return 0 // Default: unlimited
+}
+```
+
+## Related Documentation
+
+- [How to Limit Concurrent Stacks](../../docs/HOW_TO_LIMIT_CONCURRENT_STACKS.md)
+- [Concurrent Limit Implementation](../../CONCURRENT_LIMIT_IMPLEMENTATION.md)

helm/operator/templates/deployment.yaml

@@ -81,6 +81,11 @@ spec:
               port: {{ regexReplaceAll ":" .Values.operator.probeAddr "" | default "8081" }}
             initialDelaySeconds: 5
             periodSeconds: 10
+          {{- if .Values.operator.stackMaxConcurrent }}
+          env:
+            - name: STACK_MAX_CONCURRENT
+              value: {{ .Values.operator.stackMaxConcurrent | quote }}
+          {{- end }}
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           {{- if .Values.webhooks.enabled }}

helm/operator/values.yaml

@@ -49,6 +49,12 @@ operator:
   # Enable leader election for controller manager. Enabling this will ensure there is only one active controller manager.
   enableLeaderElection: true
 
+  # Maximum number of concurrent stack reconciliations
+  # Set to 0 for unlimited
+  # Recommended values: 5 for small/medium clusters, 10 for large, 20 for XL
+  # @section -- Stack Concurrency
+  stackMaxConcurrent: 5
+
   utils:
     tag: ""
 

internal/core/reconciler.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	controllerconfig "sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"k8s.io/client-go/util/workqueue"
@@ -61,10 +62,11 @@ type finalizerConfig[T client.Object] struct {
 }
 
 type ReconcilerOptions[T client.Object] struct {
-	Owns       map[client.Object][]builder.OwnsOption
-	Watchers   map[client.Object]ReconcilerOptionsWatch
-	Finalizers []finalizerConfig[T]
-	Raws       []func(Context, *builder.Builder) error
+	Owns                    map[client.Object][]builder.OwnsOption
+	Watchers                map[client.Object]ReconcilerOptionsWatch
+	Finalizers              []finalizerConfig[T]
+	Raws                    []func(Context, *builder.Builder) error
+	MaxConcurrentReconciles int
 }
 
 type ReconcilerOption[T client.Object] func(*ReconcilerOptions[T])
@@ -81,6 +83,12 @@ func WithRaw[T client.Object](fn func(Context, *builder.Builder) error) Reconcil
 	}
 }
 
+func WithMaxConcurrentReconciles[T client.Object](max int) ReconcilerOption[T] {
+	return func(options *ReconcilerOptions[T]) {
+		options.MaxConcurrentReconciles = max
+	}
+}
+
 func BuildReconcileRequests(ctx context.Context, client client.Client, scheme *runtime.Scheme, target client.Object, opts ...client.ListOption) []reconcile.Request {
 	kinds, _, err := scheme.ObjectKinds(target)
 	if err != nil {
@@ -226,6 +234,13 @@ func withReconciler[T client.Object](controller ObjectController[T], opts ...Rec
 			}
 		}
 
+		// Appliquer MaxConcurrentReconciles si spécifié
+		if options.MaxConcurrentReconciles > 0 {
+			b = b.WithOptions(controllerconfig.Options{
+				MaxConcurrentReconciles: options.MaxConcurrentReconciles,
+			})
+		}
+
 		return b.Complete(reconcile.Func(reconcileObject(mgr, controller, options)))
 	}
 }

internal/resources/orchestrations/deployments.go

@@ -184,7 +184,7 @@ func createDeployment(
 		Spec: appsv1.DeploymentSpec{
 			Template: corev1.PodTemplateSpec{
 				Spec: corev1.PodSpec{
-					ImagePullSecrets: imageConfiguration.PullSecrets,
+					ImagePullSecrets:   imageConfiguration.PullSecrets,
 					ServiceAccountName: serviceAccountName,
 					Containers: []corev1.Container{{
 						Name:          "api",

internal/resources/payments/deployments.go

@@ -290,7 +290,7 @@ func createFullDeployment(
 	return nil
 }
 
-func createWorkerDeployment(ctx core.Context, stack *v1beta1.Stack, payments *v1beta1.Payments, database *v1beta1.Database, imageConfiguration *registries.ImageConfiguration, env []v1.EnvVar, appOpts applications.ProbeOpts, ) error {
+func createWorkerDeployment(ctx core.Context, stack *v1beta1.Stack, payments *v1beta1.Payments, database *v1beta1.Database, imageConfiguration *registries.ImageConfiguration, env []v1.EnvVar, appOpts applications.ProbeOpts) error {
 	serviceAccountName, err := settings.GetAWSServiceAccount(ctx, stack.Name)
 	if err != nil {
 		return err

internal/resources/stacks/config.go

@@ -0,0 +1,18 @@
+package stacks
+
+import (
+	"os"
+	"strconv"
+)
+
+// GetStackConcurrency returns the maximum number of concurrent stack reconciliations
+// from the STACK_MAX_CONCURRENT environment variable, or a default value of 5
+func GetStackConcurrency() int {
+	if v := os.Getenv("STACK_MAX_CONCURRENT"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n >= 0 {
+			return n
+		}
+	}
+	// Default: 5 concurrent reconciliations (good balance for most clusters)
+	return 5
+}

internal/resources/stacks/init.go

@@ -366,6 +366,7 @@ func init() {
 		}),
 		WithStdReconciler(Reconcile,
 			WithOwn[*v1beta1.Stack](&corev1.Namespace{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})),
+			WithMaxConcurrentReconciles[*v1beta1.Stack](GetStackConcurrency()),
 			WithRaw[*v1beta1.Stack](func(ctx Context, b *builder.Builder) error {
 				for _, rtype := range ctx.GetScheme().AllKnownTypes() {
 					v := reflect.New(rtype).Interface()

pkg/client/formance.com/v1beta1/databases.go

@@ -51,4 +51,4 @@ func (c *databasesClient) Watch(ctx context.Context, opts metav1.ListOptions) (w
 		Resource("Databases").
 		VersionedParams(&opts, scheme.ParameterCodec).
 		Watch(ctx)
-}
+}