Insert NATS JetStream log bus + S3 raw archive

Rework the logging pipeline around a durable message bus so logs survive a
ClickHouse outage, can be replayed after a bad transform, and fan out to
multiple independent consumers. Add long-term raw-log backup to S3.

Topology becomes edge -> JetStream -> consumers -> sinks:
- Dedicated JetStream NATS cluster (3 replicas, file storage) in the logging
  namespace. Deliberately separate from app messaging (streamstack) for
  blast-radius isolation. Stream LOGS (subjects logs.>, retention=limits, 40GiB
  / 72h) is the outage buffer; durable consumers give independent offsets.
- Edge publishers (thin): the k8s DaemonSet and a new VM-ingest Deployment
  (HTTP NDJSON front door behind the logs-ingest Gateway) publish into JetStream
  (logs.k8s.<ns>.<container> / logs.vm.<host>). No parsing on the edge.
- Transform tier (StatefulSet): pulls the whole stream via the durable
  `transform` consumer, routes by subject, shapes, and remains the sole
  ClickHouse writer. Its disk buffer shrinks (JetStream is the outage buffer).
- Archiver (Deployment): its OWN durable `archiver` consumer (independent
  offsets — archive lag never affects the ClickHouse path) writes RAW,
  pre-transform events to a Ceph RGW S3 bucket (cephrgw-operator ObjectStoreUser
  + Bucket + BucketAccess) as gzipped NDJSON keyed by raw/<subject>/YYYY/MM/DD/.
  Default subject filter is Vault audit (logs.k8s.vault.>), configurable.

Auth: distinct NATS users (producer publish-only, consumer pull+ack, admin for
the stream/consumer bootstrap Job) with passwords from Vault (nats-auth Secret);
S3 creds from the BucketAccess Secret. Streams/consumers are provisioned by an
idempotent PostSync bootstrap Job.

Add local kubeconform schemas for the ceph.unkin.net CRDs (datreeio lacks them)
and extend the vector-test CI to cover the agent, VM-ingest and archiver
configs. Verified end-to-end locally: NATS ACLs, vector JetStream publish, and
durable-consumer pull+ack (at-least-once) all work.

Claude-Session: https://claude.ai/code/session_015ur3i7D2azsMAWTSVABApv

apps/base/logging/cephrgw.yaml

@@ -0,0 +1,43 @@
+---
+# S3 bucket (Ceph RGW) for the long-term raw-log archive, provisioned by the
+# in-estate cephrgw-operator. The archiver Vector deployment writes here.
+apiVersion: ceph.unkin.net/v1alpha1
+kind: ObjectStoreUser
+metadata:
+  name: logs-archive-owner
+  namespace: logging
+spec:
+  displayName: "Logging raw-archive bucket owner"
+  maxBuckets: 5
+  quota:
+    enabled: true
+    # 5 TiB soft cap; real retention is enforced RGW-side by a bucket lifecycle
+    # policy (see PR notes) — the operator does not manage lifecycle.
+    maxSizeBytes: 5497558138880
+---
+apiVersion: ceph.unkin.net/v1alpha1
+kind: Bucket
+metadata:
+  name: logs-archive
+  namespace: logging
+spec:
+  bucketName: logs-archive
+  ownerRef: logs-archive-owner
+  versioning: false
+  tags:
+    app: logging
+    purpose: raw-log-archive
+  # Keep the bucket (and its objects) if this CR is ever deleted.
+  retainOnDelete: true
+---
+apiVersion: ceph.unkin.net/v1alpha1
+kind: BucketAccess
+metadata:
+  name: logs-archive-writer
+  namespace: logging
+spec:
+  bucketRef: logs-archive
+  level: read-write
+  # Operator writes AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY (+ RGW_UID,
+  # S3_ENDPOINT, BUCKET_NAME) into this Secret; the archiver consumes it.
+  secretName: logs-archive-s3

apps/base/logging/httproute.yaml

@@ -46,7 +46,7 @@ spec:
     - backendRefs:
         - group: ""
           kind: Service
-          name: vector-aggregator
+          name: vector-vm-ingest
           port: 8080
           weight: 1
       matches:

apps/base/logging/kustomization.yaml

@@ -8,14 +8,31 @@ resources:
   - vaultstaticsecret.yaml
   - clickhouseinstallation.yaml
   - job_clickhouse-schema.yaml
+  - nats-bootstrap-job.yaml
+  - cephrgw.yaml
   - gateway.yaml
   - httproute.yaml
 
