From 165114471f4495f6ea175f95eba28d117653e4cf Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Fri, 31 Jul 2026 10:34:10 +0100 Subject: [PATCH] fix: Fixes to running on kubernetes --- .gitea/workflows/chart-release.yml | 149 +++++++ claude.md | 75 +++- deploy/chart/vantage/templates/NOTES.txt | 27 ++ deploy/chart/vantage/templates/_helpers.tpl | 73 ++++ .../chart/vantage/templates/migrate-job.yaml | 68 ++++ deploy/chart/vantage/templates/mongo.yaml | 2 + deploy/chart/vantage/templates/redis.yaml | 2 + deploy/chart/vantage/templates/server.yaml | 107 +++-- deploy/chart/vantage/templates/web.yaml | 25 +- deploy/chart/vantage/values.yaml | 49 ++- deploy/docker/.env.example | 1 - deploy/docker/docker-compose.yml | 1 - .../getting-started/self-hosted-install.md | 1 - .../docs/reference/environment-variables.md | 4 +- server/cmd/main.go | 98 ++++- server/internal/api/handlers.go | 3 + server/internal/api/health.go | 61 +++ server/internal/api/workflows.go | 66 ++-- server/internal/auth/session.go | 10 + server/internal/bus/bus.go | 220 +++++++++++ server/internal/bus/leader.go | 100 +++++ server/internal/grpc/server.go | 9 +- server/internal/proxy/session.go | 24 ++ server/internal/services/audit_retention.go | 14 +- server/internal/services/consoleproxy.go | 203 ++++++++-- server/internal/services/dispatch.go | 258 ++++++++++-- server/internal/services/steplogs.go | 373 +++++++++++++----- server/internal/services/stepresults.go | 90 +++-- server/internal/services/workflow_runner.go | 27 +- server/internal/services/workflows.go | 3 + web/app/healthz/route.ts | 15 + 31 files changed, 1880 insertions(+), 278 deletions(-) create mode 100644 .gitea/workflows/chart-release.yml create mode 100644 deploy/chart/vantage/templates/migrate-job.yaml create mode 100644 server/internal/api/health.go create mode 100644 server/internal/bus/bus.go create mode 100644 server/internal/bus/leader.go create mode 100644 web/app/healthz/route.ts diff --git a/.gitea/workflows/chart-release.yml b/.gitea/workflows/chart-release.yml new file mode 100644 index 0000000..fe11a36 --- /dev/null +++ b/.gitea/workflows/chart-release.yml @@ -0,0 +1,149 @@ +name: Chart Release + +on: + # Every push that touches the chart is validated. Publishing is separate and + # deliberate: a chart version is immutable in the registry once pushed, so + # it must come from a tag someone chose, not from whatever landed on main. + # No `paths` filter on push, deliberately. A paths filter applies to tag + # pushes too, so tagging a commit that happened not to touch the chart + # would skip the publish entirely — a release that silently does nothing. + # Validation is seconds of helm rendering; running it on every push to main + # is cheaper than that failure mode. + push: + branches: + - main + tags: + - "chart/v*" + pull_request: + paths: + - "deploy/chart/**" + workflow_dispatch: + +env: + CHART_DIR: deploy/chart/vantage + HELM_VERSION: v3.16.3 + +jobs: + chart: + runs-on: ubuntu-docker + container: alpine:3.21 + steps: + # git for actions/checkout, curl for both the Helm download and the + # registry upload, tar because the Helm tarball is not self-extracting. + - name: Setup + run: apk add --no-cache bash curl git tar nodejs npm + + - name: Install Helm + run: | + set -eu + curl -fsSL "https://get.helm.sh/helm-${HELM_VERSION}-linux-amd64.tar.gz" \ + | tar -xz -C /tmp linux-amd64/helm + mv /tmp/linux-amd64/helm /usr/local/bin/helm + helm version --short + + - name: Checkout + uses: actions/checkout@v4 + + - name: Lint + run: helm lint "$CHART_DIR" + + # Rendering is the real test. `helm lint` accepts a chart whose + # templates fail to execute, and every guard in this chart is a + # template `fail` that only fires during rendering. + - name: Render default values + run: helm template test "$CHART_DIR" > /dev/null + + - name: Render a multi-replica install + run: | + helm template test "$CHART_DIR" \ + --set server.replicaCount=3 \ + --set web.replicaCount=3 > /dev/null + + - name: Render against external Redis and MongoDB + run: | + helm template test "$CHART_DIR" \ + --set redis.enabled=false \ + --set redis.addr=redis.example.com:6379 \ + --set mongo.enabled=false \ + --set server.env.mongoUri=mongodb://mongo.example.com:27017/vantage > /dev/null + + # The guards are load-bearing, so their absence is a regression the + # same way a broken render is. Each of these must fail. + - name: Check the guards still refuse bad values + run: | + set -eu + + refuses() { + desc="$1"; shift + if helm template test "$CHART_DIR" "$@" > /dev/null 2>&1; then + echo "GUARD MISSING: $desc was accepted" + exit 1 + fi + echo "ok: refused $desc" + } + + refuses "mongo disabled with an in-chart URI" \ + --set mongo.enabled=false + refuses "redis disabled with no external address" \ + --set redis.enabled=false + refuses "multiple replicas on a ReadWriteOnce volume" \ + --set server.replicaCount=2 --set server.persistence.enabled=true + + - name: Read the chart version + id: chart + run: | + set -eu + VERSION="$(grep '^version:' "$CHART_DIR/Chart.yaml" | awk '{print $2}')" + echo "version=$VERSION" >> "$GITHUB_OUTPUT" + echo "chart version is $VERSION" + + # Chart.yaml is the source of truth for the version; the tag only + # says "publish this one". A mismatch is a mistake worth stopping + # for — the alternative is stamping the tag over Chart.yaml, which + # leaves the repository disagreeing with what was published. + - name: Check the tag matches Chart.yaml + if: startsWith(github.ref, 'refs/tags/chart/v') + run: | + set -eu + TAG_VERSION="${GITHUB_REF_NAME#chart/v}" + CHART_VERSION="${{ steps.chart.outputs.version }}" + if [ "$TAG_VERSION" != "$CHART_VERSION" ]; then + echo "tag chart/v$TAG_VERSION does not match Chart.yaml version $CHART_VERSION" + echo "bump version: in $CHART_DIR/Chart.yaml, or retag." + exit 1 + fi + + - name: Package + run: | + set -eu + mkdir -p dist + helm package "$CHART_DIR" --destination dist + ls -l dist + + - name: Publish to the Gitea chart registry + if: startsWith(github.ref, 'refs/tags/chart/v') + env: + # github.server_url is this Gitea instance, so the registry + # host needs no variable of its own and cannot drift from it. + REGISTRY: ${{ github.server_url }}/api/packages/${{ github.repository_owner }}/helm/api/charts + REGISTRY_USER: ${{ secrets.REGISTRY_USER }} + REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }} + CHART_VERSION: ${{ steps.chart.outputs.version }} + run: | + set -eu + PKG="dist/vantage-${CHART_VERSION}.tgz" + test -f "$PKG" + + # --fail-with-body so an HTTP error is a failed step with the + # server's explanation, rather than a green run that published + # nothing. A repeated version is rejected by the registry; + # that is the intended behaviour, not something to retry past. + curl --fail-with-body -sS \ + --user "${REGISTRY_USER}:${REGISTRY_PASSWORD}" \ + -X POST \ + --upload-file "$PKG" \ + "$REGISTRY" + + echo "published vantage ${CHART_VERSION}" + echo " helm repo add vantage ${{ github.server_url }}/api/packages/${{ github.repository_owner }}/helm" + echo " helm install vantage vantage/vantage --version ${CHART_VERSION}" diff --git a/claude.md b/claude.md index e52bf13..4a6ab4e 100644 --- a/claude.md +++ b/claude.md @@ -120,7 +120,7 @@ Upload a public key, assign it per server, revoke softly. The agent diffs desire ### Workflows -A library of reusable **steps** (bash or PowerShell scripts with declared inputs, outputs, and secret refs) composed into **workflows** targeting a set of servers. Running one snapshots the resolved steps into a `WorkflowRun`, then dispatches `RunStepCmd` over the agent command stream. Step stdout/stderr streams back as `StepOutputChunk` and is written to a log file on disk; the UI streams it live. Steps support `on_failure: stop|continue|retry`, per-run env passed between steps via `output_env`, and a per-run workspace directory the agent cleans up at the end. +A library of reusable **steps** (bash or PowerShell scripts with declared inputs, outputs, and secret refs) composed into **workflows** targeting a set of servers. Running one snapshots the resolved steps into a `WorkflowRun`, then dispatches `RunStepCmd` over the agent command stream. Step stdout/stderr streams back as `StepOutputChunk` and is written to MongoDB (`workflow_log_lines`, one document per line); the UI streams it live. Steps support `on_failure: stop|continue|retry`, per-run env passed between steps via `output_env`, and a per-run workspace directory the agent cleans up at the end. Default steps are seeded per org at boot (`SeedDefaultSteps`) from `VANTAGE_DEFAULT_STEPS_DIR`, which `server/Dockerfile` bakes to `/opt/default-steps` from the repo's `default_steps/`. Deliberately **not** under `/data` — that is a bind mount, so the library would be editable from the host. Adding a step there means committing a file and rebuilding, which is why `default_steps/` is in the `server` rebuild trigger. **Steps with `source: "default"` are read-only**: `UpdateStep`/`DeleteStep` refuse with `ErrDefaultStep` (409), because seeding rewrites them on every boot, so an edit would silently revert and a delete would come back. `web/` mirrors this — the step modal opens read-only, Delete is hidden, and the designer's per-step script override is `readOnly` for a default library step — but as elsewhere, the API is the boundary and the UI is the courtesy. Seeding writes straight to the collection rather than through `UpdateStep`, so the guard does not lock out the seeder. Logs are swept by retention (`workflow_log_retention_days`; nil = 30 days, 0 = forever). @@ -155,6 +155,56 @@ encrypted, single-use, and consumed when the tunnel opens. None of them reach the agent — the session is negotiated end-to-end between guacd and the target daemon, so the agent relays bytes it cannot read. +### Running more than one server replica + +An agent's `CommandStream` terminates on exactly **one** server process. Every +piece of coordination below exists because of that single fact: with several +replicas, the process asked to do something to an agent is almost never the +process holding that agent's stream. + +`server/internal/bus` is the Redis message bus that closes the gap. It adds no +infrastructure — Redis was already required for sessions — and it is **not +optional on a single-replica deployment**: dispatch takes the bus path always, +so the code running in production is the code running everywhere, rather than a +rare cross-pod branch that only fails under load. + +| Concern | How it crosses replicas | +| ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Which pod owns an agent | `vantage:agent:` holds the owner's node ID with a 30s TTL, renewed every 10s. `Dispatcher.IsConnected` is an `EXISTS` on it | +| Sending a command | published to `vantage:cmd:`; the owner pod acks on `vantage:ack:`. **Request/ack, not a queue** — a command whose owner died must fail loudly (503) rather than queue | +| Step results | the owner pod publishes to `vantage:res:`; the pod driving the run subscribes **before** dispatching, or a fast agent answers into a channel nobody has joined | +| Step output | never crosses. The dispatch envelope carries the secret mask list, so the owner pod masks and writes lines itself — unmasked bytes stay off the bus | +| Console relay | the envelope asks the owner pod to bind the listener, and the ack returns **that pod's** address for guacd. The relay's failure reason comes back on `vantage:proxyend:` | +| Background jobs | `bus.RunAsLeader` — one Redis lock named `housekeeping` | + +**Workflow logs are in MongoDB** (`workflow_log_lines`, one document per line, +with a `workflow_log_seq` counter document per run/server). Two pods write the +same log concurrently — the run's pod emits markers, the agent's pod emits +output — so ordering only means anything if both draw sequence numbers from the +same counter. `StepRun.log_offset` is that sequence number now, not a byte +offset. Writes are batched (128 lines or 250ms) and capped: 8 KB per line, +200k lines per server-run, after which one final `[vantage] log truncated` +marker is written and the rest is dropped. Without that cap a `yes` in a step +is a database incident. **Nothing writes to `/data` any more**, which is why +`server.persistence` now defaults to off and `VANTAGE_WORKFLOW_LOG_DIR` is gone. + +**The leader lock is not an optimisation.** N replicas each running the monitor +scheduler means each check fires N times, each incident notification reaches the +customer N times, and each hourly rollup is written N times; N reapers race to +purge the same Free instance. `monitorsched`, `StartReaper`, `StartLogSweeper`, +`StartAuditSweeper` and the offline sweep therefore all run inside one +`RunAsLeader("housekeeping", …)` — one role, one lock. Each takes a context +cancelled the instant leadership is lost, and must return when it is. + +Redis rather than a Kubernetes `Lease` so Compose takes the identical path: one +implementation to reason about, not two. + +Two deployment requirements come with `replicaCount > 1`: every replica must +share **one** Redis (a per-pod Redis partitions the bus and every agent looks +offline to two thirds of the fleet), and `POD_IP` must be set — the chart does +it from the downward API — because `PROXY_ADVERTISE_HOST` names the Service, and +a Service cannot address the one pod holding a console listener. + ### Inventory and OS updates Agents report CPU/memory/swap/partitions/kernel — metrics every 30s, full static snapshot every 15 min. They also check for pending OS package updates hourly and can apply them on command (`ApplyUpdatesCmd`). @@ -330,6 +380,7 @@ Key-state polling stays on the 30s `SyncKeys` interval. Full message definitions Unauthenticated: ``` +GET /healthz /readyz # liveness / readiness probes GET /install /install.ps1 # dynamic agent install scripts GET /update /update.ps1 GET /auth/bootstrap-status @@ -436,7 +487,7 @@ Paddle is merchant of record; `admin/internal/paddle` is a thin REST client (no ## MongoDB Collections -`servers` · `keys` · `assignments` · `orgs` · `users` · `org_oidc` · `settings` · `secrets` · `workflows` · `workflow_steps` · `workflow_runs` · `monitors` · `incidents` · `monitor_rollups` · `notification_channels` · `console_sessions` · `audit_logs` · `migrations` +`servers` · `keys` · `assignments` · `orgs` · `users` · `org_oidc` · `settings` · `secrets` · `workflows` · `workflow_steps` · `workflow_runs` · `workflow_log_lines` · `workflow_log_seq` · `monitors` · `incidents` · `monitor_rollups` · `notification_channels` · `console_sessions` · `audit_logs` · `migrations` Every document except `migrations` carries `org_id`. Struct definitions are the source of truth — see `server/internal/models/`. @@ -448,6 +499,7 @@ Notes that are not obvious from the structs: - `assignments.revoked_at: null` means active. Revocation is soft, preserving audit history. - `workflow_runs.steps_snapshot` freezes the resolved steps so editing the library never rewrites history. - `console_sessions.token_consumed_at` is set atomically to enforce one-time use. +- `workflow_log_lines` is keyed `(run_id, server_id, seq)` — the index is not an optimisation, every read is a range scan over it. `workflow_log_seq` holds one counter document per `run_id/server_id`, which is what lets two pods interleave into one ordered log. Neither carries `instance_id`: they are reached only through a run, and a run is already scoped. - `users.auth_source` is `local`, `oidc` or `hq`. An `hq` user was projected from a Vantage HQ account and carries `hq_user_id`; HQ owns its role, password and existence. Admin's own database is separate and holds `accounts` · `admin_instances` · `licenses` · `subscriptions` · `plans` · `catalogue` · `entitlements` · `paddle_events` · `staff_users` · `customer_users` · `instance_members` · `admin_audit`. `paddle_events` is the webhook idempotency log, unique on `event_id`: an event is claimed there before processing, and a duplicate of a handled event is a 200 no-op. `instance_members` is unique on `(instance_id, customer_user_id)` — one person holds at most one user in one instance, which makes a grant idempotent-by-refusal rather than silently doubling a projection. It is an _index_ of the control-plane rows, not the authority (see "Grants project, they do not federate"). Admin has no migrations collection; `models.Backfill` runs on every boot and is idempotent by filtering on the absence of what it writes. @@ -537,7 +589,9 @@ Windows: MSI built by CI (WiX), or `installer/setup.ps1` registering the agent a | `PROXY_ADVERTISE_HOST` | no | default `server`; the hostname guacd resolves the control plane by, handed to guacd as the relay's address. Wrong here and every console session fails at connect | | `PROXY_LISTEN_HOST` | no | default `0.0.0.0`; the interface the ephemeral relay listener binds | | `APP_ROOT_LABEL` | no | default `vantage`; wrong value disables the host/session org guard | -| `VANTAGE_WORKFLOW_LOG_DIR` | no | where run logs are written | +| `POD_IP` | no | this pod's own address, set by the Helm chart from the downward API. **Takes precedence over `PROXY_ADVERTISE_HOST`** — a console relay listener belongs to one replica, and a Service address names all of them | +| `VANTAGE_MIGRATE_ONLY` | no | run schema setup (migrations, index builders, default-step seeding) and exit without serving. `GRPC_HOST` is not required in this mode. Set by the Helm chart's pre-upgrade Job | +| `VANTAGE_SKIP_MIGRATIONS` | no | serve without running schema setup, on the assumption a Job already did. Set by the chart's Deployment whenever `server.migrationJob.enabled`. Unset under Compose, where one process still migrates and then serves | | `FREE_INSTANCE_REAP_AFTER` | no | duration past a Free licence's expiry before the instance and all its data are deleted. **Empty disables the reaper, and empty is the default.** Set to `336h` in `docker-compose.site.yml` only — a self-hosted deployment must never reap. Must match admin's value, which only names the date in warning emails | **sitesvc** (`deploy/docker-compose.site.yml` only): @@ -670,10 +724,24 @@ cd /opt/vantage && docker compose -f docker-compose.yml -f docker-compose.site.y The gap this leaves: **changing a repo variable pushes no commit, so nothing rebuilds.** After editing `ADMIN_API_URL`, `HQ_URL` or `ADMIN_ENV`, run the workflow manually — that is what `workflow_dispatch` is there for. Base images also stop being refreshed on a service nobody touches; a periodic manual run covers that. +### `chart-release.yml` — validates on every chart change, publishes on `chart/v*` tags + +Two jobs' worth of work in one, split by trigger. Any push or PR touching `deploy/chart/` lints the chart and renders it four ways: defaults, a multi-replica install, external Redis and MongoDB, and a set of values that **must be refused**. That last one is the point — every safety rail in this chart is a template `fail`, and `helm lint` happily accepts a chart whose templates never execute, so only rendering proves they still fire. + +Publishing runs only on a `chart/v*` tag, to the Gitea Helm registry at `/api/packages//helm/api/charts`. **`Chart.yaml` is the source of truth for the version**; the tag only selects which one to publish, and a tag that disagrees with `Chart.yaml` fails rather than stamping over it — the alternative leaves the repository disagreeing with what shipped. A version already in the registry is rejected by Gitea, which is intended: published chart versions are immutable. + +The registry host comes from `github.server_url`, so it cannot drift from the instance the workflow is running on, and it reuses `REGISTRY_USER` / `REGISTRY_PASSWORD` — the same `write:packages` token the images use. + +```bash +helm repo add vantage https://gitea.hostxtra.co.uk/api/packages/mrhid6/helm +helm install vantage vantage/vantage --version 0.1.0 +``` + ### Tagging ```bash git tag agent/v1.0.0 && git push origin agent/v1.0.0 # agent release +git tag chart/v0.1.0 && git push origin chart/v0.1.0 # helm chart package git push origin main # server + web deploy ``` @@ -717,6 +785,7 @@ git push origin main # server + web deploy - **`org_id` on every document** — isolation enforced at the query layer, not by separate databases. - **root only** — manages `/root/.ssh/authorized_keys`; no per-user key management. - **Windows agents are second-class by design** — register, heartbeat, run steps, report inventory; no `authorized_keys` management. +- **Both `server` and `web` scale horizontally** — see "Running more than one server replica" below. `web` holds nothing; `server` holds per-agent state that is routed between replicas over Redis rather than duplicated. - **Deletion lives in the control plane** — admin sends the warnings because it knows the billing address; the control plane performs the delete because it is the only service that knows which collections carry `instance_id`. Mirroring that list into admin would drift, and a drift there deletes the wrong rows. ## graphify diff --git a/deploy/chart/vantage/templates/NOTES.txt b/deploy/chart/vantage/templates/NOTES.txt index 8948f50..e9206dd 100644 --- a/deploy/chart/vantage/templates/NOTES.txt +++ b/deploy/chart/vantage/templates/NOTES.txt @@ -1,12 +1,39 @@ Vantage has been deployed as release "{{ .Release.Name }}" in namespace "{{ .Release.Namespace }}". Services created: +{{- if .Values.redis.enabled }} - {{ .Release.Name }}-redis (ClusterIP {{ .Values.redis.port }}) +{{- else }} + - Redis: not deployed, using external {{ .Values.redis.addr }} +{{- end }} +{{- if .Values.mongo.enabled }} - {{ .Release.Name }}-mongo (ClusterIP {{ .Values.mongo.port }}) +{{- else }} + - MongoDB: not deployed, using the external server.env.mongoUri +{{- end }} - {{ .Release.Name }}-guacd ({{ .Values.guacd.service.type }} {{ .Values.guacd.service.port }}) - {{ .Release.Name }}-server ({{ .Values.server.service.type }} http:{{ .Values.server.service.httpPort }} grpc:{{ .Values.server.service.grpcPort }}) - {{ .Release.Name }}-web ({{ .Values.web.service.type }} {{ .Values.web.service.port }}) +Scaling (server.replicaCount / web.replicaCount): + - Both scale. Pin the image tags first — replicas on different builds serve + mismatched web asset hashes, and mixed server versions share one bus. + - server replicas route agent commands, step results and console relays to + each other over Redis, so every replica must use the SAME Redis. Workflow + logs are in MongoDB, not on a volume. + - Background work (monitor scheduler, Free reaper, log and audit retention, + the offline sweep) runs on one replica at a time under a Redis leader lock. + - server.persistence must be off to scale past one replica on a ReadWriteOnce + volume. Nothing writes to it any more. +{{- if gt (int .Values.server.replicaCount) 1 }} + - Console relays are reached by pod IP; guacd must be able to dial pod IPs + directly (it can, inside the cluster network). +{{- end }} +{{- if .Values.server.migrationJob.enabled }} + - Migrations run in the {{ .Release.Name }}-migrate Job before each upgrade; + the pods skip them. Its logs are kept: kubectl logs job/{{ .Release.Name }}-migrate +{{- end }} + By default the server/web/guacd services are ClusterIP only (no host port publishing, unlike the original docker-compose file). To expose them externally, set server.service.type / web.service.type / guacd.service.type to NodePort or LoadBalancer, diff --git a/deploy/chart/vantage/templates/_helpers.tpl b/deploy/chart/vantage/templates/_helpers.tpl index f1063cb..677c83a 100644 --- a/deploy/chart/vantage/templates/_helpers.tpl +++ b/deploy/chart/vantage/templates/_helpers.tpl @@ -9,3 +9,76 @@ Common name helpers app.kubernetes.io/instance: {{ .Release.Name }} app.kubernetes.io/managed-by: {{ .Release.Service }} {{- end -}} + +{{/* +vantage.server.env renders the server container's environment. + +It lives here because two workloads need it identically: the Deployment and the +pre-upgrade migration Job. A Job that connected to a different database than the +pods it migrates for would be worse than no Job at all, so there is one copy and +both read it. +*/}} +{{- define "vantage.server.env" -}} +- name: MONGO_URI + {{- $mongoUri := tpl .Values.server.env.mongoUri . }} + {{- if and (not .Values.mongo.enabled) (contains (printf "%s-mongo" .Release.Name) $mongoUri) }} + {{- fail "mongo.enabled is false, so server.env.mongoUri must point at an external MongoDB rather than the in-chart one" }} + {{- end }} + value: {{ $mongoUri | quote }} +- name: REDIS_ADDR + {{- if .Values.redis.enabled }} + value: "{{ .Release.Name }}-redis:{{ .Values.redis.port }}" + {{- else }} + {{- if not .Values.redis.addr }} + {{- fail "redis.enabled is false, so redis.addr must be set to an external Redis host:port" }} + {{- end }} + value: {{ .Values.redis.addr | quote }} + {{- end }} +{{- if .Values.redis.auth.existingSecret }} +- name: REDIS_USERNAME + valueFrom: + secretKeyRef: + name: {{ .Values.redis.auth.existingSecret }} + key: {{ .Values.redis.auth.usernameKey }} + optional: true +- name: REDIS_PASSWORD + valueFrom: + secretKeyRef: + name: {{ .Values.redis.auth.existingSecret }} + key: {{ .Values.redis.auth.passwordKey }} +{{- else }} +{{- if .Values.redis.auth.username }} +- name: REDIS_USERNAME + value: {{ .Values.redis.auth.username | quote }} +{{- end }} +{{- if .Values.redis.auth.password }} +- name: REDIS_PASSWORD + value: {{ .Values.redis.auth.password | quote }} +{{- end }} +{{- end }} +- name: GRPC_HOST + value: {{ .Values.server.env.grpcHost | quote }} +- name: GRPC_PORT + value: {{ .Values.server.service.grpcPort | quote }} +- name: HTTP_PORT + value: {{ .Values.server.service.httpPort | quote }} +- name: KEY_ENCRYPTION_KEY + value: {{ .Values.server.env.keyEncryptionKey | quote }} +- name: GUACD_ADDR + value: "{{ .Release.Name }}-guacd:{{ .Values.guacd.service.port }}" +- name: APP_ROOT_LABEL + value: {{ .Values.server.env.appRootLabel | quote }} +- name: PROXY_ADVERTISE_HOST + value: {{ .Values.server.env.proxyAdvertiseHost | quote }} +- name: PROXY_LISTEN_HOST + value: {{ .Values.server.env.proxyListenHost | quote }} +# The address guacd dials to reach a console relay. It must name one pod, not +# the Service: the relay listener is bound by whichever pod holds that agent's +# command stream, and a Service would send guacd to a different one. POD_IP +# takes precedence over PROXY_ADVERTISE_HOST in the server for exactly this +# reason, so the setting above stays meaningful only outside Kubernetes. +- name: POD_IP + valueFrom: + fieldRef: + fieldPath: status.podIP +{{- end -}} diff --git a/deploy/chart/vantage/templates/migrate-job.yaml b/deploy/chart/vantage/templates/migrate-job.yaml new file mode 100644 index 0000000..8957afc --- /dev/null +++ b/deploy/chart/vantage/templates/migrate-job.yaml @@ -0,0 +1,68 @@ +{{- if .Values.server.migrationJob.enabled }} +{{/* +Schema setup, lifted out of the serving pods. + +Every server process used to run migrations, index builders and default-step +seeding at boot. With one replica that is fine. With two it is not: 0004 renames +the orgs collection to instances, and a sibling reading it mid-rename is a +corruption, not a retry. + +A Helm hook Job runs it once, before any pod of the new version starts. The +Deployment then sets VANTAGE_SKIP_MIGRATIONS, which is what makes the Job's +existence load-bearing rather than decorative — if you disable the Job, the +pods go back to migrating themselves and you must go back to one replica. + +hook-weight orders this after the dependency waits; before-hook-creation deletes +the previous Job so a repeat upgrade is not blocked by an immutable object. The +Job is deliberately NOT deleted on success: its logs are the record of what the +upgrade did to the database. +*/}} +apiVersion: batch/v1 +kind: Job +metadata: + name: {{ .Release.Name }}-migrate + labels: + {{- include "vantage.labels" . | nindent 4 }} + app.kubernetes.io/component: migrate + annotations: + "helm.sh/hook": pre-install,pre-upgrade + "helm.sh/hook-weight": "0" + "helm.sh/hook-delete-policy": before-hook-creation +spec: + backoffLimit: {{ .Values.server.migrationJob.backoffLimit }} + # A migration that has not finished in this long is stuck, and a stuck + # migration should fail the upgrade rather than hold it open forever. + activeDeadlineSeconds: {{ .Values.server.migrationJob.activeDeadlineSeconds }} + template: + metadata: + labels: + app.kubernetes.io/instance: {{ .Release.Name }} + app.kubernetes.io/component: migrate + spec: + restartPolicy: Never + {{- if .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml .Values.imagePullSecrets | nindent 8 }} + {{- end }} + {{- if .Values.mongo.enabled }} + # Only Mongo. The Job never opens Redis, and waiting on a Redis this + # chart may not even deploy would block an upgrade for no reason. + initContainers: + - name: wait-for-mongo + image: busybox:1.36 + command: + - sh + - -c + - | + until nc -z {{ .Release.Name }}-mongo {{ .Values.mongo.port }}; do + echo "waiting for mongo..."; sleep 2; + done + {{- end }} + containers: + - name: migrate + image: "{{ .Values.server.image.repository }}:{{ .Values.server.image.tag }}" + env: + {{- include "vantage.server.env" . | nindent 12 }} + - name: VANTAGE_MIGRATE_ONLY + value: "true" +{{- end }} diff --git a/deploy/chart/vantage/templates/mongo.yaml b/deploy/chart/vantage/templates/mongo.yaml index a27b238..4af7f54 100644 --- a/deploy/chart/vantage/templates/mongo.yaml +++ b/deploy/chart/vantage/templates/mongo.yaml @@ -1,3 +1,4 @@ +{{- if .Values.mongo.enabled }} {{- if .Values.mongo.persistence.enabled }} apiVersion: v1 kind: PersistentVolumeClaim @@ -88,3 +89,4 @@ spec: ports: - port: {{ .Values.mongo.port }} targetPort: {{ .Values.mongo.port }} +{{- end }} diff --git a/deploy/chart/vantage/templates/redis.yaml b/deploy/chart/vantage/templates/redis.yaml index 5c41cee..d6db3b1 100644 --- a/deploy/chart/vantage/templates/redis.yaml +++ b/deploy/chart/vantage/templates/redis.yaml @@ -1,3 +1,4 @@ +{{- if .Values.redis.enabled }} {{- if .Values.redis.persistence.enabled }} apiVersion: v1 kind: PersistentVolumeClaim @@ -88,3 +89,4 @@ spec: ports: - port: {{ .Values.redis.port }} targetPort: {{ .Values.redis.port }} +{{- end }} diff --git a/deploy/chart/vantage/templates/server.yaml b/deploy/chart/vantage/templates/server.yaml index 79dd8a2..74fe4c4 100644 --- a/deploy/chart/vantage/templates/server.yaml +++ b/deploy/chart/vantage/templates/server.yaml @@ -25,7 +25,19 @@ metadata: {{- include "vantage.labels" . | nindent 4 }} app.kubernetes.io/component: server spec: - replicas: 1 + {{- $replicas := int .Values.server.replicaCount }} + replicas: {{ $replicas }} + {{- if and .Values.server.persistence.enabled (eq .Values.server.persistence.accessMode "ReadWriteOnce") }} + # A ReadWriteOnce volume cannot be mounted by a second pod at all, and cannot + # be handed to a new pod while the old one still holds it. Persistence is off + # by default now that nothing writes to it; if it is on, replicas are capped + # at one and updates go through Recreate. + {{- if gt $replicas 1 }} + {{- fail "server.persistence.enabled with a ReadWriteOnce volume cannot be combined with server.replicaCount > 1. Nothing in the server writes to that volume any more (workflow logs live in MongoDB); set server.persistence.enabled=false, or use a ReadWriteMany accessMode if you are keeping it for another reason." }} + {{- end }} + strategy: + type: Recreate + {{- end }} selector: matchLabels: app.kubernetes.io/instance: {{ .Release.Name }} @@ -40,9 +52,13 @@ spec: imagePullSecrets: {{- toYaml .Values.imagePullSecrets | nindent 8 }} {{- end }} - # Wait for redis & mongo to be reachable, approximating compose's - # `depends_on: condition: service_healthy` + {{- if or .Values.redis.enabled .Values.mongo.enabled }} + # Wait for the dependencies this chart deploys to be reachable, + # approximating compose's `depends_on: condition: service_healthy`. An + # external Redis or Mongo is assumed to be up already — waiting on one + # would only turn someone else's outage into a stuck pod. initContainers: + {{- if .Values.redis.enabled }} - name: wait-for-redis image: busybox:1.36 command: @@ -52,6 +68,8 @@ spec: until nc -z {{ .Release.Name }}-redis {{ .Values.redis.port }}; do echo "waiting for redis..."; sleep 2; done + {{- end }} + {{- if .Values.mongo.enabled }} - name: wait-for-mongo image: busybox:1.36 command: @@ -61,6 +79,8 @@ spec: until nc -z {{ .Release.Name }}-mongo {{ .Values.mongo.port }}; do echo "waiting for mongo..."; sleep 2; done + {{- end }} + {{- end }} containers: - name: server image: "{{ .Values.server.image.repository }}:{{ .Values.server.image.tag }}" @@ -68,40 +88,71 @@ spec: - containerPort: {{ .Values.server.service.httpPort }} - containerPort: {{ .Values.server.service.grpcPort }} env: - - name: MONGO_URI - value: {{ tpl .Values.server.env.mongoUri . | quote }} - - name: REDIS_ADDR - value: "{{ .Release.Name }}-redis:{{ .Values.redis.port }}" - - name: GRPC_HOST - value: {{ .Values.server.env.grpcHost | quote }} - - name: GRPC_PORT - value: {{ .Values.server.service.grpcPort | quote }} - - name: HTTP_PORT - value: {{ .Values.server.service.httpPort | quote }} - - name: KEY_ENCRYPTION_KEY - value: {{ .Values.server.env.keyEncryptionKey | quote }} - - name: VANTAGE_WORKFLOW_LOG_DIR - value: {{ .Values.server.env.vantageWorkflowLogDir | quote }} - - name: GUACD_ADDR - value: "{{ .Release.Name }}-guacd:{{ .Values.guacd.service.port }}" - - name: APP_ROOT_LABEL - value: {{ .Values.server.env.appRootLabel | quote }} - - name: PROXY_ADVERTISE_HOST - value: {{ .Values.server.env.proxyAdvertiseHost | quote }} - - name: PROXY_LISTEN_HOST - value: {{ .Values.server.env.proxyListenHost | quote }} + {{- include "vantage.server.env" . | nindent 12 }} + {{- if .Values.server.migrationJob.enabled }} + # Schema setup ran in the pre-upgrade Job. Pods that repeated it + # would race each other, and the rename migration is not a race + # that tolerates a loser. + - name: VANTAGE_SKIP_MIGRATIONS + value: "true" + {{- end }} + # Liveness never touches Mongo or Redis: restarting every pod cannot + # fix a database outage, and each restart drops every agent command + # stream and console session it was carrying. Readiness does check + # both, so a pod that cannot serve leaves the Service and stays up. + startupProbe: + httpGet: + path: /healthz + port: {{ .Values.server.service.httpPort }} + periodSeconds: 5 + # Generous: without the migration Job this pod runs every migration + # before it listens, and the rename has a ten-minute budget. + failureThreshold: 150 + livenessProbe: + httpGet: + path: /healthz + port: {{ .Values.server.service.httpPort }} + periodSeconds: 20 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /readyz + port: {{ .Values.server.service.httpPort }} + periodSeconds: 10 + failureThreshold: 3 + {{- if .Values.server.persistence.enabled }} + # Nothing in the server writes here any more — workflow logs moved to + # MongoDB so that every replica can read and write them. The mount + # remains only so an operator upgrading from a file-log release can + # still reach the old files before turning persistence off. volumeMounts: - name: server-data mountPath: /data volumes: - name: server-data - {{- if .Values.server.persistence.enabled }} persistentVolumeClaim: claimName: {{ .Release.Name }}-server-data - {{- else }} - emptyDir: {} {{- end }} --- +{{- if gt (int .Values.server.replicaCount) 1 }} +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: {{ .Release.Name }}-server + labels: + {{- include "vantage.labels" . | nindent 4 }} + app.kubernetes.io/component: server +spec: + # Agents reconnect on their own, but a drain that took every replica at once + # would disconnect every agent in the fleet simultaneously and stall every + # workflow run in flight. + minAvailable: 1 + selector: + matchLabels: + app.kubernetes.io/instance: {{ .Release.Name }} + app.kubernetes.io/component: server +--- +{{- end }} apiVersion: v1 kind: Service metadata: diff --git a/deploy/chart/vantage/templates/web.yaml b/deploy/chart/vantage/templates/web.yaml index cebf903..6c8ed89 100644 --- a/deploy/chart/vantage/templates/web.yaml +++ b/deploy/chart/vantage/templates/web.yaml @@ -6,7 +6,9 @@ metadata: {{- include "vantage.labels" . | nindent 4 }} app.kubernetes.io/component: web spec: - replicas: 1 + # web holds no per-process state: sessions live in Redis and every request is + # proxied to the server. It is the one component here that scales freely. + replicas: {{ .Values.web.replicaCount }} selector: matchLabels: app.kubernetes.io/instance: {{ .Release.Name }} @@ -39,6 +41,27 @@ spec: env: - name: API_URL value: {{ tpl .Values.web.env.apiUrl . | quote }} + # /healthz is served by this Next process; /api is rewritten to the + # server, so a probe there would report the backend's health and keep + # passing while this pod was wedged. + startupProbe: + httpGet: + path: /healthz + port: {{ .Values.web.service.port }} + periodSeconds: 3 + failureThreshold: 20 + livenessProbe: + httpGet: + path: /healthz + port: {{ .Values.web.service.port }} + periodSeconds: 20 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /healthz + port: {{ .Values.web.service.port }} + periodSeconds: 10 + failureThreshold: 3 --- apiVersion: v1 kind: Service diff --git a/deploy/chart/vantage/values.yaml b/deploy/chart/vantage/values.yaml index 9d0bee5..e665771 100644 --- a/deploy/chart/vantage/values.yaml +++ b/deploy/chart/vantage/values.yaml @@ -1,6 +1,10 @@ # Default values for the vantage chart. redis: + # false deploys no Redis and points the server at `redis.addr` instead. + enabled: true + # Only read when enabled is false. host:port of an external Redis. + addr: "" image: repository: redis tag: "8" @@ -10,8 +14,22 @@ redis: storageClass: "" accessMode: ReadWriteOnce port: 6379 + # Both empty for an unauthenticated Redis. Redis 6+ ACL auth takes both; a + # legacy `requirepass` instance takes the password alone and must leave the + # username empty. Set existingSecret to keep the password out of values. + auth: + username: "" + password: "" + # Secret holding the credentials. When set, username/password above are + # ignored and these keys are read from the secret instead. + existingSecret: "" + usernameKey: username + passwordKey: password mongo: + # false deploys no MongoDB. server.env.mongoUri must then point at an + # external one — the chart cannot guess it, and refuses to render without it. + enabled: true image: repository: mongo tag: "7" @@ -31,6 +49,23 @@ guacd: port: 4822 server: + # Safe to raise. Agent commands, step results and console relays are routed + # between replicas over Redis, workflow logs live in MongoDB, and the + # background jobs (monitor scheduler, reaper, retention sweeps) run under a + # Redis leader lock so exactly one replica performs them. + # + # Two requirements come with raising it: server.persistence.enabled must be + # false (or the volume ReadWriteMany), and Redis must be shared by every + # replica — the bus is not optional and a per-pod Redis would partition it. + replicaCount: 1 + # Runs migrations, index builders and default-step seeding once, as a Helm + # pre-install/pre-upgrade hook, instead of in every starting pod. Leave it + # on for Kubernetes. Turning it off puts schema setup back in the pods. + migrationJob: + enabled: true + backoffLimit: 0 + # 15 minutes: the instance rename alone carries a 10-minute budget. + activeDeadlineSeconds: 900 image: repository: gitea.hostxtra.co.uk/mrhid6/vantage/server tag: latest @@ -42,18 +77,28 @@ server: mongoUri: "mongodb://{{ .Release.Name }}-mongo:27017/vantage" grpcHost: "{{ .Release.Name }}-server:9090" keyEncryptionKey: "" - vantageWorkflowLogDir: "" appRootLabel: vantage + # Ignored under Kubernetes: the chart sets POD_IP from the downward API + # and the server prefers it, because a console relay listener belongs to + # one pod and a Service address cannot name one. proxyAdvertiseHost: "{{ .Release.Name }}-server" proxyListenHost: "0.0.0.0" + # Off by default: nothing in the server writes to disk any more. Workflow + # logs, the only thing that ever did, are in MongoDB so that every replica + # can read and write them. Turn this on only to reach files left behind by + # a release that predates that move — and note a ReadWriteOnce volume caps + # replicaCount at 1 while it is on. persistence: - enabled: true + enabled: false size: 1Gi storageClass: "" accessMode: ReadWriteOnce hostPath: /data web: + # Stateless — safe to raise. Pin web.image.tag when you do: replicas on + # different builds serve mismatched chunk hashes and the UI 404s mid-session. + replicaCount: 1 image: repository: gitea.hostxtra.co.uk/mrhid6/vantage/web tag: latest diff --git a/deploy/docker/.env.example b/deploy/docker/.env.example index 95a7500..afae5cc 100644 --- a/deploy/docker/.env.example +++ b/deploy/docker/.env.example @@ -18,4 +18,3 @@ KEY_ENCRYPTION_KEY= MONGO_URI=mongodb://mongo:27017/vantage # Where workflow run logs are written inside the server container. -# VANTAGE_WORKFLOW_LOG_DIR=/data/workflow-logs diff --git a/deploy/docker/docker-compose.yml b/deploy/docker/docker-compose.yml index 1783d48..0845f41 100644 --- a/deploy/docker/docker-compose.yml +++ b/deploy/docker/docker-compose.yml @@ -45,7 +45,6 @@ services: GRPC_PORT: "9090" HTTP_PORT: "8080" KEY_ENCRYPTION_KEY: ${KEY_ENCRYPTION_KEY:-} - VANTAGE_WORKFLOW_LOG_DIR: ${VANTAGE_WORKFLOW_LOG_DIR:-} GUACD_ADDR: guacd:4822 PROXY_ADVERTISE_HOST: server depends_on: diff --git a/docsite/docs/getting-started/self-hosted-install.md b/docsite/docs/getting-started/self-hosted-install.md index 75179cf..625da0f 100644 --- a/docsite/docs/getting-started/self-hosted-install.md +++ b/docsite/docs/getting-started/self-hosted-install.md @@ -50,7 +50,6 @@ GRPC_HOST=vantage.example.com:9090 KEY_ENCRYPTION_KEY= # Optional: where workflow run logs are written inside the container. -VANTAGE_WORKFLOW_LOG_DIR=/data/workflow-logs ``` Generate the encryption key: diff --git a/docsite/docs/reference/environment-variables.md b/docsite/docs/reference/environment-variables.md index d9e9e60..be0e557 100644 --- a/docsite/docs/reference/environment-variables.md +++ b/docsite/docs/reference/environment-variables.md @@ -13,7 +13,7 @@ it is absent. | -------------------------- | --------------- | --------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `GRPC_HOST` | **yes** | | The `host:port` agents dial. Boot fails without it. There is deliberately no fallback to the web host: that would hand every agent a port that does not speak gRPC | | `MONGO_URI` | no | `mongodb://localhost:27017` | The database name is taken from the URI path, falling back to `vantage`. There is no separate `MONGO_DB` | -| `REDIS_ADDR` | no | `localhost:6379` | Sessions only | +| `REDIS_ADDR` | no | `localhost:6379` | Sessions, and the bus that routes agent commands between server replicas. Every replica must point at the **same** Redis | | `REDIS_USERNAME` | no | | Redis 6+ ACL user. Leave empty against a legacy `requirepass` instance, which authenticates with the password alone | | `REDIS_PASSWORD` | no | | Leave empty for an unauthenticated Redis. Both of these exist so an install can use a managed Redis rather than the bundled one | | `KEY_ENCRYPTION_KEY` | yes in practice | | 64 hex characters (32 bytes) for AES-256-GCM. Required for private keys, vault secrets, OIDC client secrets and console credentials | @@ -21,8 +21,8 @@ it is absent. | `GUACD_ADDR` | no | `guacd:4822` | The [browser console](../vantage/browser-console.md) daemon | | `PROXY_ADVERTISE_HOST` | no | `server` | The hostname **guacd** uses to reach the control plane's console relay. Wrong here and every console session fails at connect with guacd unable to resolve the relay | | `PROXY_LISTEN_HOST` | no | `0.0.0.0` | Interface the ephemeral relay listeners bind. Narrow it only if guacd shares a known interface | +| `POD_IP` | no | | Kubernetes only, set by the Helm chart from the downward API. Overrides `PROXY_ADVERTISE_HOST`, because a console relay belongs to one replica and a Service address names all of them | | `APP_ROOT_LABEL` | no | `vantage` | The app root label for the host and session organisation guard | -| `VANTAGE_WORKFLOW_LOG_DIR` | no | | Where workflow run logs are written | :::danger `KEY_ENCRYPTION_KEY` has no recovery path It encrypts SSH private keys, vault secrets, OIDC client secrets and console diff --git a/server/cmd/main.go b/server/cmd/main.go index 7426b83..23da9e1 100644 --- a/server/cmd/main.go +++ b/server/cmd/main.go @@ -4,10 +4,12 @@ import ( "context" "log" "os" + "strings" "time" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/api" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" grpcserver "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/monitorsched" @@ -18,7 +20,21 @@ import ( func main() { mongoURI := getEnv("MONGO_URI", "mongodb://localhost:27017") - if os.Getenv("GRPC_HOST") == "" { + // Two flags, so the schema work can be lifted out of the serving pods. + // + // Under Docker Compose neither is set and nothing changes: one process + // migrates and then serves. Under Kubernetes with more than one replica + // that is unsafe — every pod would run MigrateOrgToInstance at once, and + // renaming collections while a sibling reads them is not a race anyone + // wins. The chart therefore runs a pre-upgrade Job with MIGRATE_ONLY and + // starts the Deployment with SKIP_MIGRATIONS. + migrateOnly := boolEnv("VANTAGE_MIGRATE_ONLY") + skipMigrations := boolEnv("VANTAGE_SKIP_MIGRATIONS") + + // Not required in migrate-only mode: that process never serves gRPC, and + // demanding it would put an agent-facing address in a Job that has no + // business knowing one. + if !migrateOnly && os.Getenv("GRPC_HOST") == "" { log.Fatal("GRPC_HOST is required (host:port agents dial for gRPC)") } @@ -28,6 +44,24 @@ func main() { } log.Println("connected to MongoDB") + if migrateOnly || !skipMigrations { + runSchemaSetup() + } else { + log.Println("VANTAGE_SKIP_MIGRATIONS set: assuming migrations ran elsewhere") + } + + if migrateOnly { + log.Println("VANTAGE_MIGRATE_ONLY set: schema setup complete, exiting") + return + } + + serve() +} + +// runSchemaSetup performs every write that must happen exactly once before the +// application serves: migrations, index builders and default step seeding. It +// is fatal on anything that would leave the schema half-moved. +func runSchemaSetup() { // Migrations 0001 to 0003 still speak the pre-rename shape (orgs, org_id), // so they must run before 0004 renames everything underneath them. if err := services.RunMigrations(); err != nil { @@ -87,9 +121,9 @@ func main() { } } - services.StartLogSweeper() - services.StartAuditSweeper() +} +func serve() { redisAddr := getEnv("REDIS_ADDR", "localhost:6379") redisUser := os.Getenv("REDIS_USERNAME") redisPass := os.Getenv("REDIS_PASSWORD") @@ -98,15 +132,16 @@ func main() { } log.Println("connected to Redis") - go func() { - ticker := time.NewTicker(2 * time.Minute) - defer ticker.Stop() - for range ticker.C { - if err := services.MarkOfflineServers(); err != nil { - log.Printf("mark offline error: %v", err) - } - } - }() + // The bus carries agent commands and step results between replicas. It is + // not optional even on a single-replica deployment: dispatch takes the same + // path either way, so the code exercised in production is the code + // exercised everywhere. + if err := bus.Init(redisAddr, redisUser, redisPass); err != nil { + log.Fatalf("failed to connect the message bus: %v", err) + } + log.Printf("message bus ready as node %s", bus.NodeID()) + + ctx := context.Background() go func() { if err := grpcserver.StartGRPC(9090); err != nil { @@ -114,9 +149,33 @@ func main() { } }() - monitorsched.Start(context.Background()) + // Everything below runs on exactly one replica at a time. + // + // These are cluster-singleton jobs, not per-pod work: N replicas would mean + // every monitor check firing N times, every incident notification delivered + // to the customer N times, every retention sweep deleting concurrently, and + // N reapers racing to purge the same instance. They share one lock rather + // than holding four, because they are one role — housekeeping — and + // splitting them would only spread that role across pods for no benefit. + bus.RunAsLeader(ctx, "housekeeping", func(jobCtx context.Context) { + services.StartLogSweeper(jobCtx) + services.StartAuditSweeper(jobCtx) + services.StartReaper(jobCtx) + monitorsched.Start(jobCtx) - services.StartReaper(context.Background()) + ticker := time.NewTicker(2 * time.Minute) + defer ticker.Stop() + for { + select { + case <-jobCtx.Done(): + return + case <-ticker.C: + if err := services.MarkOfflineServers(); err != nil { + log.Printf("mark offline error: %v", err) + } + } + } + }) r := gin.New() r.Use(gin.Recovery()) @@ -149,3 +208,14 @@ func getEnv(key, fallback string) string { } return fallback } + +// boolEnv reads a flag env var. Anything other than a recognised truthy value +// is false, so a typo leaves the safe default (migrate here, serve here) rather +// than silently skipping schema setup. +func boolEnv(key string) bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) { + case "1", "true", "yes", "on": + return true + } + return false +} diff --git a/server/internal/api/handlers.go b/server/internal/api/handlers.go index c4300f9..c9045de 100644 --- a/server/internal/api/handlers.go +++ b/server/internal/api/handlers.go @@ -20,6 +20,9 @@ func actorFromCtx(c *gin.Context) string { } func RegisterRoutes(r *gin.Engine) { + r.GET("/healthz", handleHealthz) + r.GET("/readyz", handleReadyz) + r.GET("/install", handleInstallScript) r.GET("/install.ps1", handleInstallScriptWindows) r.GET("/update", handleUpdateScript) diff --git a/server/internal/api/health.go b/server/internal/api/health.go new file mode 100644 index 0000000..665e5d3 --- /dev/null +++ b/server/internal/api/health.go @@ -0,0 +1,61 @@ +package api + +import ( + "context" + "net/http" + "time" + + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/auth" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" + "github.com/gin-gonic/gin" +) + +// The two probes answer different questions on purpose. +// +// /healthz is liveness: the process is up and serving. It touches nothing +// external, because a Mongo outage must not make Kubernetes restart every +// server pod — a restart loop cannot fix someone else's database, and it +// destroys every open command stream and console session on the way. +// +// /readyz is readiness: this pod can serve a request end to end, which needs +// both Mongo and Redis. A failing readiness probe pulls the pod out of the +// Service and leaves it running, which is the behaviour that matters during a +// dependency blip. +// +// Both sit outside /api, so neither the session middleware nor the licence +// gate applies. Neither reveals anything beyond up or down. + +const probeTimeout = 2 * time.Second + +func handleHealthz(c *gin.Context) { + c.JSON(http.StatusOK, gin.H{"status": "ok"}) +} + +func handleReadyz(c *gin.Context) { + ctx, cancel := context.WithTimeout(c.Request.Context(), probeTimeout) + defer cancel() + + checks := gin.H{"mongo": "ok", "redis": "ok"} + ready := true + + if db.Client == nil { + checks["mongo"] = "not initialised" + ready = false + } else if err := db.Client.Ping(ctx, nil); err != nil { + checks["mongo"] = "unreachable" + ready = false + } + + if err := auth.PingRedis(ctx); err != nil { + checks["redis"] = "unreachable" + ready = false + } + + status := http.StatusOK + state := "ok" + if !ready { + status = http.StatusServiceUnavailable + state = "unready" + } + c.JSON(status, gin.H{"status": state, "checks": checks}) +} diff --git a/server/internal/api/workflows.go b/server/internal/api/workflows.go index 27a667b..4ecef14 100644 --- a/server/internal/api/workflows.go +++ b/server/internal/api/workflows.go @@ -5,7 +5,6 @@ import ( "fmt" "io" "net/http" - "os" "regexp" "strconv" "strings" @@ -50,23 +49,42 @@ func getServerRunLog(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"}) return } - path := services.ServerRunLogPath(runID, serverID) - b, err := os.ReadFile(path) - if err != nil { + if !services.HasServerRunLog(runID, serverID) { c.JSON(http.StatusNotFound, gin.H{"error": "no logs"}) return } - c.Data(http.StatusOK, "text/plain; charset=utf-8", b) + + c.Header("Content-Type", "text/plain; charset=utf-8") + c.Status(http.StatusOK) + + // Streamed in pages rather than read whole. A log capped at 200k lines is + // tens of megabytes, and holding that in memory per concurrent download is + // how one curious user takes a pod down. + var after int64 + for { + lines, last, err := services.ReadServerRunLog(runID, serverID, after, logPageSize) + if err != nil || len(lines) == 0 { + return + } + for _, l := range lines { + _, _ = c.Writer.WriteString(l) + _, _ = c.Writer.WriteString("\n") + } + c.Writer.Flush() + after = last + } } +// logPageSize bounds one read of the log store. Large enough that a normal log +// is one or two queries, small enough that no single response buffers much. +const logPageSize = 2000 + func streamServerRunLog(c *gin.Context) { runID, serverID := c.Param("runId"), c.Param("serverId") if !uuidLike.MatchString(runID) || !uuidLike.MatchString(serverID) { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid id"}) return } - path := services.ServerRunLogPath(runID, serverID) - c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") @@ -78,29 +96,27 @@ func streamServerRunLog(c *gin.Context) { return } - var offset int64 + // The cursor is a sequence number now, not a byte offset: lines come from + // the log store, which any pod can read, rather than from a file only this + // one has. + var after int64 sendNew := func() { - f, err := os.Open(path) - if err != nil { - return - } - defer f.Close() - if _, err := f.Seek(offset, 0); err != nil { - return - } - buf := make([]byte, 32*1024) for { - n, _ := f.Read(buf) - if n <= 0 { - break + lines, last, err := services.ReadServerRunLog(runID, serverID, after, logPageSize) + if err != nil || len(lines) == 0 { + return } - offset += int64(n) - - for _, line := range splitSSE(buf[:n]) { - _, _ = c.Writer.WriteString("data: " + line + "\n") + after = last + for _, line := range lines { + for _, part := range splitSSE([]byte(line)) { + _, _ = c.Writer.WriteString("data: " + part + "\n") + } + _, _ = c.Writer.WriteString("\n") } - _, _ = c.Writer.WriteString("\n") flusher.Flush() + if len(lines) < logPageSize { + return + } } } diff --git a/server/internal/auth/session.go b/server/internal/auth/session.go index 7d42924..c7146ec 100644 --- a/server/internal/auth/session.go +++ b/server/internal/auth/session.go @@ -5,6 +5,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "errors" "time" "github.com/redis/go-redis/v9" @@ -41,6 +42,15 @@ func InitRedis(addr, username, password string) error { return rdb.Ping(ctx).Err() } +// PingRedis reports whether the session store is reachable. Readiness depends +// on it: a pod that cannot reach Redis can authenticate nobody. +func PingRedis(ctx context.Context) error { + if rdb == nil { + return errors.New("redis not initialised") + } + return rdb.Ping(ctx).Err() +} + func randomHex(n int) (string, error) { b := make([]byte, n) if _, err := rand.Read(b); err != nil { diff --git a/server/internal/bus/bus.go b/server/internal/bus/bus.go new file mode 100644 index 0000000..8662dbe --- /dev/null +++ b/server/internal/bus/bus.go @@ -0,0 +1,220 @@ +// Package bus is the control plane's inter-process message bus. +// +// It exists because an agent's CommandStream lands on exactly one server +// process. With a single replica that process is also the one handling every +// REST request, so an in-memory map was enough. With several replicas it is +// not: the pod asked to run a workflow step is almost never the pod holding +// that agent's stream, and a map cannot reach across the gap. +// +// Redis is already a hard dependency (sessions), so the bus adds no new +// infrastructure. It carries three things: +// +// presence which pod, if any, currently holds an agent's stream +// commands a request/ack exchange delivering a ServerCommand to that pod +// results step results travelling back to the pod driving the run +// +// Everything here is deliberately best-effort delivery with an explicit ack +// rather than a queue. A command whose owner pod died between the presence +// check and the publish must fail loudly and immediately — the caller answers +// 503 and the operator retries — not sit in a queue waiting for a stream that +// no longer exists. +package bus + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "time" + + "github.com/redis/go-redis/v9" +) + +// ErrNoResponder means nothing was subscribed to the channel, or the subscriber +// did not ack within the timeout. Either way the command did not reach an agent. +var ErrNoResponder = errors.New("no responder on channel") + +var ( + rdb *redis.Client + nodeID string +) + +// Init connects the bus. It takes its own client rather than sharing the +// session store's: a subscription occupies its connection for as long as it +// lives, and every agent stream on this pod holds one, so they must not come +// out of the pool that ordinary session reads depend on. +func Init(addr, username, password string) error { + rdb = redis.NewClient(&redis.Options{ + Addr: addr, + Username: username, + Password: password, + }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + return err + } + + // Hostname is the pod name under Kubernetes, which makes a log line naming + // a node directly actionable. The random suffix keeps two processes on one + // host (or a reused pod name) from claiming each other's locks. + host, _ := os.Hostname() + if host == "" { + host = "server" + } + b := make([]byte, 4) + _, _ = rand.Read(b) + nodeID = host + "-" + hex.EncodeToString(b) + return nil +} + +// NodeID identifies this process on the bus. Stable for the process lifetime. +func NodeID() string { return nodeID } + +// Client exposes the underlying Redis client for the leader election and +// presence helpers in this package. It is nil before Init. +func Client() *redis.Client { return rdb } + +// Channel names. Every key and channel is prefixed so a Redis shared with the +// session store (which uses km:) stays legible. +const ( + prefix = "vantage:" + + // CommandChannel carries envelopes to whichever pod holds serverID's stream. + CommandChannel = prefix + "cmd:" + // ackChannel carries the owner pod's answer back to the requesting pod. + ackChannel = prefix + "ack:" + // ResultChannel carries a StepResult back to the pod driving the run. + ResultChannel = prefix + "res:" + // ProxyEndChannel carries a console relay's terminal reason back to the pod + // serving the WebSocket, which is the pod that has to write the audit event. + ProxyEndChannel = prefix + "proxyend:" + + // PresenceKey records which node holds an agent's command stream. + PresenceKey = prefix + "agent:" + // leaderKey records the holder of a named singleton job. + leaderKey = prefix + "leader:" +) + +// Publish sends v, JSON-encoded, to channel. It reports how many subscribers +// received it, which is the only signal Redis pub/sub gives that anyone was +// listening. +func Publish(ctx context.Context, channel string, v any) (int64, error) { + b, err := json.Marshal(v) + if err != nil { + return 0, err + } + return rdb.Publish(ctx, channel, b).Result() +} + +// Subscribe returns a channel of raw payloads and a function that unsubscribes. +// It blocks until Redis confirms the subscription, so a caller may publish +// immediately afterwards without racing its own subscriber. +func Subscribe(ctx context.Context, channel string) (<-chan []byte, func(), error) { + ps := rdb.Subscribe(ctx, channel) + if _, err := ps.Receive(ctx); err != nil { + _ = ps.Close() + return nil, nil, err + } + + out := make(chan []byte, 64) + go func() { + defer close(out) + for msg := range ps.Channel() { + select { + case out <- []byte(msg.Payload): + default: + // A subscriber too slow to keep up would otherwise stall every + // other subscriber sharing this connection. Dropping is correct + // here: commands are acked, and a dropped ack fails the caller + // loudly rather than hanging it. + } + } + }() + + return out, func() { _ = ps.Close() }, nil +} + +// Request publishes req to channel and waits for a single reply on replyTo. +// +// The reply subscription is established before the publish, so a responder that +// answers instantly cannot beat the subscriber into place. A publish that +// reaches no subscriber fails immediately with ErrNoResponder rather than +// burning the whole timeout: nobody is going to answer. +func Request(ctx context.Context, channel, replyTo string, req any, timeout time.Duration) ([]byte, error) { + waitCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + replies, unsub, err := Subscribe(waitCtx, replyTo) + if err != nil { + return nil, fmt.Errorf("subscribe reply channel: %w", err) + } + defer unsub() + + n, err := Publish(waitCtx, channel, req) + if err != nil { + return nil, err + } + if n == 0 { + return nil, ErrNoResponder + } + + select { + case b, ok := <-replies: + if !ok { + return nil, ErrNoResponder + } + return b, nil + case <-waitCtx.Done(): + return nil, ErrNoResponder + } +} + +// Reply answers a Request on its reply channel. +func Reply(ctx context.Context, replyTo string, v any) error { + _, err := Publish(ctx, replyTo, v) + return err +} + +// AckChannelFor names the reply channel for a command. The command ID is +// already unique per dispatch, so it needs no further qualification. +func AckChannelFor(commandID string) string { return ackChannel + commandID } + +// SetPresence claims serverID for this node for ttl. Called repeatedly by the +// pod holding the stream; the TTL is what bounds how long a crashed pod keeps +// claiming an agent it can no longer reach. +func SetPresence(ctx context.Context, serverID string, ttl time.Duration) error { + return rdb.Set(ctx, PresenceKey+serverID, nodeID, ttl).Err() +} + +// ClearPresence releases serverID, but only if this node still holds it. A +// blind DEL would let a pod whose stream had already been re-established +// elsewhere delete the new owner's claim on its way out. +func ClearPresence(ctx context.Context, serverID string) error { + return releaseIfOwner.Run(ctx, rdb, []string{PresenceKey + serverID}, nodeID).Err() +} + +// PresenceHolder returns the node holding serverID's stream, or "" if none does. +func PresenceHolder(ctx context.Context, serverID string) string { + v, err := rdb.Get(ctx, PresenceKey+serverID).Result() + if err != nil { + return "" + } + return v +} + +// IsConnected reports whether any pod currently holds serverID's stream. +func IsConnected(ctx context.Context, serverID string) bool { + n, err := rdb.Exists(ctx, PresenceKey+serverID).Result() + return err == nil && n > 0 +} + +var releaseIfOwner = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) +end +return 0 +`) diff --git a/server/internal/bus/leader.go b/server/internal/bus/leader.go new file mode 100644 index 0000000..6e82524 --- /dev/null +++ b/server/internal/bus/leader.go @@ -0,0 +1,100 @@ +package bus + +import ( + "context" + "log" + "time" + + "github.com/redis/go-redis/v9" +) + +const ( + leaderTTL = 30 * time.Second + leaderRenew = 10 * time.Second +) + +// RunAsLeader runs job on exactly one process at a time, cluster-wide. +// +// The background work this guards is not merely wasteful when duplicated. Every +// replica running the monitor scheduler means every check fires N times, every +// incident notification is delivered N times to the customer, and every hourly +// rollup is written N times. The Free-instance reaper is worse: it deletes +// whole instances, and two processes deleting the same one concurrently is not +// a race anyone wins. +// +// Redis rather than a Kubernetes Lease so that Docker Compose, which has no +// API server, takes the identical code path — one implementation to reason +// about, not two. +// +// job is given a context cancelled the moment leadership is lost, and must +// return when it is cancelled. Losing the lock (a paused process, a Redis +// blip) is treated as fatal to that run of the job: the successor may already +// have started, and two schedulers overlapping is the exact thing being +// prevented. +func RunAsLeader(ctx context.Context, name string, job func(context.Context)) { + go func() { + key := leaderKey + name + for { + if ctx.Err() != nil { + return + } + ok, err := rdb.SetNX(ctx, key, nodeID, leaderTTL).Result() + if err != nil { + log.Printf("leader %s: acquire failed: %v", name, err) + } + if !ok { + select { + case <-ctx.Done(): + return + case <-time.After(leaderRenew): + } + continue + } + + log.Printf("leader %s: acquired by %s", name, nodeID) + jobCtx, cancel := context.WithCancel(ctx) + go job(jobCtx) + holdLeadership(ctx, key, name) + cancel() + log.Printf("leader %s: released by %s", name, nodeID) + + // Give up the key on a clean shutdown so a successor takes over in + // milliseconds rather than waiting out the TTL. + if ctx.Err() != nil { + relCtx, relCancel := context.WithTimeout(context.Background(), 2*time.Second) + _ = releaseIfOwner.Run(relCtx, rdb, []string{key}, nodeID).Err() + relCancel() + return + } + } + }() +} + +// holdLeadership renews the lock until it is lost or ctx ends. +func holdLeadership(ctx context.Context, key, name string) { + t := time.NewTicker(leaderRenew) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + n, err := renewIfOwner.Run(ctx, rdb, []string{key}, nodeID, int(leaderTTL/time.Millisecond)).Int() + if err != nil && err != redis.Nil { + log.Printf("leader %s: renew failed, standing down: %v", name, err) + return + } + if n == 0 { + log.Printf("leader %s: lock lost, standing down", name) + return + } + } + } +} + +var renewIfOwner = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("PEXPIRE", KEYS[1], ARGV[2]) +end +return 0 +`) diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index fd11bfa..8974f33 100644 --- a/server/internal/grpc/server.go +++ b/server/internal/grpc/server.go @@ -178,8 +178,13 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err log.Printf("update last seen %s: %v", srv.ServerID, err) } - ch := services.Dispatcher.Connect(srv.ServerID) - defer services.Dispatcher.Disconnect(srv.ServerID) + // Serve claims this agent's presence on the bus and subscribes this pod to + // its command channel, so a dispatch issued by any other replica arrives + // here. The teardown releases both — an agent that reconnects to a + // different pod must not leave this one advertising a stream it no longer + // has. + ch, release := services.Dispatcher.Serve(stream.Context(), srv.ServerID) + defer release() log.Printf("agent %s connected command stream", srv.ServerID) defer log.Printf("agent %s disconnected command stream", srv.ServerID) diff --git a/server/internal/proxy/session.go b/server/internal/proxy/session.go index cdd07a1..847be17 100644 --- a/server/internal/proxy/session.go +++ b/server/internal/proxy/session.go @@ -37,10 +37,24 @@ type Session struct { reason string conn net.Conn closing bool + onEnd func(string) rendezvous *time.Timer } +// OnEnd registers a callback invoked exactly once, when the session finishes +// tearing down, with the recorded reason (empty if it ended cleanly). +// +// It exists because the process that observes a relay failing is not +// necessarily the process that has to record it: with several replicas the +// listener lives on the pod holding the agent's stream, while the audit event +// belongs to the pod serving the browser's WebSocket. +func (s *Session) OnEnd(fn func(string)) { + s.mu.Lock() + s.onEnd = fn + s.mu.Unlock() +} + // NewSession binds an ephemeral port on listenHost. allowed is the set of IPs // permitted to connect; an empty set allows any, which is the degraded case // when guacd's host could not be resolved. @@ -112,6 +126,16 @@ func (s *Session) Close(reason string) { if conn != nil { _ = conn.Close() } + + // Read after teardown, not before: relay()'s goroutines set the reason + // as they unwind, and reporting one recorded earlier than that would + // name the symptom rather than the cause. + s.mu.Lock() + fn, final := s.onEnd, s.reason + s.mu.Unlock() + if fn != nil { + fn(final) + } }) } diff --git a/server/internal/services/audit_retention.go b/server/internal/services/audit_retention.go index 120545d..a57ed6e 100644 --- a/server/internal/services/audit_retention.go +++ b/server/internal/services/audit_retention.go @@ -15,13 +15,21 @@ import ( // An immediate pass then daily, following StartLogSweeper's shape. Daily rather // than hourly because the unit of retention is a day: sweeping twenty-four times // to delete the same nothing is load without a purpose. -func StartAuditSweeper() { +// It takes a context because it runs under leader election: several replicas +// all trimming the same audit logs is duplicated deletion of customer data, and +// the loop must stop the moment this process stops being the leader. +func StartAuditSweeper(ctx context.Context) { go func() { sweepAuditLogs() t := time.NewTicker(24 * time.Hour) defer t.Stop() - for range t.C { - sweepAuditLogs() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + sweepAuditLogs() + } } }() } diff --git a/server/internal/services/consoleproxy.go b/server/internal/services/consoleproxy.go index f960c6a..d9e1a37 100644 --- a/server/internal/services/consoleproxy.go +++ b/server/internal/services/consoleproxy.go @@ -1,12 +1,17 @@ package services import ( + "context" + "encoding/json" "errors" "fmt" "log" "net" "os" + "sync" + "time" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/proxy" ) @@ -16,23 +21,51 @@ import ( // so this is fatal rather than a degraded mode. var ErrAgentOffline = errors.New("agent is not connected") -// ConsoleProxy is a pending relay: a bound listener guacd can dial and a -// dispatched command telling the agent to meet it. +// A console session spans two processes once there is more than one replica. +// +// The browser's WebSocket lands on an arbitrary pod. The agent's ProxyStream +// lands on the pod holding that agent's command stream. The relay listener has +// to be on the latter — that is the only process that can match an incoming +// ProxyStream to a waiting listener — while guacd is dialled from the former. +// +// So the WebSocket's pod asks, over the bus, for a relay to be bound on the +// agent's pod, and gets back an address to hand to guacd. That address is the +// owner pod's own, which is why it must resolve to a single pod (POD_IP under +// Kubernetes) rather than to the Service, which would send guacd to a pod +// holding no listener roughly (n-1)/n of the time. +// +// Teardown needs no message of its own. When the browser goes away guac closes +// its connection to the relay, the relay sees the read end, and the session +// closes itself — the same path a single-process deployment always took. Only +// the *reason* has to cross back, because the pod that writes the audit event +// is not the pod that observed the failure. + +// How long Close waits for the relay's terminal reason to arrive before giving +// up. The audit event is written immediately afterwards; a reason that has not +// crossed the bus in this long is not going to improve the record by being +// waited for longer. +const proxyEndGrace = 2 * time.Second + +// ConsoleProxy is a relay as seen by the pod serving the WebSocket. type ConsoleProxy struct { ProxyID string Host string Port int - session *proxy.Session + serverID string + + mu sync.Mutex + reason string + closed bool + stop func() + ended chan struct{} } -func (c *ConsoleProxy) Close() { - proxy.Default.Remove(c.ProxyID) - c.session.Close("") +// proxyEnd is the terminal event a relay's owner pod publishes. +type proxyEnd struct { + Reason string `json:"reason"` } -func (c *ConsoleProxy) Reason() string { return c.session.Reason() } - func proxyListenHost() string { if v := os.Getenv("PROXY_LISTEN_HOST"); v != "" { return v @@ -40,7 +73,14 @@ func proxyListenHost() string { return "0.0.0.0" } +// proxyAdvertiseHost is the address guacd will dial to reach a relay bound by +// *this* process. POD_IP wins over the configured value because with several +// replicas the configured value names the Service, and a Service cannot address +// one pod. The chart sets POD_IP from the downward API. func proxyAdvertiseHost() string { + if v := os.Getenv("POD_IP"); v != "" { + return v + } if v := os.Getenv("PROXY_ADVERTISE_HOST"); v != "" { return v } @@ -74,17 +114,55 @@ func guacdHosts(addr string) []string { return ips } -// DispatchOpenProxy tells the agent to dial its own loopback on port and relay -// it back under proxyID. -func DispatchOpenProxy(serverID, proxyID string, port uint32) error { - return Dispatcher.dispatch(serverID, &pb.ServerCommand{ - CommandId: proxyID, - OpenProxy: &pb.OpenProxyCmd{ProxyId: proxyID, Port: port}, - }) +// localRelay is a listener bound by this process on behalf of a remote request. +type localRelay struct { + proxyID string + host string + port int + session *proxy.Session } -// OpenConsoleProxy binds a relay listener, registers it, and asks the agent to -// connect. The caller must Close the result. +// openLocalRelay binds a listener here and registers it, so the agent's +// ProxyStream — which will arrive at this process — can be matched to it. +// Called on the owner pod, from the dispatch handler. +func openLocalRelay(instanceID, serverID, proxyID string) (*localRelay, error) { + sess, err := proxy.NewSession(proxyListenHost(), guacdHosts(guacdAddr())) + if err != nil { + return nil, err + } + + sess.OnEnd(func(reason string) { + ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout) + defer cancel() + if _, err := bus.Publish(ctx, bus.ProxyEndChannel+proxyID, proxyEnd{Reason: reason}); err != nil { + log.Printf("proxy: publish end for %s: %v", proxyID, err) + } + }) + + proxy.Default.Add(&proxy.Entry{ + ProxyID: proxyID, + InstanceID: instanceID, + ServerID: serverID, + Session: sess, + }) + + return &localRelay{ + proxyID: proxyID, + host: proxyAdvertiseHost(), + port: sess.Port(), + session: sess, + }, nil +} + +// abandon tears down a relay that was bound but whose command never reached the +// agent, so the listener does not sit out its rendezvous timeout for nothing. +func (r *localRelay) abandon() { + proxy.Default.Remove(r.proxyID) + r.session.Close("dispatch_failed") +} + +// OpenConsoleProxy asks the pod holding serverID's stream to bind a relay and +// tell the agent to meet it. The caller must Close the result. func OpenConsoleProxy(instanceID, serverID string, targetPort int) (*ConsoleProxy, error) { if !Dispatcher.IsConnected(serverID) { return nil, ErrAgentOffline @@ -95,28 +173,85 @@ func OpenConsoleProxy(instanceID, serverID string, targetPort int) (*ConsoleProx return nil, fmt.Errorf("generate proxy id: %w", err) } - sess, err := proxy.NewSession(proxyListenHost(), guacdHosts(guacdAddr())) + // Subscribed before the relay is asked for: a relay that fails immediately + // (the agent never claims it, the dial is refused) publishes its reason at + // once, and that reason is the whole content of the audit event. + cp := &ConsoleProxy{ProxyID: proxyID, serverID: serverID, ended: make(chan struct{})} + endCtx, endCancel := context.WithCancel(context.Background()) + ends, unsub, err := bus.Subscribe(endCtx, bus.ProxyEndChannel+proxyID) if err != nil { - return nil, err + endCancel() + return nil, fmt.Errorf("subscribe relay end: %w", err) } + cp.stop = func() { + endCancel() + unsub() + } + go cp.watchEnd(ends) - proxy.Default.Add(&proxy.Entry{ - ProxyID: proxyID, - InstanceID: instanceID, - ServerID: serverID, - Session: sess, + ack, err := Dispatcher.send(CommandEnvelope{ + ServerID: serverID, + Command: &pb.ServerCommand{ + CommandId: proxyID, + OpenProxy: &pb.OpenProxyCmd{ProxyId: proxyID, Port: uint32(targetPort)}, + }, + Proxy: &ProxyRelayRequest{InstanceID: instanceID, ProxyID: proxyID}, }) - - if err := DispatchOpenProxy(serverID, proxyID, uint32(targetPort)); err != nil { - proxy.Default.Remove(proxyID) - sess.Close("dispatch_failed") + if err != nil { + cp.stop() + if errors.Is(err, ErrAgentNotConnected) { + return nil, ErrAgentOffline + } return nil, err } + if ack.ProxyHost == "" || ack.ProxyPort == 0 { + cp.stop() + return nil, fmt.Errorf("relay opened without an address") + } - return &ConsoleProxy{ - ProxyID: proxyID, - Host: proxyAdvertiseHost(), - Port: sess.Port(), - session: sess, - }, nil + cp.Host = ack.ProxyHost + cp.Port = ack.ProxyPort + return cp, nil +} + +func (c *ConsoleProxy) watchEnd(ends <-chan []byte) { + defer close(c.ended) + b, ok := <-ends + if !ok { + return + } + var e proxyEnd + if err := json.Unmarshal(b, &e); err != nil { + return + } + c.mu.Lock() + if c.reason == "" { + c.reason = e.Reason + } + c.mu.Unlock() +} + +// Close waits briefly for the relay's terminal reason, then releases the +// subscription. It is safe to call more than once. +func (c *ConsoleProxy) Close() { + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return + } + c.closed = true + c.mu.Unlock() + + select { + case <-c.ended: + case <-time.After(proxyEndGrace): + } + c.stop() +} + +// Reason reports why the relay ended, empty if it ended cleanly. +func (c *ConsoleProxy) Reason() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.reason } diff --git a/server/internal/services/dispatch.go b/server/internal/services/dispatch.go index 9b70bcf..4369d50 100644 --- a/server/internal/services/dispatch.go +++ b/server/internal/services/dispatch.go @@ -1,63 +1,253 @@ package services import ( + "context" "encoding/json" + "errors" "fmt" + "log" "net/http" "strings" - "sync" + "time" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb" "github.com/google/uuid" ) -type commandDispatcher struct { - mu sync.RWMutex - channels map[string]chan *pb.ServerCommand +// Commands travel over the bus even when the sender is also the owner. +// +// An agent's CommandStream terminates on exactly one process, and with several +// replicas that is almost never the process handling the REST request that +// wants to talk to it. Publishing unconditionally — rather than checking for a +// local stream first and falling back — means one code path, exercised on every +// deployment including the single-replica ones, instead of a rare cross-pod +// path that only fails in production. +const ( + // How long a caller waits for the owning pod to acknowledge. Generous + // enough to cross a loaded cluster, short enough that a REST handler + // answering 503 does not look hung. + dispatchAckTimeout = 5 * time.Second + + // Presence must outlive a renew or two, or a momentarily slow pod would + // look offline and its agent would be declared unreachable. + presenceTTL = 30 * time.Second + presenceRenew = 10 * time.Second +) + +// CommandEnvelope is what actually crosses the bus. It is the command plus the +// small amount of context the owning pod needs to act on it locally. +type CommandEnvelope struct { + ServerID string `json:"server_id"` + Command *pb.ServerCommand `json:"command"` + ReplyTo string `json:"reply_to"` + Log *LogRequest `json:"log,omitempty"` + Proxy *ProxyRelayRequest `json:"proxy,omitempty"` } -var Dispatcher = &commandDispatcher{ - channels: make(map[string]chan *pb.ServerCommand), +// LogRequest asks the owner pod to open a step log before it dispatches. +// +// Step output arrives on the owner pod's gRPC stream, so that is where it is +// masked and written. Shipping raw output back to the run's pod first would put +// unmasked bytes on the bus for no gain. +type LogRequest struct { + RunID string `json:"run_id"` + ServerID string `json:"server_id"` + Mask []string `json:"mask,omitempty"` } -func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand { - ch := make(chan *pb.ServerCommand, 16) - d.mu.Lock() - d.channels[serverID] = ch - d.mu.Unlock() - return ch +// ProxyRelayRequest asks the owner pod to bind a console relay listener and +// register it before dispatching OpenProxyCmd. +// +// The listener has to live on the owner pod: the agent's ProxyStream arrives +// there, and only there can it be matched to a waiting listener. The pod +// serving the browser's WebSocket learns the address from the ack and hands +// that to guacd. +type ProxyRelayRequest struct { + InstanceID string `json:"instance_id"` + ProxyID string `json:"proxy_id"` } -func (d *commandDispatcher) Disconnect(serverID string) { - d.mu.Lock() - delete(d.channels, serverID) - d.mu.Unlock() +// CommandAck is the owner pod's answer. +type CommandAck struct { + OK bool `json:"ok"` + Error string `json:"error,omitempty"` + Node string `json:"node,omitempty"` + ProxyHost string `json:"proxy_host,omitempty"` + ProxyPort int `json:"proxy_port,omitempty"` } +type commandDispatcher struct{} + +var Dispatcher = &commandDispatcher{} + +// ErrAgentNotConnected is returned when no pod holds the agent's stream. +var ErrAgentNotConnected = errors.New("agent is not connected") + +// Serve subscribes this pod to serverID's command channel and claims presence +// for it, returning the channel the gRPC handler should send from and a +// teardown function. +// +// send is the local delivery: it is called on the bus goroutine and must not +// block for long, which is why it pushes onto a buffered channel rather than +// writing to the gRPC stream directly. +func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan *pb.ServerCommand, func()) { + out := make(chan *pb.ServerCommand, 16) + + envelopes, unsub, err := bus.Subscribe(ctx, bus.CommandChannel+serverID) + if err != nil { + // Without a subscription this pod cannot receive commands for the + // agent. Claiming presence anyway would advertise a stream nobody can + // reach, which is worse than the agent appearing offline. + log.Printf("dispatch: subscribe for %s failed, commands will not reach it: %v", serverID, err) + close(out) + return out, func() {} + } + + runCtx, cancel := context.WithCancel(ctx) + + if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil { + log.Printf("dispatch: claim presence for %s: %v", serverID, err) + } + go func() { + t := time.NewTicker(presenceRenew) + defer t.Stop() + for { + select { + case <-runCtx.Done(): + return + case <-t.C: + if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil { + log.Printf("dispatch: renew presence for %s: %v", serverID, err) + } + } + } + }() + + go func() { + for { + select { + case <-runCtx.Done(): + return + case raw, ok := <-envelopes: + if !ok { + return + } + d.handleEnvelope(runCtx, raw, out) + } + } + }() + + return out, func() { + cancel() + unsub() + relCtx, relCancel := context.WithTimeout(context.Background(), 2*time.Second) + if err := bus.ClearPresence(relCtx, serverID); err != nil { + log.Printf("dispatch: release presence for %s: %v", serverID, err) + } + relCancel() + } +} + +// handleEnvelope performs the owner-pod side of a dispatch: any local setup the +// command needs, then queueing it for the stream, then the ack. +func (d *commandDispatcher) handleEnvelope(ctx context.Context, raw []byte, out chan *pb.ServerCommand) { + var env CommandEnvelope + if err := json.Unmarshal(raw, &env); err != nil { + log.Printf("dispatch: undecodable envelope: %v", err) + return + } + if env.Command == nil || env.ReplyTo == "" { + return + } + + ack := CommandAck{OK: true, Node: bus.NodeID()} + + if env.Log != nil { + if err := StepLogs.Open(env.Command.CommandId, env.Log.RunID, env.Log.ServerID, env.Log.Mask); err != nil { + log.Printf("dispatch: open step log for %s: %v", env.Command.CommandId, err) + } + } + + var relay *localRelay + if env.Proxy != nil { + r, err := openLocalRelay(env.Proxy.InstanceID, env.ServerID, env.Proxy.ProxyID) + if err != nil { + ack = CommandAck{OK: false, Error: err.Error(), Node: bus.NodeID()} + } else { + relay = r + ack.ProxyHost = r.host + ack.ProxyPort = r.port + } + } + + if ack.OK { + select { + case out <- env.Command: + default: + ack = CommandAck{OK: false, Error: "command queue full", Node: bus.NodeID()} + if relay != nil { + relay.abandon() + } + } + } + + if err := bus.Reply(ctx, env.ReplyTo, ack); err != nil { + log.Printf("dispatch: reply on %s: %v", env.ReplyTo, err) + } +} + +// IsConnected reports whether any pod holds the agent's command stream. func (d *commandDispatcher) IsConnected(serverID string) bool { - d.mu.RLock() - _, ok := d.channels[serverID] - d.mu.RUnlock() - return ok + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + return bus.IsConnected(ctx, serverID) } func (d *commandDispatcher) dispatch(serverID string, cmd *pb.ServerCommand) error { - d.mu.RLock() - ch, ok := d.channels[serverID] - d.mu.RUnlock() - if !ok { - return fmt.Errorf("agent for server %s is not connected", serverID) - } - select { - case ch <- cmd: - return nil - default: - return fmt.Errorf("command queue full for server %s", serverID) - } + _, err := d.send(CommandEnvelope{ServerID: serverID, Command: cmd}) + return err } -func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error { - return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd}) +// send publishes the envelope and waits for the owning pod's ack. A missing +// responder and a refusing responder are both errors: in neither case did the +// command reach the agent. +func (d *commandDispatcher) send(env CommandEnvelope) (CommandAck, error) { + if env.Command == nil || env.Command.CommandId == "" { + return CommandAck{}, fmt.Errorf("command id is required") + } + env.ReplyTo = bus.AckChannelFor(env.Command.CommandId) + + ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout) + defer cancel() + + raw, err := bus.Request(ctx, bus.CommandChannel+env.ServerID, env.ReplyTo, env, dispatchAckTimeout) + if err != nil { + if errors.Is(err, bus.ErrNoResponder) { + return CommandAck{}, fmt.Errorf("%w: %s", ErrAgentNotConnected, env.ServerID) + } + return CommandAck{}, err + } + + var ack CommandAck + if err := json.Unmarshal(raw, &ack); err != nil { + return CommandAck{}, fmt.Errorf("undecodable ack: %w", err) + } + if !ack.OK { + return ack, fmt.Errorf("agent for server %s: %s", env.ServerID, ack.Error) + } + return ack, nil +} + +// DispatchRunStep sends a step to the agent and asks the owning pod to capture +// its output. mask holds the secret values that must not reach the log. +func DispatchRunStep(serverID, commandID, runID string, mask []string, cmd *pb.RunStepCmd) error { + _, err := Dispatcher.send(CommandEnvelope{ + ServerID: serverID, + Command: &pb.ServerCommand{CommandId: commandID, RunStep: cmd}, + Log: &LogRequest{RunID: runID, ServerID: serverID, Mask: mask}, + }) + return err } func DispatchCleanupWorkspace(serverID, workspaceID string) { diff --git a/server/internal/services/steplogs.go b/server/internal/services/steplogs.go index d45cacb..884467d 100644 --- a/server/internal/services/steplogs.go +++ b/server/internal/services/steplogs.go @@ -2,9 +2,8 @@ package services import ( "bytes" + "context" "log" - "os" - "path/filepath" "strings" "sync" "time" @@ -12,47 +11,130 @@ import ( "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db" "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" ) -func WorkflowLogDir() string { - dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR") - if dir == "" { - dir = filepath.Join("data", "workflow-logs") - } - _ = os.MkdirAll(dir, 0700) - return dir -} +// Workflow run logs live in MongoDB, not on disk. +// +// They used to be a file per (run, server) under a volume. That is the cheapest +// possible writer, and it is wrong the moment there is more than one server +// process: step output arrives on whichever pod holds the agent's stream, the +// run's markers are written by whichever pod started the run, and the browser +// asks for the log through whichever pod the load balancer picked. Three pods, +// one file, one local disk — two of them see an empty log. +// +// Mongo makes every pod an equal reader and writer, which is the property that +// matters. It costs writes on the hot path, so the writer batches (see +// stepLogWriter) and both caps below exist to keep a runaway step from turning +// a workflow into a database incident. +const ( + // A single line longer than this is truncated. Base64 blobs and minified + // output are the usual cause; nobody reads column 9000 of a log line. + maxLogLineBytes = 8 * 1024 -func ServerRunLogPath(runID, serverID string) string { - return filepath.Join(WorkflowLogDir(), runID, serverID+".log") + // A single (run, server) log stops accepting lines here, with one final + // marker saying so. 200k lines is far past what anyone reads and still + // only a few tens of MB. Without a cap, `yes` in a step fills the database. + maxLogLines = 200_000 + + // The writer flushes on whichever comes first. Batching is what keeps a + // chatty step to a handful of writes a second instead of one per line. + logFlushLines = 128 + logFlushInterval = 250 * time.Millisecond +) + +const logLinesCol = "workflow_log_lines" +const logSeqCol = "workflow_log_seq" + +func logCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 10*time.Second) } func logTS() string { return time.Now().UTC().Format("2006-01-02T15:04:05.000") + "Z" } -func AppendMarker(runID, serverID, text string) (int64, error) { - path := ServerRunLogPath(runID, serverID) - if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { - return 0, err +func seqID(runID, serverID string) string { return runID + "/" + serverID } + +// reserveSeq atomically claims n consecutive sequence numbers for a +// (run, server) log and returns the first. +// +// The counter is a document rather than a per-process integer because two pods +// write the same log concurrently: the run's pod emits markers while the +// agent's pod emits step output. Ordering between them is only meaningful if +// they draw from the same counter. +func reserveSeq(ctx context.Context, runID, serverID string, n int) (int64, error) { + var doc struct { + Seq int64 `bson:"seq"` } - f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) + err := db.Col(logSeqCol).FindOneAndUpdate(ctx, + bson.M{"_id": seqID(runID, serverID)}, + bson.M{"$inc": bson.M{"seq": int64(n)}}, + options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After), + ).Decode(&doc) if err != nil { return 0, err } - defer f.Close() - off, _ := f.Seek(0, 2) - if _, err := f.WriteString("[" + logTS() + "] " + text + "\n"); err != nil { - return off, err - } - return off, nil + return doc.Seq - int64(n), nil } +type logLine struct { + RunID string `bson:"run_id"` + ServerID string `bson:"server_id"` + Seq int64 `bson:"seq"` + At time.Time `bson:"at"` + Line string `bson:"line"` +} + +// writeLines reserves a block of sequence numbers and inserts the batch. +func writeLines(ctx context.Context, runID, serverID string, lines []string) (int64, error) { + if len(lines) == 0 { + return 0, nil + } + start, err := reserveSeq(ctx, runID, serverID, len(lines)) + if err != nil { + return 0, err + } + now := time.Now().UTC() + docs := make([]any, 0, len(lines)) + for i, l := range lines { + docs = append(docs, logLine{ + RunID: runID, + ServerID: serverID, + Seq: start + int64(i), + At: now, + Line: l, + }) + } + // Unordered: one rejected document must not discard the rest of the batch. + _, err = db.Col(logLinesCol).InsertMany(ctx, docs, options.InsertMany().SetOrdered(false)) + return start, err +} + +// AppendMarker writes one control line (step banners, retries, run outcome) and +// returns its sequence number, which is what a StepRun's log_offset records so +// the UI can scroll to where a step began. +func AppendMarker(runID, serverID, text string) (int64, error) { + ctx, cancel := logCtx() + defer cancel() + return writeLines(ctx, runID, serverID, []string{"[" + logTS() + "] " + text}) +} + +// stepLogWriter accumulates one command's output, splits it into lines, masks +// secrets and flushes batches to Mongo. type stepLogWriter struct { - mu sync.Mutex - f *os.File + mu sync.Mutex + runID string + serverID string + secrets []string + carry []byte - secrets []string + pending []string + written int + capped bool + + stop chan struct{} + done chan struct{} } type stepLogRegistry struct { @@ -60,18 +142,30 @@ type stepLogRegistry struct { writers map[string]*stepLogWriter } +// The registry stays process-local, and correctly so: a step's output arrives +// on the pod holding that agent's stream, and that is the same pod the +// dispatch envelope asked to open the writer. Nothing here crosses pods — +// only the lines it produces do, by virtue of landing in Mongo. var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)} -func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error { - if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { - return err +func (r *stepLogRegistry) Open(commandID, runID, serverID string, secrets []string) error { + w := &stepLogWriter{ + runID: runID, + serverID: serverID, + secrets: secrets, + stop: make(chan struct{}), + done: make(chan struct{}), } - f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) - if err != nil { - return err - } - w := &stepLogWriter{f: f, secrets: secrets} + go w.flushLoop() + r.mu.Lock() + if old := r.writers[commandID]; old != nil { + // A retry reuses the command ID. Close the previous attempt's writer + // rather than leaking its flush goroutine. + r.mu.Unlock() + old.close() + r.mu.Lock() + } r.writers[commandID] = w r.mu.Unlock() return nil @@ -89,24 +183,22 @@ func (r *stepLogRegistry) Append(commandID string, data []byte) { return } w.mu.Lock() - defer w.mu.Unlock() buf := append(w.carry, data...) for { i := bytes.IndexByte(buf, '\n') if i < 0 { break } - w.writeLine(buf[:i]) + w.queueLine(buf[:i]) buf = buf[i+1:] } w.carry = append([]byte{}, buf...) -} + full := len(w.pending) >= logFlushLines + w.mu.Unlock() -func (w *stepLogWriter) writeLine(line []byte) { - masked := maskBytes(line, w.secrets) - _, _ = w.f.WriteString("[" + logTS() + "] ") - _, _ = w.f.Write(masked) - _, _ = w.f.WriteString("\n") + if full { + w.flush() + } } func (r *stepLogRegistry) Close(commandID string) { @@ -117,13 +209,72 @@ func (r *stepLogRegistry) Close(commandID string) { if w == nil { return } + w.close() +} + +// queueLine masks, truncates and enqueues a line. Caller holds w.mu. +func (w *stepLogWriter) queueLine(line []byte) { + if w.capped { + return + } + if w.written+len(w.pending) >= maxLogLines { + w.capped = true + w.pending = append(w.pending, "["+logTS()+"] [vantage] log truncated: this step exceeded the per-server line limit") + return + } + masked := maskBytes(line, w.secrets) + if len(masked) > maxLogLineBytes { + masked = append(masked[:maxLogLineBytes], []byte(" [truncated]")...) + } + w.pending = append(w.pending, "["+logTS()+"] "+string(masked)) +} + +func (w *stepLogWriter) flush() { + w.mu.Lock() + if len(w.pending) == 0 { + w.mu.Unlock() + return + } + batch := w.pending + w.pending = nil + w.written += len(batch) + runID, serverID := w.runID, w.serverID + w.mu.Unlock() + + ctx, cancel := logCtx() + defer cancel() + if _, err := writeLines(ctx, runID, serverID, batch); err != nil { + // Dropped rather than retried. A log line is not worth stalling the + // step it describes, and a Mongo that cannot take writes has larger + // problems than a missing line. + log.Printf("step log: write %d line(s) for run %s: %v", len(batch), runID, err) + } +} + +func (w *stepLogWriter) flushLoop() { + defer close(w.done) + t := time.NewTicker(logFlushInterval) + defer t.Stop() + for { + select { + case <-w.stop: + return + case <-t.C: + w.flush() + } + } +} + +func (w *stepLogWriter) close() { + close(w.stop) + <-w.done w.mu.Lock() - defer w.mu.Unlock() if len(w.carry) > 0 { - w.writeLine(w.carry) + w.queueLine(w.carry) w.carry = nil } - _ = w.f.Close() + w.mu.Unlock() + w.flush() } func maskBytes(b []byte, secrets []string) []byte { @@ -137,86 +288,132 @@ func maskBytes(b []byte, secrets []string) []byte { return []byte(s) } -func StartLogSweeper() { +// ReadServerRunLog returns the log for one server in a run, from sequence +// number after onwards, along with the highest sequence returned. +// +// Callers page with it rather than fetching everything: the live stream asks +// repeatedly for what is new, and the plain-text endpoint walks the whole log +// in chunks so a very large one is never held in memory whole. +func ReadServerRunLog(runID, serverID string, after int64, limit int) ([]string, int64, error) { + ctx, cancel := logCtx() + defer cancel() + + cur, err := db.Col(logLinesCol).Find(ctx, + bson.M{"run_id": runID, "server_id": serverID, "seq": bson.M{"$gt": after}}, + options.Find().SetSort(bson.D{{Key: "seq", Value: 1}}).SetLimit(int64(limit)), + ) + if err != nil { + return nil, after, err + } + defer cur.Close(ctx) + + lines := make([]string, 0, 64) + last := after + for cur.Next(ctx) { + var l logLine + if err := cur.Decode(&l); err != nil { + return lines, last, err + } + lines = append(lines, l.Line) + last = l.Seq + } + return lines, last, cur.Err() +} + +// HasServerRunLog reports whether any line exists, so a log read can answer 404 +// rather than 200 with an empty body. +func HasServerRunLog(runID, serverID string) bool { + ctx, cancel := logCtx() + defer cancel() + err := db.Col(logLinesCol).FindOne(ctx, bson.M{"run_id": runID, "server_id": serverID}).Err() + return err == nil +} + +func StartLogSweeper(ctx context.Context) { go func() { sweepLogs() t := time.NewTicker(time.Hour) defer t.Stop() - for range t.C { - sweepLogs() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + sweepLogs() + } } }() } +// sweepLogs deletes the lines of finished runs past their instance's retention. +// +// It walks workflow_runs rather than the log collection: retention is a +// property of the run (its instance, its finish time), and a run row is the +// only place both are recorded. func sweepLogs() { - base := WorkflowLogDir() - entries, err := os.ReadDir(base) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + cur, err := db.Col("workflow_runs").Find(ctx, + bson.M{"finished_at": bson.M{"$ne": nil}}, + options.Find().SetProjection(bson.M{"run_id": 1, "instance_id": 1, "finished_at": 1}), + ) if err != nil { + log.Printf("log sweep: list runs: %v", err) return } + defer cur.Close(ctx) + cache := map[string]int{} now := time.Now() - for _, e := range entries { - if !e.IsDir() { - continue + for cur.Next(ctx) { + var run struct { + RunID string `bson:"run_id"` + InstanceID string `bson:"instance_id"` + FinishedAt *time.Time `bson:"finished_at"` } - runID := e.Name() - dir := filepath.Join(base, runID) - - instanceID, finishedAt, found, err := runRetentionInfo(runID) - if err != nil { - - log.Printf("log sweep: retention lookup failed for run %s: %v", runID, err) - continue - } - if found && finishedAt == nil { + if err := cur.Decode(&run); err != nil || run.FinishedAt == nil { continue } - days, ok := cache[instanceID] + days, ok := cache[run.InstanceID] if !ok { days = defaultRetentionDays - if instanceID != "" { - if v, err := GetWorkflowLogRetentionDays(instanceID); err == nil { + if run.InstanceID != "" { + if v, err := GetWorkflowLogRetentionDays(run.InstanceID); err == nil { days = v } } - cache[instanceID] = days + cache[run.InstanceID] = days } if days <= 0 { continue } - cutoff := now.AddDate(0, 0, -days) - - if found { - if finishedAt.Before(cutoff) { - _ = os.RemoveAll(dir) - } + if !run.FinishedAt.Before(now.AddDate(0, 0, -days)) { continue } - if fi, e := os.Stat(dir); e == nil && fi.ModTime().Before(cutoff) { - _ = os.RemoveAll(dir) + if _, err := db.Col(logLinesCol).DeleteMany(ctx, bson.M{"run_id": run.RunID}); err != nil { + log.Printf("log sweep: delete lines for run %s: %v", run.RunID, err) + continue } + _, _ = db.Col(logSeqCol).DeleteMany(ctx, bson.M{"_id": bson.M{"$regex": "^" + run.RunID + "/"}}) } } const defaultRetentionDays = 30 -func runRetentionInfo(runID string) (string, *time.Time, bool, error) { - ctx, cancel := wfCtx() +// EnsureLogIndexes builds the indexes the log store depends on. The compound +// index is not an optimisation: every read is a range scan over it, and without +// it a log read collection-scans every line in the database. +func EnsureLogIndexes() error { + ctx, cancel := logCtx() defer cancel() - var run struct { - InstanceID string `bson:"instance_id"` - FinishedAt *time.Time `bson:"finished_at"` + if _, err := db.Col(logLinesCol).Indexes().CreateOne(ctx, mongo.IndexModel{ + Keys: bson.D{{Key: "run_id", Value: 1}, {Key: "server_id", Value: 1}, {Key: "seq", Value: 1}}, + }); err != nil { + return err } - err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run) - if err == mongo.ErrNoDocuments { - return "", nil, false, nil - } - if err != nil { - return "", nil, false, err - } - return run.InstanceID, run.FinishedAt, true, nil + return nil } diff --git a/server/internal/services/stepresults.go b/server/internal/services/stepresults.go index ae476b9..e8872c0 100644 --- a/server/internal/services/stepresults.go +++ b/server/internal/services/stepresults.go @@ -1,43 +1,75 @@ package services import ( - "sync" + "context" + "encoding/json" + "log" + "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus" "gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb" ) -type stepResultRegistry struct { - mu sync.Mutex - pending map[string]chan *pb.StepResult -} - -var StepResults = &stepResultRegistry{pending: make(map[string]chan *pb.StepResult)} - -func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult { - ch := make(chan *pb.StepResult, 1) - r.mu.Lock() - r.pending[commandID] = ch - r.mu.Unlock() - return ch -} - -func (r *stepResultRegistry) Cancel(commandID string) { - r.mu.Lock() - delete(r.pending, commandID) - r.mu.Unlock() +// Step results travel back over the bus for the same reason commands travel out +// over it: the pod driving a workflow run and the pod holding the agent's +// stream are two different processes, and a map in one of them cannot be read +// by the other. +// +// Await subscribes before the command is dispatched (see dispatchAndWait), +// which is what stops a fast agent from answering into a channel nobody is +// listening on yet. + +type stepResultRegistry struct{} + +var StepResults = &stepResultRegistry{} + +// Await subscribes to a command's result channel. The returned cancel function +// must be called once the caller is done, whether a result arrived or not — +// it is what releases the Redis subscription. +func (r *stepResultRegistry) Await(commandID string) (<-chan *pb.StepResult, func()) { + out := make(chan *pb.StepResult, 1) + + ctx, cancel := context.WithCancel(context.Background()) + raw, unsub, err := bus.Subscribe(ctx, bus.ResultChannel+commandID) + if err != nil { + log.Printf("step results: subscribe for %s: %v", commandID, err) + cancel() + close(out) + return out, func() {} + } + + go func() { + defer close(out) + select { + case <-ctx.Done(): + return + case b, ok := <-raw: + if !ok { + return + } + var res pb.StepResult + if err := json.Unmarshal(b, &res); err != nil { + log.Printf("step results: undecodable result for %s: %v", commandID, err) + return + } + out <- &res + } + }() + + return out, func() { + cancel() + unsub() + } } +// Deliver publishes a result received from an agent. Called on the pod holding +// that agent's stream, which is not usually the pod waiting for it. func (r *stepResultRegistry) Deliver(res *pb.StepResult) { - if res == nil { + if res == nil || res.CommandId == "" { return } - r.mu.Lock() - ch, ok := r.pending[res.CommandId] - if ok { - delete(r.pending, res.CommandId) - } - r.mu.Unlock() - if ok { - ch <- res + ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout) + defer cancel() + if _, err := bus.Publish(ctx, bus.ResultChannel+res.CommandId, res); err != nil { + log.Printf("step results: publish for %s: %v", res.CommandId, err) } } diff --git a/server/internal/services/workflow_runner.go b/server/internal/services/workflow_runner.go index 53a8150..34cb35e 100644 --- a/server/internal/services/workflow_runner.go +++ b/server/internal/services/workflow_runner.go @@ -240,7 +240,6 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep marker := fmt.Sprintf("===== step %d/%d: %s (%s) =====", step.Order+1, len(steps), step.Name, step.Interpreter) offset, _ := AppendMarker(runID, serverID, marker) - logPath := ServerRunLogPath(runID, serverID) secretsSlice := secretValues(secretVals) commandID := uuid.New().String() @@ -250,15 +249,17 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep _, _ = AppendMarker(runID, serverID, fmt.Sprintf("retry %d/%d after failure", attempts-1, maxAttempts-1)) } - _ = StepLogs.Open(commandID, logPath, secretsSlice) - res = dispatchAndWait(serverID, commandID, &pb.RunStepCmd{ + // The log writer is opened by the pod that owns this agent's + // stream, not here: that is where the output arrives, and the mask + // list travels with the dispatch so it is applied before anything + // is stored. + res = dispatchAndWait(serverID, commandID, runID, secretsSlice, &pb.RunStepCmd{ Interpreter: step.Interpreter, Script: step.Script, Env: cmdEnv, TimeoutSeconds: 0, WorkspaceId: runID, }) - StepLogs.Close(commandID) if res != nil && res.ExitCode == 0 { break } @@ -322,10 +323,14 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep }) } -func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepResult { - ch := StepResults.Await(commandID) - if err := DispatchRunStep(serverID, commandID, cmd); err != nil { - StepResults.Cancel(commandID) +// dispatchAndWait subscribes to the result before dispatching, because the two +// happen on different pods and an agent that answers quickly would otherwise +// publish into a channel this process had not yet joined. +func dispatchAndWait(serverID, commandID, runID string, mask []string, cmd *pb.RunStepCmd) *pb.StepResult { + ch, done := StepResults.Await(commandID) + defer done() + + if err := DispatchRunStep(serverID, commandID, runID, mask, cmd); err != nil { return &pb.StepResult{ExitCode: 1, Stderr: "[vantage] dispatch failed: " + err.Error()} } wait := time.Duration(cmd.TimeoutSeconds)*time.Second + stepDispatchGrace @@ -333,10 +338,12 @@ func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepRes wait = 30*time.Minute + stepDispatchGrace } select { - case res := <-ch: + case res, ok := <-ch: + if !ok || res == nil { + return &pb.StepResult{ExitCode: 1, Stderr: "[vantage] lost the result channel before the agent answered"} + } return res case <-time.After(wait): - StepResults.Cancel(commandID) return &pb.StepResult{ExitCode: 124, Stderr: "[vantage] timed out waiting for agent result"} } } diff --git a/server/internal/services/workflows.go b/server/internal/services/workflows.go index 0c6c516..4a3fabb 100644 --- a/server/internal/services/workflows.go +++ b/server/internal/services/workflows.go @@ -60,6 +60,9 @@ func EnsureWorkflowIndexes() error { }); err != nil { return err } + if err := EnsureLogIndexes(); err != nil { + return err + } _, err := db.Col("workflow_runs").Indexes().CreateOne(ctx, mongo.IndexModel{ Keys: bson.D{{Key: "run_id", Value: 1}}, Options: options.Index().SetUnique(true), }) diff --git a/web/app/healthz/route.ts b/web/app/healthz/route.ts new file mode 100644 index 0000000..83dd595 --- /dev/null +++ b/web/app/healthz/route.ts @@ -0,0 +1,15 @@ +import { NextResponse } from "next/server"; + +// Liveness and readiness for the Next server itself. Deliberately at /healthz +// and not /api/healthz: next.config.ts rewrites the whole of /api to the Go +// server, so a probe there would report the backend's health instead of this +// process's — and would keep passing while this pod was wedged. +// +// It answers without touching the backend on purpose. web is stateless; a +// backend outage must not take every web replica out of its Service as well, +// which would turn one failure into two. +export const dynamic = "force-dynamic"; + +export function GET() { + return NextResponse.json({ status: "ok" }); +}