unkinben
c39af2f9c3
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
64 lines
1.9 KiB
YAML
64 lines
1.9 KiB
YAML
---
|
|
# `vector test` unit tests for the aggregator transforms. Merged with
|
|
# 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
|
|
type: log
|
|
log_fields:
|
|
message: "hello from pod"
|
|
stream: "stdout"
|
|
timestamp: "2026-07-27T00:00:00Z"
|
|
kubernetes.pod_name: "web-abc"
|
|
kubernetes.pod_namespace: "shop"
|
|
kubernetes.container_name: "web"
|
|
kubernetes.pod_node_name: "node-1"
|
|
outputs:
|
|
- extract_from: k8s_shape
|
|
conditions:
|
|
- type: vrl
|
|
source: |
|
|
assert_eq!(.source, "k8s")
|
|
assert_eq!(.namespace, "shop")
|
|
assert_eq!(.pod, "web-abc")
|
|
assert_eq!(.container, "web")
|
|
assert_eq!(.host, "node-1")
|
|
assert_eq!(.stream, "stdout")
|
|
assert_eq!(.message, "hello from pod")
|
|
|
|
- name: vm_log_is_normalised
|
|
inputs:
|
|
- insert_at: vm_shape
|
|
type: log
|
|
log_fields:
|
|
message: "sshd started"
|
|
host: "vm-db-1"
|
|
severity: "info"
|
|
role: "database"
|
|
outputs:
|
|
- extract_from: vm_shape
|
|
conditions:
|
|
- type: vrl
|
|
source: |
|
|
assert_eq!(.source, "vm")
|
|
assert_eq!(.host, "vm-db-1")
|
|
assert_eq!(.severity, "info")
|
|
assert_eq!(.message, "sshd started")
|
|
assert_eq!(.labels.role, "database")
|