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

---
# Declarative JetStream provisioning: the LOGS stream + durable consumers.
# ArgoCD PostSync hook, idempotent create-or-UPDATE, re-runs each sync.
#
# Stream LOGS: file storage, 3 replicas, retention=limits (NOT workqueue) so the
# transform tier AND the archiver each independently see every message — reading
# never deletes; only max-age/max-bytes do. S2 compression is on (logs compress
# well). Replay window = max-age (3d default). Beyond that, the S3 archive is the
# ONLY long-term source — everything else is gone after 3 days (accepted design).
#
# TUNABLE LIMITS LIVE IN A CONFIGMAP (nats-stream-limits): max_age, max_bytes,
# dupe_window. Change the ConfigMap and re-sync — this Job re-runs and applies
# the new limits via `nats stream edit` (no manual surgery). The ConfigMap is
# generated with a content-hash suffix (kustomize), so editing it changes both
# the ConfigMap name AND this Job's env reference → the PostSync hook Job's spec
# changes and Argo re-runs it (belt-and-suspenders on top of hooks running each
# sync; hook-delete-policy=BeforeHookCreation recreates it every time).
#
# 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 max-age, 3d): 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: artifactapi.k8s.syd1.au.unkin.net/dockerhub/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
            # Tunable stream limits — sourced from the ConfigMap.
            - name: MAX_AGE
              valueFrom:
                configMapKeyRef:
                  name: nats-stream-limits
                  key: max_age
            - name: MAX_BYTES
              valueFrom:
                configMapKeyRef:
                  name: nats-stream-limits
                  key: max_bytes
            - name: DUPE_WINDOW
              valueFrom:
                configMapKeyRef:
                  name: nats-stream-limits
                  key: dupe_window
            # 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 (max_age=$MAX_AGE max_bytes=$MAX_BYTES dupe=$DUPE_WINDOW) ..."
              # Create if absent; otherwise converge the mutable limits from the
              # ConfigMap. (storage/retention/replicas are immutable, set only on
              # create.)
              nats stream add LOGS \
                --subjects='logs.>' --storage=file --replicas=3 \
                --retention=limits --discard=old --compression=s2 \
                --max-age="$MAX_AGE" --max-bytes="$MAX_BYTES" \
                --max-msgs=-1 --max-msgs-per-subject=-1 --max-msg-size=-1 \
                --max-consumers=-1 --dupe-window="$DUPE_WINDOW" --defaults 2>/dev/null \
              && echo "  created" \
              || nats stream edit -f LOGS \
                --subjects='logs.>' --discard=old --compression=s2 \
                --max-age="$MAX_AGE" --max-bytes="$MAX_BYTES" \
                --max-msgs=-1 --max-msgs-per-subject=-1 --max-msg-size=-1 \
                --max-consumers=-1 --dupe-window="$DUPE_WINDOW"

              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: {}