GKE high-concurrency worker-imports (128-256 threads)

README.md

@@ -96,7 +96,7 @@ The chart produces **one `config.yml`** (rendered into a ConfigMap and mounted a
 
 ### Sizing
 
-Defaults are minimal for dev / smoke testing. For production, scale per-component:
+Defaults are minimal for dev / smoke testing. For production, scale per-component. On GKE, use [`examples/gke/values-gke-high-concurrency.yaml`](examples/gke/values-gke-high-concurrency.yaml) for a 128-thread import/eval pool (4 pods × 32 threads, HPA to 8 pods / 256 threads):
 
 ```yaml
 efficientai:
@@ -115,17 +115,17 @@ efficientai:
       minAvailable: 2
 
   worker:
-    replicaCount: 2
+    replicaCount: 1
+    queues: celery,audio-metrics
+    concurrency: 8
     resources:
       limits:   { cpu: "2", memory: "8Gi" }
-      requests: { cpu: "1", memory: "4Gi" }
+      requests: { cpu: "2", memory: "4Gi" }
 
   workerImports:
-    replicaCount: 1
-    concurrency: 8
-    resources:
-      limits:   { cpu: "2", memory: "4Gi" }
-      requests: { cpu: "500m", memory: "2Gi" }
+    replicaCount: 4
+    pool: threads
+    concurrency: 32
 
 postgresql:
   primary:
@@ -154,6 +154,7 @@ Every example in `examples/` is a values overlay or manifest. See [`examples/REA
 | `external-redis.yaml` | External standalone Redis and external Redis Cluster (e.g. AWS ElastiCache config endpoint) |
 | `external-s3.yaml` | External S3 bucket with credentials sourced via `secretKeyRef` |
 | `gke/values-gcs.yaml` | GKE + GCS via Workload Identity, observability config, GCE Ingress |
+| `gke/values-gke-high-concurrency.yaml` | GKE overlay: 4–8 worker-imports pods × 32 threads (128–256 Celery threads) |
 | `ingress-alb.yaml` | AWS Load Balancer Controller with redirect action and custom per-host backend |
 | `sso-oidc.yaml` | External OIDC SSO (Okta-style) wired through `efficientai.web.additionalEnv` |
 | `topology-spread.yaml` | Zone- and host-aware spread constraints for `web`, `worker`, and `workerImports` |

charts/efficientai/README.md

@@ -6,7 +6,7 @@ This chart deploys:
 
 - **web** — the FastAPI API (also serves the built frontend) — `Deployment`, `Service`, optional `Ingress`, `HPA`, `PDB`.
 - **worker** — a Celery worker for the default queue — `Deployment` (optional `HPA`, `PDB`).
-- **worker-imports** — a dedicated Celery worker for the `imports`, `diarization`, and `evaluations` queues (concurrency 8 by default) — `Deployment`.
+- **worker-imports** — a dedicated Celery worker for the `imports`, `diarization`, and `evaluations` queues (thread pool, concurrency 32 by default) — `Deployment`.
 - **postgresql** (optional, Bitnami subchart) — toggleable via `postgresql.deploy`.
 - **redis** (optional, Bitnami subchart) — toggleable via `redis.deploy`. Cluster mode supported via external endpoints.
 
@@ -134,15 +134,28 @@ Every component exposes the same surface:
 | `efficientai.web.ingress.tls` | `[]` |
 | `efficientai.web.probes.liveness` / `probes.readiness` | HTTP `/api/v1/health` on port `8000` |
 
+#### Worker-only
+
+| Key | Default |
+|---|---|
+| `efficientai.worker.queues` | `""` (no `--queues`; drains all queues when worker-imports is disabled) |
+| `efficientai.worker.concurrency` | `0` (omit `--concurrency` unless set > 0) |
+
+When `efficientai.workerImports.enabled=true` and `efficientai.worker.command` is empty, set `efficientai.worker.queues` / `.concurrency` to split pools (docker-compose default: `celery,audio-metrics` @ `8`). When worker-imports is disabled, leave both empty so the default worker continues draining every queue.
+
 #### Worker-imports-only
 
 | Key | Default |
 |---|---|
 | `efficientai.workerImports.enabled` | `true` |
 | `efficientai.workerImports.queues` | `imports,diarization,evaluations` |
+| `efficientai.workerImports.pool` | `""` (omit `--pool`; set to `threads` for I/O-bound work) |
 | `efficientai.workerImports.concurrency` | `8` |
+| `efficientai.workerImports.spreadAcrossNodes` | `false` (release-scoped host spread when `true`) |
+
+If `efficientai.workerImports.command` is left empty, the chart builds `eai worker --config /app/config.yml --loglevel info --queues <queues> [--pool <pool>] --concurrency <n>`. Requires `efficientai-worker` **>= 1.5.0** (images with `eai worker --queues/--pool/--concurrency`). For older images, set `efficientai.workerImports.command` to invoke `celery -A app.workers.celery_app worker ...` directly.
 
-If `efficientai.workerImports.command` is left empty, the chart builds `celery -A app.workers.celery_app worker --loglevel=info --queues=<queues> --concurrency=<concurrency>` automatically. (We invoke `celery` directly rather than `eai worker --queues ... --concurrency ...` so the queue / concurrency flags work across all `efficientai-worker` image versions — older images' `eai worker` CLI doesn't expose those flags. Celery picks up the broker URL from the `CELERY_BROKER_URL` env the pod already exports.)
+For a 128-thread import/eval pool on GKE, use [examples/gke/values-gke-high-concurrency.yaml](../../examples/gke/values-gke-high-concurrency.yaml) (4 pods × 32 threads, HPA to 8 pods).
 
 ### Postgres (`postgresql.*`)
 

charts/efficientai/templates/NOTES.txt

@@ -43,3 +43,19 @@ To reach it from your machine:
 WARNING: efficientai.secretKey is empty. Generate one and re-run helm upgrade:
   openssl rand -hex 32
 {{- end }}
+
+{{- if .Values.efficientai.workerImports.enabled }}
+{{- if ge (int .Values.efficientai.workerImports.concurrency) 16 }}
+
+Worker-imports concurrency is {{ .Values.efficientai.workerImports.concurrency }} per pod.
+Ensure Postgres/Redis connection pools are sized for replicas × concurrency.
+For the 128-thread GKE overlay, use examples/gke/values-gke-high-concurrency.yaml.
+{{- end }}
+{{- end }}
+
+{{- if .Values.efficientai.workerImports.enabled }}
+Default worker drains: {{ default "celery,audio-metrics" .Values.efficientai.worker.queues }} (set efficientai.worker.queues / .concurrency).
+Worker-imports drains: {{ .Values.efficientai.workerImports.queues }}.
+Requires efficientai-worker image with `eai worker --queues/--pool/--concurrency` support (>= 1.5.0).
+For older images, set efficientai.workerImports.command to invoke celery directly.
+{{- end }}

charts/efficientai/templates/_helpers.tpl

@@ -298,3 +298,47 @@ Common application env block (secret_key, encryption_key, license, HF token, blo
 - name: FRONTEND_DIR
   value: "/app/frontend/dist"
 {{- end -}}
+
+{{/*
+Build the default worker container command unless efficientai.worker.command is set.
+Usage: {{- include "efficientai.worker.command" . | nindent 12 }}
+*/}}
+{{- define "efficientai.worker.command" -}}
+{{- $w := .Values.efficientai.worker -}}
+{{- if gt (len $w.command) 0 -}}
+{{ toYaml $w.command }}
+{{- else -}}
+{{- $args := list "eai" "worker" "--config" "/app/config.yml" "--loglevel" "info" -}}
+{{- /* Only restrict queues when a dedicated worker-imports pool is enabled. */ -}}
+{{- if .Values.efficientai.workerImports.enabled -}}
+{{- if $w.queues -}}
+{{- $args = concat $args (list "--queues" $w.queues) -}}
+{{- end -}}
+{{- if $w.concurrency -}}
+{{- $args = concat $args (list "--concurrency" (printf "%d" (int $w.concurrency))) -}}
+{{- end -}}
+{{- end -}}
+{{ toYaml $args }}
+{{- end -}}
+{{- end -}}
+
+{{/*
+Build the default worker-imports container command unless efficientai.workerImports.command is set.
+Usage: {{- include "efficientai.workerImports.command" . | nindent 12 }}
+*/}}
+{{- define "efficientai.workerImports.command" -}}
+{{- $wi := .Values.efficientai.workerImports -}}
+{{- if gt (len $wi.command) 0 -}}
+{{ toYaml $wi.command }}
+{{- else -}}
+{{- $args := list "eai" "worker" "--config" "/app/config.yml" "--loglevel" "info" -}}
+{{- $args = concat $args (list "--queues" (default "imports,diarization,evaluations" $wi.queues)) -}}
+{{- with $wi.pool -}}
+{{- if ne . "" -}}
+{{- $args = concat $args (list "--pool" .) -}}
+{{- end -}}
+{{- end -}}
+{{- $args = concat $args (list "--concurrency" (printf "%d" (int (default 8 $wi.concurrency)))) -}}
+{{ toYaml $args }}
+{{- end -}}
+{{- end -}}

charts/efficientai/templates/worker-imports/deployment.yaml

@@ -2,13 +2,20 @@
 {{- $component := "worker-imports" -}}
 {{- $name := include "efficientai.componentFullname" (dict "root" . "component" $component) -}}
 {{- $wi := .Values.efficientai.workerImports -}}
-{{- /* We invoke celery directly (rather than `eai worker`) so the --queues
-       and --concurrency flags work across all app versions. `eai worker`
-       just shells out to celery internally, and celery picks up the broker
-       URL from the CELERY_BROKER_URL env var that the pod already exports. */ -}}
-{{- $defaultCmd := list "celery" "-A" "app.workers.celery_app" "worker" "--loglevel=info" (printf "--queues=%s" (default "imports,diarization,evaluations" $wi.queues)) (printf "--concurrency=%d" (int (default 8 $wi.concurrency))) -}}
-{{- $cmd := default $defaultCmd $wi.command -}}
-{{- if eq (len $wi.command) 0 -}}{{- $cmd = $defaultCmd -}}{{- end -}}
+{{- $autoSpread := list -}}
+{{- if $wi.spreadAcrossNodes -}}
+{{- $autoSpread = list (dict
+    "maxSkew" 1
+    "topologyKey" "kubernetes.io/hostname"
+    "whenUnsatisfiable" "DoNotSchedule"
+    "labelSelector" (dict "matchLabels" (dict
+      "app.kubernetes.io/name" (include "efficientai.name" .)
+      "app.kubernetes.io/instance" .Release.Name
+      "app.kubernetes.io/component" $component
+    ))
+  ) -}}
+{{- end -}}
+{{- $topologySpreadConstraints := concat $autoSpread ($wi.pod.topologySpreadConstraints | default list) -}}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -65,9 +72,9 @@ spec:
       affinity:
         {{- toYaml . | nindent 8 }}
       {{- end }}
-      {{- with $wi.pod.topologySpreadConstraints }}
+      {{- if gt (len $topologySpreadConstraints) 0 }}
       topologySpreadConstraints:
-        {{- toYaml . | nindent 8 }}
+        {{- toYaml $topologySpreadConstraints | nindent 8 }}
       {{- end }}
       containers:
         - name: worker-imports
@@ -78,7 +85,7 @@ spec:
             {{- toYaml . | nindent 12 }}
           {{- end }}
           command:
-            {{- toYaml $cmd | nindent 12 }}
+            {{- include "efficientai.workerImports.command" . | nindent 12 }}
           env:
             {{- include "efficientai.postgresql.envBlock" . | nindent 12 }}
             {{- include "efficientai.redis.envBlock" . | nindent 12 }}

charts/efficientai/templates/worker/deployment.yaml

@@ -69,7 +69,7 @@ spec:
             {{- toYaml . | nindent 12 }}
           {{- end }}
           command:
-            {{- toYaml .Values.efficientai.worker.command | nindent 12 }}
+            {{- include "efficientai.worker.command" . | nindent 12 }}
           env:
             {{- include "efficientai.postgresql.envBlock" . | nindent 12 }}
             {{- include "efficientai.redis.envBlock" . | nindent 12 }}

charts/efficientai/tests/gcs_config_test.yaml

@@ -15,16 +15,16 @@ tests:
       gcs.prefix: audio/
     asserts:
       - matchRegex:
-          path: data.config\.yml
+          path: data["config.yml"]
           pattern: "blob_provider: gcs"
       - matchRegex:
-          path: data.config\.yml
+          path: data["config.yml"]
           pattern: "gcs:\\n\\s+bucket_name: test-audio-bucket"
       - matchRegex:
-          path: data.config\.yml
+          path: data["config.yml"]
           pattern: "project_id: test-gcp-project"
       - matchRegex:
-          path: data.config\.yml
+          path: data["config.yml"]
           pattern: "s3:\\n\\s+enabled: false"
 
   - it: wires GCS env into web pods

charts/efficientai/tests/worker_deployment_test.yaml

@@ -4,10 +4,11 @@ templates:
   - templates/worker-imports/deployment.yaml
   - templates/configmap.yaml
 tests:
-  - it: renders the worker deployment with worker image and command
+  - it: renders the worker without queue flags when worker-imports is disabled
     template: templates/worker/deployment.yaml
     set:
       efficientai.secretKey.value: "x"
+      efficientai.workerImports.enabled: false
     asserts:
       - hasDocuments:
           count: 1
@@ -16,9 +17,32 @@ tests:
           pattern: "^ghcr.io/efficientai-tech/efficientai-worker:.+"
       - contains:
           path: spec.template.spec.containers[0].command
-          content: --loglevel
+          content: eai
+      - notContains:
+          path: spec.template.spec.containers[0].command
+          content: --queues
+
+  - it: renders queue flags on worker when worker-imports is enabled
+    template: templates/worker/deployment.yaml
+    set:
+      efficientai.secretKey.value: "x"
+      efficientai.worker.queues: celery,audio-metrics
+      efficientai.worker.concurrency: 8
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].command
+          content: --queues
+      - contains:
+          path: spec.template.spec.containers[0].command
+          content: celery,audio-metrics
+      - contains:
+          path: spec.template.spec.containers[0].command
+          content: --concurrency
+      - contains:
+          path: spec.template.spec.containers[0].command
+          content: "8"
 
-  - it: renders the worker-imports deployment invoking celery with --queues/--concurrency by default
+  - it: renders the worker-imports deployment invoking eai worker by default
     template: templates/worker-imports/deployment.yaml
     set:
       efficientai.secretKey.value: "x"
@@ -27,30 +51,36 @@ tests:
           count: 1
       - contains:
           path: spec.template.spec.containers[0].command
-          content: celery
+          content: eai
+      - contains:
+          path: spec.template.spec.containers[0].command
+          content: --queues
       - contains:
           path: spec.template.spec.containers[0].command
-          content: app.workers.celery_app
+          content: imports,diarization,evaluations
       - contains:
           path: spec.template.spec.containers[0].command
-          content: --queues=imports,diarization,evaluations
+          content: --concurrency
       - contains:
           path: spec.template.spec.containers[0].command
-          content: --concurrency=8
+          content: "8"
+      - notContains:
+          path: spec.template.spec.containers[0].command
+          content: --pool
 
   - it: respects custom workerImports.concurrency and queues
     template: templates/worker-imports/deployment.yaml
     set:
       efficientai.secretKey.value: "x"
-      efficientai.workerImports.concurrency: 8
+      efficientai.workerImports.concurrency: 16
       efficientai.workerImports.queues: "imports,heavy"
     asserts:
       - contains:
           path: spec.template.spec.containers[0].command
-          content: "--queues=imports,heavy"
+          content: imports,heavy
       - contains:
           path: spec.template.spec.containers[0].command
-          content: "--concurrency=8"
+          content: "16"
 
   - it: omits worker-imports deployment when disabled
     template: templates/worker-imports/deployment.yaml