diff --git a/.woodpecker/vector-test.yaml b/.woodpecker/vector-test.yaml index 9cea466..8949ce3 100644 --- a/.woodpecker/vector-test.yaml +++ b/.woodpecker/vector-test.yaml @@ -5,11 +5,18 @@ steps: - name: vector-test image: timberio/vector:0.57.0-debian commands: - # Dummy creds + a writable data_dir so the full topology (incl. the - # ClickHouse sink disk buffer) builds; tests only exercise the transforms. + # Dummy creds + writable dirs so the full topologies build; the unit tests + # only exercise the transforms (sources are not started). - export CLICKHOUSE_USER=ci CLICKHOUSE_PASSWORD=ci - - mkdir -p /vector-data-dir + - export NATS_PRODUCER_PASSWORD=ci NATS_CONSUMER_PASSWORD=ci + - mkdir -p /vector-data-dir /etc/vault-ca + - cp /etc/ssl/certs/ca-certificates.crt /etc/vault-ca/ca.crt + # Transform tier + VM ingest: unit-tested transforms. - vector test apps/base/logging/vector/aggregator.yaml apps/base/logging/vector/aggregator-tests.yaml + - vector test apps/base/logging/vector/vm-ingest.yaml apps/base/logging/vector/vm-ingest-tests.yaml + # Agent + archiver have no transforms to unit-test; validate they build. + - vector validate --no-environment apps/base/logging/vector/agent.yaml + - vector validate --no-environment apps/base/logging/vector/archiver.yaml backend_options: kubernetes: serviceAccountName: default diff --git a/apps/base/logging/cephrgw.yaml b/apps/base/logging/cephrgw.yaml new file mode 100644 index 0000000..ee1bd8e --- /dev/null +++ b/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 diff --git a/apps/base/logging/httproute.yaml b/apps/base/logging/httproute.yaml index ba8f241..2400863 100644 --- a/apps/base/logging/httproute.yaml +++ b/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: diff --git a/apps/base/logging/kustomization.yaml b/apps/base/logging/kustomization.yaml index 934f466..cf94855 100644 --- a/apps/base/logging/kustomization.yaml +++ b/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 diff --git a/apps/base/logging/nats-bootstrap-job.yaml b/apps/base/logging/nats-bootstrap-job.yaml new file mode 100644 index 0000000..72e5ad9 --- /dev/null +++ b/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: {} diff --git a/apps/base/logging/vaultstaticsecret.yaml b/apps/base/logging/vaultstaticsecret.yaml index d37129c..45b56ab 100644 --- a/apps/base/logging/vaultstaticsecret.yaml +++ b/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 diff --git a/apps/base/logging/vector/agent.yaml b/apps/base/logging/vector/agent.yaml new file mode 100644 index 0000000..efbc35f --- /dev/null +++ b/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 diff --git a/apps/base/logging/vector/aggregator-tests.yaml b/apps/base/logging/vector/aggregator-tests.yaml index 94a8c60..8a194fa 100644 --- a/apps/base/logging/vector/aggregator-tests.yaml +++ b/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 diff --git a/apps/base/logging/vector/aggregator.yaml b/apps/base/logging/vector/aggregator.yaml index 78bd816..ad35505 100644 --- a/apps/base/logging/vector/aggregator.yaml +++ b/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 diff --git a/apps/base/logging/vector/archiver.yaml b/apps/base/logging/vector/archiver.yaml new file mode 100644 index 0000000..c7347b2 --- /dev/null +++ b/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 diff --git a/apps/base/logging/vector/vm-ingest-tests.yaml b/apps/base/logging/vector/vm-ingest-tests.yaml new file mode 100644 index 0000000..fed2928 --- /dev/null +++ b/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") diff --git a/apps/base/logging/vector/vm-ingest.yaml b/apps/base/logging/vector/vm-ingest.yaml new file mode 100644 index 0000000..8e2b069 --- /dev/null +++ b/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.). 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 diff --git a/apps/overlays/au-syd1/logging/kustomization.yaml b/apps/overlays/au-syd1/logging/kustomization.yaml index 0ae2aa6..a81c658 100644 --- a/apps/overlays/au-syd1/logging/kustomization.yaml +++ b/apps/overlays/au-syd1/logging/kustomization.yaml @@ -8,18 +8,38 @@ resources: - ../../../base/logging helmCharts: - # Node-level agent: tails every pod's logs and forwards to the aggregator. + # Dedicated JetStream NATS cluster — the durable log bus. + - name: nats + repo: https://nats-io.github.io/k8s/helm/charts + version: "2.14.2" + releaseName: nats + namespace: logging + valuesFile: values-nats.yaml + # Edge agent (DaemonSet): tails every pod's logs, publishes to JetStream. - name: vector repo: https://helm.vector.dev version: "0.57.0" releaseName: vector-agent namespace: logging valuesFile: values-vector-agent.yaml - # Aggregator: receives from agents (vector proto) and from VMs (HTTP), - # shapes events, and writes to ClickHouse. + # VM ingest (Deployment): HTTP NDJSON front door -> JetStream. + - name: vector + repo: https://helm.vector.dev + version: "0.57.0" + releaseName: vector-vm-ingest + namespace: logging + valuesFile: values-vector-vm-ingest.yaml + # Transform tier (StatefulSet): JetStream consumer -> shape -> ClickHouse. - name: vector repo: https://helm.vector.dev version: "0.57.0" releaseName: vector-aggregator namespace: logging valuesFile: values-vector-aggregator.yaml + # Archiver (Deployment): independent JetStream consumer -> raw logs to S3. + - name: vector + repo: https://helm.vector.dev + version: "0.57.0" + releaseName: vector-archiver + namespace: logging + valuesFile: values-vector-archiver.yaml diff --git a/apps/overlays/au-syd1/logging/values-nats.yaml b/apps/overlays/au-syd1/logging/values-nats.yaml new file mode 100644 index 0000000..f6f14e9 --- /dev/null +++ b/apps/overlays/au-syd1/logging/values-nats.yaml @@ -0,0 +1,83 @@ +# Dedicated JetStream-enabled NATS cluster for the log bus. Deliberately NOT +# shared with app messaging (streamstack et al. run their own NATS in their own +# repo) — a separate cluster isolates logging blast-radius from app messaging +# and lets us size retention/storage purely for the log outage-buffer + replay +# use-case. +fullnameOverride: nats + +config: + cluster: + enabled: true + replicas: 3 + jetstream: + enabled: true + fileStore: + pvc: + size: 50Gi + storageClassName: cephrbd-fast-delete + # Per-user auth with publish/subscribe separation. Passwords are injected as + # env vars from the Vault-synced nats-auth Secret (NATS expands $VAR in config). + merge: + authorization: + users: + # Bootstrap Job (stream/consumer management) — full JetStream API. + - user: log-admin + password: $NATS_ADMIN_PASSWORD + # Edge publishers (k8s DaemonSet + VM ingest) — publish only. + - user: log-producer + password: $NATS_PRODUCER_PASSWORD + permissions: + publish: + allow: + - "logs.>" + subscribe: + allow: + - "_INBOX.>" + # Consumers (transform tier + archiver) — pull + ack only, no publish + # to log subjects. + - user: log-consumer + password: $NATS_CONSUMER_PASSWORD + permissions: + publish: + allow: + - "$JS.API.CONSUMER.>" + - "$JS.API.STREAM.INFO.>" + - "$JS.ACK.LOGS.>" + subscribe: + allow: + - "_INBOX.>" + +container: + env: + NATS_ADMIN_PASSWORD: + valueFrom: + secretKeyRef: + name: nats-auth + key: admin_password + NATS_PRODUCER_PASSWORD: + valueFrom: + secretKeyRef: + name: nats-auth + key: producer_password + NATS_CONSUMER_PASSWORD: + valueFrom: + secretKeyRef: + name: nats-auth + key: consumer_password + resources: + requests: + cpu: 250m + memory: 512Mi + limits: + cpu: "2" + memory: 4Gi + +# Roll the StatefulSet when nats-auth changes. +podTemplate: + merge: + metadata: + annotations: + reloader.stakater.com/auto: "true" + +natsBox: + enabled: false diff --git a/apps/overlays/au-syd1/logging/values-vector-agent.yaml b/apps/overlays/au-syd1/logging/values-vector-agent.yaml index 3e8e064..e9551e7 100644 --- a/apps/overlays/au-syd1/logging/values-vector-agent.yaml +++ b/apps/overlays/au-syd1/logging/values-vector-agent.yaml @@ -1,7 +1,7 @@ -# Vector Agent (DaemonSet) — captures ALL pod logs on every node (including -# control-plane, via the blanket toleration) and forwards them to the -# aggregator over the Vector native protocol. No per-app parsing here: shaping -# and the ClickHouse write live on the aggregator. +# Vector EDGE agent (DaemonSet) — thin publisher. Tails every node's pod logs +# (incl. control-plane via the blanket toleration) and publishes them into +# JetStream over the Vector NATS sink. No parsing; only a routing subject token +# is attached. Shaping happens in the transform tier after JetStream. role: Agent fullnameOverride: vector-agent @@ -13,10 +13,16 @@ serviceAccount: podLabels: vector.dev/exclude: "true" -# Run on every node so no host's pod logs are missed. tolerations: - operator: Exists +env: + - name: NATS_PRODUCER_PASSWORD + valueFrom: + secretKeyRef: + name: nats-auth + key: producer_password + resources: requests: cpu: 100m @@ -25,20 +31,14 @@ resources: cpu: "1" memory: 1Gi -# Disable the chart's auto-generated service for the agent (it ships, not serves). service: enabled: false -customConfig: - data_dir: /vector-data-dir - api: - enabled: false - sources: - kubernetes_logs: - type: kubernetes_logs - sinks: - to_aggregator: - type: vector - inputs: - - kubernetes_logs - address: vector-aggregator.logging.svc.cluster.local:6000 +# Pipeline is the single source of truth in apps/base/logging/vector/agent.yaml, +# mounted via existingConfigMaps (avoids the chart's customConfig Helm-tpl pass). +dataDir: /vector-data-dir +existingConfigMaps: + - vector-agent-config + +workloadResourceAnnotations: + reloader.stakater.com/auto: "true" diff --git a/apps/overlays/au-syd1/logging/values-vector-aggregator.yaml b/apps/overlays/au-syd1/logging/values-vector-aggregator.yaml index 681d21f..6f6667b 100644 --- a/apps/overlays/au-syd1/logging/values-vector-aggregator.yaml +++ b/apps/overlays/au-syd1/logging/values-vector-aggregator.yaml @@ -1,38 +1,33 @@ -# Vector AGGREGATOR (StatefulSet) — the "brain" tier: sole ClickHouse writer, -# owns all transforms, holds the only ClickHouse credentials, HA replicas with -# on-disk buffers to ride out a ClickHouse outage. +# Vector TRANSFORM tier (StatefulSet) — the "brain": sole ClickHouse writer, +# owns all transforms, holds the only ClickHouse + NATS-consumer credentials. +# It is a pure JetStream pull consumer (no inbound ports) — durability lives in +# JetStream, so the local disk buffer is small (5Gi PVC / 2GiB buffer). # -# The pipeline itself is NOT inlined here: it lives in -# apps/base/logging/vector/aggregator.yaml (single source of truth, unit-tested -# by `vector test` in CI) and is mounted via existingConfigMaps. This file only -# owns the k8s deployment shape (ports, creds, storage, HA). +# Pipeline is the single source of truth in apps/base/logging/vector/ +# aggregator.yaml (unit-tested by `vector test` in CI), mounted via +# existingConfigMaps. role: Aggregator fullnameOverride: vector-aggregator replicas: 2 -# Reload the StatefulSet when the aggregator ConfigMap or the creds Secret change. workloadResourceAnnotations: reloader.stakater.com/auto: "true" podLabels: vector.dev/exclude: "true" -# Mount the aggregator pipeline ConfigMap (rendered from base by kustomize). -# dataDir must be set when using existingConfigMaps; it is also where the -# clickhouse sink's disk buffer lives (backed by the PVC below). dataDir: /vector-data-dir existingConfigMaps: - vector-aggregator-config -# On-disk buffer storage so a ClickHouse outage does not drop logs. persistence: enabled: true storageClassName: cephrbd-fast-delete - size: 20Gi + size: 5Gi accessModes: - ReadWriteOnce -# ClickHouse basic-auth creds — the ONLY place these are consumed. +# The ONLY place ClickHouse + NATS-consumer creds are consumed. env: - name: CLICKHOUSE_USER valueFrom: @@ -44,14 +39,14 @@ env: secretKeyRef: name: clickhouse-credentials key: password + - name: NATS_CONSUMER_PASSWORD + valueFrom: + secretKeyRef: + name: nats-auth + key: consumer_password +# Pure consumer: expose only the Vector API for debugging. containerPorts: - - name: vector - containerPort: 6000 - protocol: TCP - - name: http-ingest - containerPort: 8080 - protocol: TCP - name: api containerPort: 8686 protocol: TCP @@ -60,14 +55,6 @@ service: enabled: true type: ClusterIP ports: - - name: vector - port: 6000 - targetPort: 6000 - protocol: TCP - - name: http-ingest - port: 8080 - targetPort: 8080 - protocol: TCP - name: api port: 8686 targetPort: 8686 diff --git a/apps/overlays/au-syd1/logging/values-vector-archiver.yaml b/apps/overlays/au-syd1/logging/values-vector-archiver.yaml new file mode 100644 index 0000000..c7c13c8 --- /dev/null +++ b/apps/overlays/au-syd1/logging/values-vector-archiver.yaml @@ -0,0 +1,52 @@ +# Vector ARCHIVER tier (Deployment) — independent JetStream consumer writing raw +# logs to S3 (Ceph RGW). Isolated from the ClickHouse path (own durable +# consumer). Pipeline: apps/base/logging/vector/archiver.yaml. +role: Stateless-Aggregator +fullnameOverride: vector-archiver +replicas: 1 + +workloadResourceAnnotations: + reloader.stakater.com/auto: "true" + +podLabels: + vector.dev/exclude: "true" + +dataDir: /vector-data-dir +existingConfigMaps: + - vector-archiver-config + +env: + - name: NATS_CONSUMER_PASSWORD + valueFrom: + secretKeyRef: + name: nats-auth + key: consumer_password + +# S3 creds (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY) from the cephrgw-operator +# BucketAccess Secret. +envFrom: + - secretRef: + name: logs-archive-s3 + +# Trust the internal unkin.net Vault-PKI CA to verify s3.ceph.unkin.net. +# vault-ca-cert is reflected into every namespace from the certificates ns. +extraVolumes: + - name: vault-ca-cert + secret: + secretName: vault-ca-cert +extraVolumeMounts: + - name: vault-ca-cert + mountPath: /etc/vault-ca/ca.crt + subPath: ca.crt + readOnly: true + +service: + enabled: false + +resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "1" + memory: 1Gi diff --git a/apps/overlays/au-syd1/logging/values-vector-vm-ingest.yaml b/apps/overlays/au-syd1/logging/values-vector-vm-ingest.yaml new file mode 100644 index 0000000..39b5df5 --- /dev/null +++ b/apps/overlays/au-syd1/logging/values-vector-vm-ingest.yaml @@ -0,0 +1,52 @@ +# Vector VM-INGEST tier (Deployment) — VM front door. HTTP NDJSON in (behind the +# logs-ingest Gateway), publishes into JetStream. Stateless publisher. +# Pipeline: apps/base/logging/vector/vm-ingest.yaml (unit-tested in CI). +role: Stateless-Aggregator +fullnameOverride: vector-vm-ingest +replicas: 2 + +workloadResourceAnnotations: + reloader.stakater.com/auto: "true" + +podLabels: + vector.dev/exclude: "true" + +dataDir: /vector-data-dir +existingConfigMaps: + - vector-vm-ingest-config + +env: + - name: NATS_PRODUCER_PASSWORD + valueFrom: + secretKeyRef: + name: nats-auth + key: producer_password + +containerPorts: + - name: http-ingest + containerPort: 8080 + protocol: TCP + - name: api + containerPort: 8686 + protocol: TCP + +service: + enabled: true + type: ClusterIP + ports: + - name: http-ingest + port: 8080 + targetPort: 8080 + protocol: TCP + - name: api + port: 8686 + targetPort: 8686 + protocol: TCP + +resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "1" + memory: 1Gi diff --git a/schemas/ceph.unkin.net/bucket_v1alpha1.json b/schemas/ceph.unkin.net/bucket_v1alpha1.json new file mode 100644 index 0000000..37cdf6f --- /dev/null +++ b/schemas/ceph.unkin.net/bucket_v1alpha1.json @@ -0,0 +1,176 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "apiVersion": { + "type": "string" + }, + "kind": { + "type": "string" + }, + "metadata": { + "type": "object" + }, + "spec": { + "properties": { + "bucketName": { + "type": "string" + }, + "managePolicy": { + "default": true, + "type": "boolean" + }, + "objectLock": { + "properties": { + "days": { + "format": "int32", + "type": "integer" + }, + "enabled": { + "type": "boolean" + }, + "mode": { + "enum": [ + "GOVERNANCE", + "COMPLIANCE" + ], + "type": "string" + }, + "years": { + "format": "int32", + "type": "integer" + } + }, + "required": [ + "enabled" + ], + "type": "object" + }, + "ownerRef": { + "type": "string" + }, + "placementTarget": { + "type": "string" + }, + "purgeOnDelete": { + "type": "boolean" + }, + "quota": { + "properties": { + "enabled": { + "default": true, + "type": "boolean" + }, + "maxObjects": { + "format": "int64", + "type": "integer" + }, + "maxSizeBytes": { + "format": "int64", + "type": "integer" + } + }, + "type": "object" + }, + "retainOnDelete": { + "type": "boolean" + }, + "tags": { + "additionalProperties": { + "type": "string" + }, + "type": "object" + }, + "versioning": { + "type": "boolean" + }, + "zonegroup": { + "type": "string" + } + }, + "required": [ + "ownerRef" + ], + "type": "object" + }, + "status": { + "properties": { + "adopted": { + "type": "boolean" + }, + "bucketID": { + "type": "string" + }, + "bucketName": { + "type": "string" + }, + "conditions": { + "items": { + "properties": { + "lastTransitionTime": { + "format": "date-time", + "type": "string" + }, + "message": { + "maxLength": 32768, + "type": "string" + }, + "observedGeneration": { + "format": "int64", + "minimum": 0, + "type": "integer" + }, + "reason": { + "maxLength": 1024, + "minLength": 1, + "pattern": "^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$", + "type": "string" + }, + "status": { + "enum": [ + "True", + "False", + "Unknown" + ], + "type": "string" + }, + "type": { + "maxLength": 316, + "pattern": "^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$", + "type": "string" + } + }, + "required": [ + "lastTransitionTime", + "message", + "reason", + "status", + "type" + ], + "type": "object" + }, + "type": "array", + "x-kubernetes-list-map-keys": [ + "type" + ], + "x-kubernetes-list-type": "map" + }, + "observedGeneration": { + "format": "int64", + "type": "integer" + }, + "owner": { + "type": "string" + }, + "phase": { + "type": "string" + }, + "policyPrincipals": { + "format": "int32", + "type": "integer" + } + }, + "type": "object" + } + }, + "type": "object" +} diff --git a/schemas/ceph.unkin.net/bucketaccess_v1alpha1.json b/schemas/ceph.unkin.net/bucketaccess_v1alpha1.json new file mode 100644 index 0000000..9e783e2 --- /dev/null +++ b/schemas/ceph.unkin.net/bucketaccess_v1alpha1.json @@ -0,0 +1,191 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "apiVersion": { + "type": "string" + }, + "kind": { + "type": "string" + }, + "metadata": { + "type": "object" + }, + "spec": { + "properties": { + "actions": { + "items": { + "type": "string" + }, + "type": "array" + }, + "bucketRef": { + "type": "string" + }, + "conditions": { + "properties": { + "secureTransportOnly": { + "type": "boolean" + }, + "sourceIPs": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + "type": "object" + }, + "level": { + "enum": [ + "read-only", + "read-write", + "full" + ], + "type": "string" + }, + "paths": { + "items": { + "type": "string" + }, + "type": "array" + }, + "rawStatements": { + "items": { + "properties": { + "actions": { + "items": { + "type": "string" + }, + "type": "array" + }, + "conditions": { + "additionalProperties": { + "additionalProperties": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": "object" + }, + "type": "object" + }, + "effect": { + "default": "Allow", + "enum": [ + "Allow", + "Deny" + ], + "type": "string" + }, + "resources": { + "items": { + "type": "string" + }, + "type": "array" + }, + "sid": { + "type": "string" + } + }, + "required": [ + "actions" + ], + "type": "object" + }, + "type": "array" + }, + "retainOnDelete": { + "type": "boolean" + }, + "secretName": { + "type": "string" + }, + "uid": { + "type": "string" + }, + "userRef": { + "type": "string" + } + }, + "required": [ + "bucketRef", + "level" + ], + "type": "object" + }, + "status": { + "properties": { + "bound": { + "type": "boolean" + }, + "conditions": { + "items": { + "properties": { + "lastTransitionTime": { + "format": "date-time", + "type": "string" + }, + "message": { + "maxLength": 32768, + "type": "string" + }, + "observedGeneration": { + "format": "int64", + "minimum": 0, + "type": "integer" + }, + "reason": { + "maxLength": 1024, + "minLength": 1, + "pattern": "^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$", + "type": "string" + }, + "status": { + "enum": [ + "True", + "False", + "Unknown" + ], + "type": "string" + }, + "type": { + "maxLength": 316, + "pattern": "^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$", + "type": "string" + } + }, + "required": [ + "lastTransitionTime", + "message", + "reason", + "status", + "type" + ], + "type": "object" + }, + "type": "array", + "x-kubernetes-list-map-keys": [ + "type" + ], + "x-kubernetes-list-type": "map" + }, + "observedGeneration": { + "format": "int64", + "type": "integer" + }, + "phase": { + "type": "string" + }, + "secretName": { + "type": "string" + }, + "uid": { + "type": "string" + } + }, + "type": "object" + } + }, + "type": "object" +} diff --git a/schemas/ceph.unkin.net/objectstoreuser_v1alpha1.json b/schemas/ceph.unkin.net/objectstoreuser_v1alpha1.json new file mode 100644 index 0000000..3648da2 --- /dev/null +++ b/schemas/ceph.unkin.net/objectstoreuser_v1alpha1.json @@ -0,0 +1,132 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "apiVersion": { + "type": "string" + }, + "kind": { + "type": "string" + }, + "metadata": { + "type": "object" + }, + "spec": { + "properties": { + "displayName": { + "type": "string" + }, + "email": { + "type": "string" + }, + "maxBuckets": { + "default": 1000, + "format": "int32", + "type": "integer" + }, + "quota": { + "properties": { + "enabled": { + "default": true, + "type": "boolean" + }, + "maxObjects": { + "format": "int64", + "type": "integer" + }, + "maxSizeBytes": { + "format": "int64", + "type": "integer" + } + }, + "type": "object" + }, + "retainOnDelete": { + "type": "boolean" + }, + "secretName": { + "type": "string" + }, + "suspended": { + "type": "boolean" + }, + "uid": { + "type": "string" + } + }, + "type": "object" + }, + "status": { + "properties": { + "adopted": { + "type": "boolean" + }, + "conditions": { + "items": { + "properties": { + "lastTransitionTime": { + "format": "date-time", + "type": "string" + }, + "message": { + "maxLength": 32768, + "type": "string" + }, + "observedGeneration": { + "format": "int64", + "minimum": 0, + "type": "integer" + }, + "reason": { + "maxLength": 1024, + "minLength": 1, + "pattern": "^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$", + "type": "string" + }, + "status": { + "enum": [ + "True", + "False", + "Unknown" + ], + "type": "string" + }, + "type": { + "maxLength": 316, + "pattern": "^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$", + "type": "string" + } + }, + "required": [ + "lastTransitionTime", + "message", + "reason", + "status", + "type" + ], + "type": "object" + }, + "type": "array", + "x-kubernetes-list-map-keys": [ + "type" + ], + "x-kubernetes-list-type": "map" + }, + "observedGeneration": { + "format": "int64", + "type": "integer" + }, + "phase": { + "type": "string" + }, + "secretName": { + "type": "string" + }, + "uid": { + "type": "string" + } + }, + "type": "object" + } + }, + "type": "object" +}