-# Aggregator pipeline is the single source of truth (also validated by
-# `vector test` in CI). Mounted into the aggregator via `existingConfigMaps`.
+# Vector pipelines are the single source of truth (also validated by
+# `vector test` in CI). Mounted into each tier via `existingConfigMaps`.
 configMapGenerator:
+  - name: vector-agent-config
+    files:
+      - agent.yaml=vector/agent.yaml
+    options:
+      disableNameSuffixHash: true
   - name: vector-aggregator-config
     files:
       - aggregator.yaml=vector/aggregator.yaml
     options:
       disableNameSuffixHash: true
+  - name: vector-vm-ingest-config
+    files:
+      - vm-ingest.yaml=vector/vm-ingest.yaml
+    options:
+      disableNameSuffixHash: true
+  - name: vector-archiver-config
+    files:
+      - archiver.yaml=vector/archiver.yaml
+    options:
+      disableNameSuffixHash: true

apps/base/logging/nats-bootstrap-job.yaml

@@ -0,0 +1,122 @@
+---
+# Declarative JetStream provisioning: the LOGS stream + durable consumers.
+# ArgoCD PostSync hook, idempotent (add-or-converge), re-runs each sync.
+#
+# Stream LOGS: file storage, 3 replicas, retention=limits (NOT workqueue) so
+# multiple durable consumers fan out and can independently replay within the
+# window. Sized as the ClickHouse-outage buffer: 40 GiB / 72h (per-replica PVC
+# is 50Gi, see values-nats.yaml). Beyond that window the S3 archive is the
+# long-term replay source.
+#
+# Consumers (independent offsets = true fan-out):
+#   transform  -> whole log stream, feeds the ClickHouse transform tier
+#   archiver   -> configurable security-relevant subset, feeds the S3 archiver.
+#                 Default filter is Vault audit (logs.k8s.vault.>); ADD subjects
+#                 by editing ARCHIVE_SUBJECTS (space-separated -> repeated
+#                 --filter). Exact default set is an open decision for Ben.
+#
+# Runbook (replay):
+#   (a) reprocess from JetStream (within 72h): scale the transform tier to 0,
+#       then `nats consumer rm LOGS transform` and re-run this Job (recreates at
+#       DeliverAll), or `nats consumer edit`/`--replay` from a start seq/time.
+#   (b) long-horizon (beyond JetStream): re-ingest S3 archive objects back
+#       through the transform tier (vector aws_s3 source or a one-shot Job).
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: nats-bootstrap
+  namespace: logging
+  annotations:
+    argocd.argoproj.io/hook: PostSync
+    argocd.argoproj.io/hook-delete-policy: BeforeHookCreation
+  labels:
+    app.kubernetes.io/name: nats-bootstrap
+    app.kubernetes.io/component: bootstrap
+spec:
+  backoffLimit: 20
+  activeDeadlineSeconds: 1800
+  ttlSecondsAfterFinished: 3600
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: nats-bootstrap
+        vector.dev/exclude: "true"
+    spec:
+      restartPolicy: OnFailure
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 1000
+        runAsGroup: 1000
+      containers:
+        - name: nats-bootstrap
+          image: natsio/nats-box:0.18.0
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            capabilities:
+              drop:
+                - ALL
+          env:
+            - name: HOME
+              value: /tmp
+            - name: NATS_URL
+              value: "nats://nats.logging.svc.cluster.local:4222"
+            - name: NATS_ADMIN_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: nats-auth
+                  key: admin_password
+            # Space-separated subject filters for the archiver consumer.
+            - name: ARCHIVE_SUBJECTS
+              value: "logs.k8s.vault.>"
+          command:
+            - /bin/sh
+            - -ec
+            - |
+              export NATS_USER=log-admin NATS_PASSWORD="$NATS_ADMIN_PASSWORD"
+              echo "Waiting for NATS + JetStream ..."
+              until nats --server "$NATS_URL" account info >/dev/null 2>&1; do
+                echo "  not ready, retry in 5s"; sleep 5
+              done
+
+              echo "Ensuring stream LOGS ..."
+              nats stream add LOGS \
+                --subjects='logs.>' --storage=file --replicas=3 \
+                --retention=limits --discard=old \
+                --max-age=72h --max-bytes=42949672960 \
+                --max-msgs=-1 --max-msgs-per-subject=-1 --max-msg-size=-1 \
+                --max-consumers=-1 --dupe-window=2m --defaults 2>/dev/null \
+              || nats stream edit -f LOGS \
+                --subjects='logs.>' --discard=old \
+                --max-age=72h --max-bytes=42949672960 --dupe-window=2m
+
+              echo "Ensuring consumer transform ..."
+              nats consumer add LOGS transform \
+                --pull --filter='logs.>' --deliver=all --ack=explicit \
+                --max-deliver=-1 --replay=instant --defaults 2>/dev/null \
+              || echo "  transform already exists"
+
+              echo "Ensuring consumer archiver (filters: $ARCHIVE_SUBJECTS) ..."
+              filter_args=""
+              for s in $ARCHIVE_SUBJECTS; do filter_args="$filter_args --filter=$s"; done
+              # shellcheck disable=SC2086
+              nats consumer add LOGS archiver \
+                --pull $filter_args --deliver=all --ack=explicit \
+                --max-deliver=-1 --replay=instant --defaults 2>/dev/null \
+              || echo "  archiver already exists"
+
+              echo "Done."
+              nats stream info LOGS
+          resources:
+            requests:
+              cpu: 50m
+              memory: 64Mi
+            limits:
+              cpu: 500m
+              memory: 256Mi
+          volumeMounts:
+            - name: tmp
+              mountPath: /tmp
+      volumes:
+        - name: tmp
+          emptyDir: {}

