diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a32c39cec4..a6e01dee17 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -336,13 +336,22 @@ jobs: -f k8s/helm/commonly/values.yaml \ -f k8s/helm/commonly/values-dev.yaml \ > /tmp/dev-rendered.yaml + helm template commonly-dev-worker k8s/helm/commonly \ + -f k8s/helm/commonly/values.yaml \ + -f k8s/helm/commonly/values-dev.yaml \ + --set backend.worker.enabled=true \ + --set backend.worker.startPaused=false \ + > /tmp/dev-worker-rendered.yaml + + - name: Verify backend worker render + run: bash scripts/verify-backend-worker-chart.sh - name: Lint rendered dev chart run: | kubeconform --strict --summary -output text \ --schema-location "https://raw.githubusercontent.com/datreeio/CRDs-catalog/${CRDS_CATALOG_SHA}/{{.Group}}/{{.ResourceKind}}_{{.ResourceAPIVersion}}.json" \ --schema-location default \ - /tmp/dev-rendered.yaml + /tmp/dev-rendered.yaml /tmp/dev-worker-rendered.yaml # The dev render has kubeconform coverage, but staging is otherwise first # rendered at deploy time. Parse both values combinations before merge. diff --git a/backend/__tests__/unit/helm/opsRunnerEnvMirror.test.js b/backend/__tests__/unit/helm/opsRunnerEnvMirror.test.js index 16b4fc3f87..cae7e56132 100644 --- a/backend/__tests__/unit/helm/opsRunnerEnvMirror.test.js +++ b/backend/__tests__/unit/helm/opsRunnerEnvMirror.test.js @@ -1,12 +1,14 @@ /** * Drift guard for the ops-runner env mirror. * - * templates/ops/ops-runner-deployment.yaml mirrors the backend container's env - * + volume mounts VERBATIM (it can't DRY-extract into a helper without changing + * templates/ops/ops-runner-deployment.yaml mirrors the backend container's app + * dependencies and volume mounts (it can't DRY-extract into a helper without changing * backend's byte-for-byte render — the env block has blank lines nindent would * re-indent). A "KEEP IN SYNC" banner alone drifts (recorded instances across * the chart), so this pins the contract: the two Deployments must declare the - * SAME env entries — the same (name, source) tuples, where source is a + * SAME env entries except RUN_WORKER, which controls the serving process and + * is not read by the sleeping ops-runner — the same (name, source) tuples, + * where source is a * secretKeyRef name+key, a configMapKeyRef name+key, or a literal value. If * backend gains/loses an env var, or changes a var's source, and ops-runner * isn't updated to match, this fails. @@ -93,7 +95,14 @@ if (typeof describe === 'function') { // ops-runner the same number of times, and vice versa. toEqual on the // sorted arrays reports exactly which env entry drifted. test('ops-runner declares the same env (name + source) tuples as backend', () => { - expect([...opsTuples].sort()).toEqual([...backendTuples].sort()); + const withoutWorkerGate = (tuple) => !tuple.startsWith('RUN_WORKER\t'); + expect([...opsTuples].filter(withoutWorkerGate).sort()) + .toEqual([...backendTuples].filter(withoutWorkerGate).sort()); + expect(opsTuples.every(withoutWorkerGate)).toBe(true); + expect(backendTuples.filter((tuple) => !withoutWorkerGate(tuple))).toEqual([ + 'RUN_WORKER\tvalue:"1"', + 'RUN_WORKER\tvalue:"0"', + ]); }); }); } diff --git a/backend/__tests__/unit/server.test.js b/backend/__tests__/unit/server.test.js index 0e56844417..1201869022 100644 --- a/backend/__tests__/unit/server.test.js +++ b/backend/__tests__/unit/server.test.js @@ -48,6 +48,23 @@ jest.mock('../../config/init-pg-db', () => { const initPGDB = mockInitPGDB; return initPGDB; }); +let mockInitPgRetention = jest.fn(); +jest.mock('../../services/pgRetentionService', () => { + const initPgRetention = mockInitPgRetention; + return { initPgRetention }; +}); +let mockInitInstallationCleanup = jest.fn(); +jest.mock('../../services/agentInstallationCleanupService', () => { + const initInstallationCleanup = mockInitInstallationCleanup; + return { initInstallationCleanup }; +}); +let mockInitPgMessageReconcile = jest.fn(); +jest.mock('../../services/pgMessageReconcileService', () => { + const initPgMessageReconcile = mockInitPgMessageReconcile; + return { initPgMessageReconcile }; +}); +jest.mock('../../services/schedulerService', () => ({ start: jest.fn() })); +jest.mock('../../services/discordGatewayService', () => ({ start: jest.fn() })); // Replace pg routes with simple routers jest.mock('../../routes/pg-status', () => { @@ -89,6 +106,9 @@ describe('server pg boot routes', () => { beforeEach(() => { mockConnectPG = jest.fn().mockResolvedValue(null); mockInitPGDB = jest.fn(); + mockInitPgRetention = jest.fn(); + mockInitInstallationCleanup = jest.fn(); + mockInitPgMessageReconcile = jest.fn(); }); afterEach(() => { @@ -96,6 +116,7 @@ describe('server pg boot routes', () => { jest.clearAllMocks(); delete process.env.PG_HOST; delete process.env.PG_BOOT_RETRY_BASE_DELAY_MS; + delete process.env.RUN_WORKER; }); it('mounts the real status router even when PG is not configured', async () => { @@ -161,6 +182,33 @@ describe('server pg boot routes', () => { await request(app).get('/api/pg/messages').expect(404); }); + it.each([ + ['web', '0', false], + ['worker', '1', true], + ['legacy unset', undefined, true], + ])('%s boot mounts PG routes and applies the worker cron gate', async (_label, runWorker, shouldStart) => { + const priorNodeEnv = process.env.NODE_ENV; + process.env.NODE_ENV = 'production'; + process.env.PG_HOST = 'x'; + if (runWorker !== undefined) process.env.RUN_WORKER = runWorker; + mockConnectPG.mockResolvedValue({}); + mockInitPGDB.mockResolvedValue(true); + try { + // This imports the actual server boot composition, including its + // createPgBoot onMounted callback. Only the cron starters are mocked. + // eslint-disable-next-line global-require, import/no-unresolved, import/extensions + const { app } = require('../../server'); + await new Promise((resolve) => { setTimeout(resolve, 30); }); + await request(app).get('/api/pg/messages').expect(200); + expect(mockInitPgRetention).toHaveBeenCalledTimes(shouldStart ? 1 : 0); + expect(mockInitInstallationCleanup).toHaveBeenCalledTimes(shouldStart ? 1 : 0); + expect(mockInitPgMessageReconcile).toHaveBeenCalledTimes(shouldStart ? 1 : 0); + } finally { + if (priorNodeEnv === undefined) delete process.env.NODE_ENV; + else process.env.NODE_ENV = priorNodeEnv; + } + }); + it('wires the readiness probe to the route table, not to the boot flag (TASK-168)', () => { // The states reachable in this file cannot tell the two apart — when PG is // missing, the flag and the route table are both false; when it works, both diff --git a/backend/__tests__/unit/services/backgroundWorkerStart.test.js b/backend/__tests__/unit/services/backgroundWorkerStart.test.js new file mode 100644 index 0000000000..2415d70b2a --- /dev/null +++ b/backend/__tests__/unit/services/backgroundWorkerStart.test.js @@ -0,0 +1,34 @@ +const { startBackgroundServices } = require('../../../services/backgroundWorkerStart'); + +describe('background worker start gate', () => { + let scheduler; + let discordGateway; + + beforeEach(() => { + scheduler = { start: jest.fn() }; + discordGateway = { start: jest.fn() }; + jest.spyOn(console, 'log').mockImplementation(() => {}); + }); + + afterEach(() => jest.restoreAllMocks()); + + test.each([ + ['unset', undefined, true], + ['on', '1', true], + ['off', '0', false], + ])('RUN_WORKER %s', (_label, runWorker, shouldStart) => { + const env = { NODE_ENV: 'production', DISCORD_BOT_TOKEN: 'token' }; + if (runWorker !== undefined) env.RUN_WORKER = runWorker; + + expect(startBackgroundServices(env, { scheduler, discordGateway })).toBe(shouldStart); + expect(scheduler.start).toHaveBeenCalledTimes(shouldStart ? 1 : 0); + expect(discordGateway.start).toHaveBeenCalledTimes(shouldStart ? 1 : 0); + }); + + test('keeps the test environment and tokenless gateway stopped', () => { + expect(startBackgroundServices({ NODE_ENV: 'test', DISCORD_BOT_TOKEN: 'token' }, { scheduler, discordGateway })).toBe(false); + expect(startBackgroundServices({ NODE_ENV: 'production' }, { scheduler, discordGateway })).toBe(true); + expect(scheduler.start).toHaveBeenCalledTimes(1); + expect(discordGateway.start).not.toHaveBeenCalled(); + }); +}); diff --git a/backend/server.ts b/backend/server.ts index 83d9408c34..c22f31b18f 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -383,15 +383,11 @@ mongoose.connection.once('open', () => { // Start the summarizer scheduler const schedulerService = require('./services/schedulerService'); const discordGatewayService = require('./services/discordGatewayService'); - -if (process.env.NODE_ENV !== 'test') { - console.log('Starting summarizer scheduler...'); - schedulerService.start(); - - if (process.env.DISCORD_BOT_TOKEN) { - discordGatewayService.start(); - } -} +const { startBackgroundServices, shouldRunBackgroundServices } = require('./services/backgroundWorkerStart'); +startBackgroundServices(process.env, { + scheduler: schedulerService, + discordGateway: discordGatewayService, +}); // PostgreSQL status is mounted unconditionally, and NOT as a placeholder: // checkStatus reads the pool and the schema itself, so it answers @@ -441,7 +437,7 @@ if (process.env.PG_HOST) { // Kick off the daily 30-day message retention cron. Kept out // of schedulerService.ts on purpose so other tracks can edit // that file without stomping on this cron. - if (process.env.NODE_ENV !== 'test') { + if (shouldRunBackgroundServices(process.env)) { try { const { initPgRetention } = require('./services/pgRetentionService'); initPgRetention(); diff --git a/backend/services/backgroundWorkerStart.ts b/backend/services/backgroundWorkerStart.ts new file mode 100644 index 0000000000..5f59cf18ec --- /dev/null +++ b/backend/services/backgroundWorkerStart.ts @@ -0,0 +1,22 @@ +type BackgroundServices = { + scheduler: { start: () => void }; + discordGateway: { start: () => void }; +}; + +export function startBackgroundServices( + env: NodeJS.ProcessEnv, + services: BackgroundServices, +): boolean { + if (!shouldRunBackgroundServices(env)) return false; + + console.log('Starting summarizer scheduler...'); + services.scheduler.start(); + if (env.DISCORD_BOT_TOKEN) services.discordGateway.start(); + return true; +} + +export function shouldRunBackgroundServices(env: NodeJS.ProcessEnv): boolean { + // Unset keeps single-process installs scheduling until the chart moves the + // singleton work into its own Deployment. + return env.NODE_ENV !== 'test' && env.RUN_WORKER !== '0'; +} diff --git a/k8s/helm/commonly/templates/_helpers.tpl b/k8s/helm/commonly/templates/_helpers.tpl index d01b4cd42e..c62bc7bd8d 100644 --- a/k8s/helm/commonly/templates/_helpers.tpl +++ b/k8s/helm/commonly/templates/_helpers.tpl @@ -64,6 +64,17 @@ Backend selector labels app: backend {{- end }} +{{/* The worker must never match the backend Service selector. */}} +{{- define "commonly.backendWorker.labels" -}} +{{ include "commonly.labels" . }} +app: backend-worker +{{- end }} + +{{- define "commonly.backendWorker.selectorLabels" -}} +{{ include "commonly.selectorLabels" . }} +app: backend-worker +{{- end }} + {{/* Frontend labels */}} diff --git a/k8s/helm/commonly/templates/core/backend-deployment.yaml b/k8s/helm/commonly/templates/core/backend-deployment.yaml index 54aa6bbb6a..61517210b9 100644 --- a/k8s/helm/commonly/templates/core/backend-deployment.yaml +++ b/k8s/helm/commonly/templates/core/backend-deployment.yaml @@ -1,13 +1,18 @@ -{{- if .Values.backend.enabled }} +{{- define "commonly.backend.deployment" -}} apiVersion: apps/v1 kind: Deployment metadata: - name: backend + name: {{ if .worker }}backend-worker{{ else }}backend{{ end }} namespace: {{ include "commonly.namespace" . }} labels: - {{- include "commonly.backend.labels" . | nindent 4 }} + {{- if .worker }}{{ include "commonly.backendWorker.labels" . | nindent 4 }}{{ else }}{{ include "commonly.backend.labels" . | nindent 4 }}{{ end }} spec: - replicas: {{ .Values.backend.replicaCount }} + replicas: {{ if .worker }}{{ if .Values.backend.worker.startPaused }}0{{ else }}1{{ end }}{{ else }}{{ .Values.backend.replicaCount }}{{ end }} + {{- if .worker }} + # A rolling surge would briefly run two schedulers and two Discord gateways. + strategy: + type: Recreate + {{- else }} # Overlap is safe: the catalog is ephemeral, ObjectStore defaults to Mongo, # and request/session state is externalized to Mongo, PostgreSQL, and Redis. minReadySeconds: 10 @@ -16,13 +21,14 @@ spec: rollingUpdate: maxUnavailable: 0 maxSurge: 1 + {{- end }} selector: matchLabels: - {{- include "commonly.backend.selectorLabels" . | nindent 6 }} + {{- if .worker }}{{ include "commonly.backendWorker.selectorLabels" . | nindent 6 }}{{ else }}{{ include "commonly.backend.selectorLabels" . | nindent 6 }}{{ end }} template: metadata: labels: - {{- include "commonly.backend.selectorLabels" . | nindent 8 }} + {{- if .worker }}{{ include "commonly.backendWorker.selectorLabels" . | nindent 8 }}{{ else }}{{ include "commonly.backend.selectorLabels" . | nindent 8 }}{{ end }} spec: serviceAccountName: {{ .Values.rbac.serviceAccount.name }} terminationGracePeriodSeconds: 180 @@ -64,6 +70,13 @@ spec: # Node Environment - name: NODE_ENV value: {{ .Values.backend.env.nodeEnv }} + {{- if .worker }} + - name: RUN_WORKER + value: "1" + {{- else if .Values.backend.worker.enabled }} + - name: RUN_WORKER + value: "0" + {{- end }} - name: PORT value: {{ .Values.backend.service.port | quote }} - name: HOSTED_MCP_HIDDEN_ENTRIES @@ -693,11 +706,17 @@ spec: periodSeconds: {{ .Values.backend.readinessProbe.periodSeconds }} {{- end }} - {{- if and .Values.backend.nodeSelector (gt (len .Values.backend.nodeSelector) 0) }} + {{- if and .worker .Values.backend.worker.nodeSelector (gt (len .Values.backend.worker.nodeSelector) 0) }} + nodeSelector: + {{- toYaml .Values.backend.worker.nodeSelector | nindent 8 }} + {{- else if and (not .worker) .Values.backend.nodeSelector (gt (len .Values.backend.nodeSelector) 0) }} nodeSelector: {{- toYaml .Values.backend.nodeSelector | nindent 8 }} {{- end }} - {{- if and .Values.backend.tolerations (gt (len .Values.backend.tolerations) 0) }} + {{- if and .worker .Values.backend.worker.tolerations (gt (len .Values.backend.worker.tolerations) 0) }} + tolerations: + {{- toYaml .Values.backend.worker.tolerations | nindent 8 }} + {{- else if and (not .worker) .Values.backend.tolerations (gt (len .Values.backend.tolerations) 0) }} tolerations: {{- toYaml .Values.backend.tolerations | nindent 8 }} {{- end }} @@ -725,4 +744,7 @@ spec: configMap: name: skills-catalog {{- end }} +{{- end -}} +{{- if .Values.backend.enabled }} +{{ include "commonly.backend.deployment" (merge (dict "worker" false) .) }} {{- end }} diff --git a/k8s/helm/commonly/templates/core/backend-worker-deployment.yaml b/k8s/helm/commonly/templates/core/backend-worker-deployment.yaml new file mode 100644 index 0000000000..93963c33ec --- /dev/null +++ b/k8s/helm/commonly/templates/core/backend-worker-deployment.yaml @@ -0,0 +1,3 @@ +{{- if and .Values.backend.enabled .Values.backend.worker.enabled }} +{{ include "commonly.backend.deployment" (merge (dict "worker" true) .) }} +{{- end }} diff --git a/k8s/helm/commonly/values-dev.yaml b/k8s/helm/commonly/values-dev.yaml index 843ad1f0d6..187e9ddfdb 100644 --- a/k8s/helm/commonly/values-dev.yaml +++ b/k8s/helm/commonly/values-dev.yaml @@ -34,6 +34,16 @@ agentProvisioning: backend: replicaCount: 1 + worker: + enabled: false + # Cron and the Discord gateway must survive spot reclamation. + nodeSelector: + pool: dev + tolerations: + - key: "pool" + operator: "Equal" + value: "dev" + effect: "NoSchedule" image: repository: gcr.io/YOUR_GCP_PROJECT_ID/commonly-backend tag: "20260417214615-r4" diff --git a/k8s/helm/commonly/values.yaml b/k8s/helm/commonly/values.yaml index f1497542c2..0710666d97 100644 --- a/k8s/helm/commonly/values.yaml +++ b/k8s/helm/commonly/values.yaml @@ -88,6 +88,14 @@ agentProvisioning: backend: enabled: true replicaCount: 1 + worker: + # Opt in only when a dedicated worker can replace web's background jobs. + enabled: false + # Enabling the worker first disables web jobs but keeps worker replicas at + # zero. After old web pods terminate, set this false to start the worker. + startPaused: true + nodeSelector: {} + tolerations: [] image: repository: gcr.io/YOUR_GCP_PROJECT_ID/commonly-backend tag: latest diff --git a/scripts/verify-backend-worker-chart.sh b/scripts/verify-backend-worker-chart.sh new file mode 100644 index 0000000000..b603edee27 --- /dev/null +++ b/scripts/verify-backend-worker-chart.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +set -euo pipefail + +chart=k8s/helm/commonly +yq_bin=${YQ_BIN:-yq} +tmp_dir=$(mktemp -d) +trap 'rm -rf "$tmp_dir"' EXIT + +for lane in default dev staging; do + values=(-f "$chart/values.yaml") + if [[ $lane != default ]]; then + values+=(-f "$chart/values-dev.yaml") + fi + if [[ $lane == staging ]]; then + values+=(-f "$chart/values-staging.yaml") + fi + + helm template "commonly-$lane" "$chart" "${values[@]}" > "$tmp_dir/$lane-off.yaml" + helm template "commonly-$lane" "$chart" "${values[@]}" --set backend.worker.enabled=true --set backend.worker.startPaused=false > "$tmp_dir/$lane-on.yaml" + helm template "commonly-$lane" "$chart" "${values[@]}" --set backend.worker.enabled=true > "$tmp_dir/$lane-paused.yaml" + + off="$tmp_dir/$lane-off.yaml" + on="$tmp_dir/$lane-on.yaml" + paused="$tmp_dir/$lane-paused.yaml" + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .metadata.name' "$off")" = "" + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .metadata.name' "$on")" = backend-worker + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.replicas' "$on")" = 1 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.replicas' "$on")" = 1 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.replicas' "$paused")" = 0 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.replicas' "$paused")" = 1 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER") | .value' "$paused")" = 0 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER") | .value' "$paused")" = 1 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.strategy.type' "$on")" = Recreate + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER") | .value' "$off")" = "" + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER") | .value' "$on")" = 0 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER") | .value' "$on")" = 1 + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.selector.matchLabels.app' "$on")" = backend-worker + test "$("$yq_bin" -N 'select(.kind == "Service" and .metadata.name == "backend") | .spec.selector.app' "$on")" = backend + test "$("$yq_bin" -N 'select((.kind == "Service" or .kind == "Ingress") and .metadata.name == "backend-worker") | .kind' "$on")" = "" + + # With the worker enabled, no existing resource changes except the web + # container's RUN_WORKER=0 env. The worker's env otherwise matches web. + "$yq_bin" -N '.' "$off" > "$tmp_dir/$lane-normalized-off.yaml" + "$yq_bin" -N 'select(.kind != "Deployment" or .metadata.name != "backend-worker") | del(.spec.template.spec.containers[].env[] | select(.name == "RUN_WORKER"))' "$on" > "$tmp_dir/$lane-normalized-on.yaml" + diff -u "$tmp_dir/$lane-normalized-off.yaml" "$tmp_dir/$lane-normalized-on.yaml" + "$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.template.spec.containers[].env | map(select(.name != "RUN_WORKER"))' "$on" > "$tmp_dir/$lane-web-env.yaml" + "$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.containers[].env | map(select(.name != "RUN_WORKER"))' "$on" > "$tmp_dir/$lane-worker-env.yaml" + diff -u "$tmp_dir/$lane-web-env.yaml" "$tmp_dir/$lane-worker-env.yaml" + "$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend") | .spec.template.spec.containers[].image' "$on" > "$tmp_dir/$lane-web-image.yaml" + "$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.containers[].image' "$on" > "$tmp_dir/$lane-worker-image.yaml" + diff -u "$tmp_dir/$lane-web-image.yaml" "$tmp_dir/$lane-worker-image.yaml" + + if [[ $lane != default ]]; then + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.nodeSelector.pool' "$on")" = dev + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.tolerations[] | select(.key == "pool" and .value == "dev" and .effect == "NoSchedule") | .key' "$on")" = pool + test "$("$yq_bin" -N 'select(.kind == "Deployment" and .metadata.name == "backend-worker") | .spec.template.spec.containers[].env[] | select(.name == "AGENT_PROVISIONER_K8S") | .value' "$on")" = 1 + fi +done