apps/base/logging/vaultstaticsecret.yaml

@@ -26,3 +26,25 @@ spec:
   refreshAfter: 5m
   type: kv-v2
   vaultAuthRef: default
+---
+# NATS JetStream auth. Distinct passwords for the producer (edge), consumer
+# (transform tier + archiver) and admin (bootstrap Job) users. Seed once:
+#   for k in admin producer consumer; do declare P_$k=$(openssl rand -base64 24); done
+#   vault kv put kv/kubernetes/namespace/logging/default/nats-auth \
+#     admin_password="$P_admin" producer_password="$P_producer" consumer_password="$P_consumer"
+apiVersion: secrets.hashicorp.com/v1beta1
+kind: VaultStaticSecret
+metadata:
+  name: nats-auth
+  namespace: logging
+spec:
+  destination:
+    create: true
+    name: nats-auth
+    overwrite: true
+  hmacSecretData: true
+  mount: kv
+  path: kubernetes/namespace/logging/default/nats-auth
+  refreshAfter: 5m
+  type: kv-v2
+  vaultAuthRef: default

apps/base/logging/vector/agent.yaml

@@ -0,0 +1,43 @@
+---
+# Vector EDGE agent pipeline (DaemonSet) — thin publisher, single source of
+# truth. Mounted via existingConfigMaps (NOT the chart's customConfig, whose
+# Helm `tpl` pass collides with Vector's own {{ }} / ${ } syntax). Tails all pod
+# logs, attaches only routing tokens, publishes to JetStream. No parsing.
+data_dir: /vector-data-dir
+
+api:
+  enabled: false
+
+sources:
+  kubernetes_logs:
+    type: kubernetes_logs
+
+transforms:
+  # Routing metadata only: NATS-subject-safe namespace + container tokens.
+  keymeta:
+    type: remap
+    inputs:
+      - kubernetes_logs
+    source: |
+      ns = to_string(.kubernetes.pod_namespace || "unknown") ?? "unknown"
+      .ns_token = replace(ns, r'[^a-zA-Z0-9_-]', "_")
+      cont = to_string(.kubernetes.container_name || "unknown") ?? "unknown"
+      .cont_token = replace(cont, r'[^a-zA-Z0-9_-]', "_")
+
+sinks:
+  to_jetstream:
+    type: nats
+    inputs:
+      - keymeta
+    url: nats://nats.logging.svc.cluster.local:4222
+    connection_name: vector-agent
+    subject: "logs.k8s.{{ ns_token }}.{{ cont_token }}"
+    jetstream:
+      enabled: true
+    auth:
+      strategy: user_password
+      user_password:
+        user: log-producer
+        password: ${NATS_PRODUCER_PASSWORD}
+    encoding:
+      codec: json

apps/base/logging/vector/aggregator-tests.yaml

@@ -3,6 +3,20 @@
 # aggregator.yaml in CI (.woodpecker/vector-test.yaml). This is the pattern the
 # per-app parsing follow-ups extend: add a test per new transform here.
 tests:
+  - name: subject_routes_k8s_vs_vm
+    inputs:
+      - insert_at: route
+        type: log
+        log_fields:
+          subject: "logs.k8s.shop.web"
+          message: "routed"
+    outputs:
+      - extract_from: route.k8s
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.message, "routed")
+
   - name: k8s_log_is_normalised
     inputs:
       - insert_at: k8s_shape

apps/base/logging/vector/aggregator.yaml

@@ -1,16 +1,16 @@
 ---
-# Vector AGGREGATOR pipeline (the "brain") — single source of truth.
+# Vector TRANSFORM tier (the "brain") — single source of truth, also validated
+# by `vector test` in CI. Consumes the whole log stream from JetStream via the
+# durable `transform` consumer (at-least-once; durable offsets tracked by
+# JetStream), routes by subject, normalises into the logs.raw columns, and is
+# the ONLY ClickHouse writer. Per-app parsing is added here as follow-ups:
+# insert a transform and append its id to the clickhouse sink `inputs` — no edge
+# or VM rollout required.
 #
-# This file is the ConfigMap the aggregator StatefulSet runs AND the config
-# `vector test` validates in CI (see .woodpecker/vector-test.yaml, which merges
-# this with aggregator-tests.yaml). The edge tier (k8s DaemonSet + future VM
-# agents) stays thin: it only collects and attaches source metadata, then
-# forwards over the Vector-native protocol. All shaping, routing, enrichment,
-# batching, buffering and the ONLY ClickHouse credentials live here.
-#
-# Per-app pipelines arrive as follow-up tasks and are aggregator-only changes:
-# add a `remap`/`route`/enrichment transform below and append its id to the
-# clickhouse sink `inputs` — no fleet/DaemonSet rollout required.
+# Durability model: JetStream (72h / 40GiB) is the outage buffer. If ClickHouse
+# is down the sink blocks, back-pressure stops acking, and JetStream retains
+# messages for replay. The local disk buffer is small (survives pod restarts of
+# in-flight events only).
 data_dir: /vector-data-dir
 
 api:
@@ -18,30 +18,35 @@ api:
   address: 0.0.0.0:8686
 
 sources:
-  # In-cluster pod logs from the Vector agent DaemonSet (Vector-native proto).
-  from_agents:
-    type: vector
-    address: 0.0.0.0:6000
-
-  # Puppet-managed VM logs over HTTP (NDJSON), exposed via the logs-ingest
-  # Gateway at logs-ingest.k8s.syd1.au.unkin.net.
-  vm_http:
-    type: http_server
-    address: 0.0.0.0:8080
-    path: /
-    method: POST
+  js_in:
+    type: nats
+    url: nats://nats.logging.svc.cluster.local:4222
+    connection_name: vector-transform
+    subject: "logs.>"
+    jetstream:
+      stream: LOGS
+      consumer: transform
+    auth:
+      strategy: user_password
+      user_password:
+        user: log-consumer
+        password: ${NATS_CONSUMER_PASSWORD}
     decoding:
       codec: json
-    framing:
-      method: newline_delimited
 
 transforms:
-  # ---- Default normalisation into the logs.raw columns. This is intentionally
-  # the ONLY shaping today; per-app parsing is added here as follow-ups. ----
+  route:
+    type: route
+    inputs:
+      - js_in
+    route:
+      k8s: 'starts_with(to_string(.subject) ?? "", "logs.k8s.")'
+      vm: 'starts_with(to_string(.subject) ?? "", "logs.vm.")'
+
   k8s_shape:
     type: remap
     inputs:
-      - from_agents
+      - route.k8s
     source: |
       ts = .timestamp || now()
       node = to_string(.kubernetes.pod_node_name || "") ?? ""
@@ -64,10 +69,11 @@ transforms:
         "labels": lbls,
         "fields": {}
       }
+
   vm_shape:
     type: remap
     inputs:
-      - vm_http
+      - route.vm
     source: |
       ts = .timestamp || .ts || now()
       host = to_string(.host || .hostname || "") ?? ""
@@ -107,16 +113,14 @@ sinks:
       strategy: basic
       user: "${CLICKHOUSE_USER}"
       password: "${CLICKHOUSE_PASSWORD}"
-    # Fat, infrequent inserts keep ClickHouse part-count low (avoid too-many-parts).
     batch:
       max_events: 500000
       max_bytes: 134217728
       timeout_secs: 10
-    # Ride out a ClickHouse outage without dropping logs: on-disk buffer on the
-    # aggregator PVC; block upstream (back-pressure to the edge) when full.
+    # Small local buffer — JetStream is the real outage buffer now.
     buffer:
       type: disk
-      max_size: 8589934592
+      max_size: 2147483648
       when_full: block
     healthcheck:
       enabled: true

apps/base/logging/vector/archiver.yaml

@@ -0,0 +1,62 @@
+---
+# Vector ARCHIVER tier — long-term raw-log backup to S3 (Ceph RGW). Independent
+# durable JetStream consumer (`archiver`) so its offsets/lag are fully isolated
+# from the ClickHouse transform path (archive lag can never stall ingest — true
+# fan-out). Writes RAW, pre-transform events (as they sit in JetStream) as
+# gzipped NDJSON, partitioned by subject + date. This is the long-horizon replay
+# source beyond JetStream's 72h retention window.
+data_dir: /vector-data-dir
+
+api:
+  enabled: true
+  address: 0.0.0.0:8686
+
+sources:
+  js_archive:
+    type: nats
+    url: nats://nats.logging.svc.cluster.local:4222
+    connection_name: vector-archiver
+    subject: "logs.>"
+    jetstream:
+      stream: LOGS
+      consumer: archiver
+    auth:
+      strategy: user_password
+      user_password:
+        user: log-consumer
+        password: ${NATS_CONSUMER_PASSWORD}
+    decoding:
+      codec: json
+
+sinks:
+  s3:
+    type: aws_s3
+    inputs:
+      - js_archive
+    bucket: logs-archive
+    endpoint: https://s3.ceph.unkin.net
+    region: us-east-1
+    force_path_style: true
+    tls:
+      ca_file: /etc/vault-ca/ca.crt
+    # AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY come from the logs-archive-s3
+    # Secret (cephrgw-operator) via envFrom on the deployment.
+    key_prefix: "raw/{{ subject }}/%Y/%m/%d/"
+    compression: gzip
+    encoding:
+      codec: json
+    framing:
+      method: newline_delimited
+    filename_time_format: "%Y%m%dT%H%M%SZ"
+    filename_append_uuid: true
+    batch:
+      max_bytes: 134217728
+      timeout_secs: 300
+    buffer:
+      type: memory
+      max_events: 5000
+      when_full: block
+    # Disabled so slow BucketAccess credential propagation doesn't crash-loop
+    # the pod; RGW reachability is proven by the operator's own health.
+    healthcheck:
+      enabled: false

apps/base/logging/vector/vm-ingest-tests.yaml

@@ -0,0 +1,29 @@
+---
+# `vector test` unit tests for the VM-ingest routing transform.
+tests:
+  - name: host_token_is_subject_safe
+    inputs:
+      - insert_at: tag
+        type: log
+        log_fields:
+          host: "db1.syd1.example.net"
+          message: "sshd accepted"
+    outputs:
+      - extract_from: tag
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.host_token, "db1_syd1_example_net")
+
+  - name: missing_host_defaults_to_unknown
+    inputs:
+      - insert_at: tag
+        type: log
+        log_fields:
+          message: "no host field"
+    outputs:
+      - extract_from: tag
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.host_token, "unknown")

apps/base/logging/vector/vm-ingest.yaml

@@ -0,0 +1,50 @@
+---
+# Vector VM-INGEST tier — the VM front door. Thin: accepts NDJSON over HTTPS
+# (behind the logs-ingest Gateway) from puppet-managed VMs, attaches only a
+# routing token, and publishes into JetStream (subject logs.vm.<host>). No
+# parsing here — shaping happens in the transform tier after JetStream, so VM
+# logs get the same durability/replay/fan-out as k8s logs.
+data_dir: /vector-data-dir
+
+api:
+  enabled: true
+  address: 0.0.0.0:8686
+
+sources:
+  vm_http:
+    type: http_server
+    address: 0.0.0.0:8080
+    path: /
+    method: POST
+    decoding:
+      codec: json
+    framing:
+      method: newline_delimited
+
+transforms:
+  # Routing metadata only: derive a NATS-subject-safe host token.
+  tag:
+    type: remap
+    inputs:
+      - vm_http
+    source: |
+      host = to_string(.host || .hostname || "unknown") ?? "unknown"
+      .host_token = replace(host, r'[^a-zA-Z0-9_-]', "_")
+
+sinks:
+  to_jetstream:
+    type: nats
+    inputs:
+      - tag
+    url: nats://nats.logging.svc.cluster.local:4222
+    connection_name: vector-vm-ingest
+    subject: "logs.vm.{{ host_token }}"
+    jetstream:
+      enabled: true
+    auth:
+      strategy: user_password
+      user_password:
+        user: log-producer
+        password: ${NATS_PRODUCER_PASSWORD}
+    encoding:
+      codec: json