--- # 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. # # Routing model (two stages): # 1. app_route — peels off Tier-1 per-app streams by subject / source tag and # hands each to a dedicated parse transform that emits the full logs.raw # shape plus structured .fields. Conditions are MUTUALLY EXCLUSIVE, so an # event is claimed by at most one app (no double-insert). # 2. route (generic catch-all) — everything app_route did NOT claim # (app_route._unmatched) is split k8s/vm and shaped generically. This is the # fallback for all un-parsed traffic and MUST stay intact. # Add a new per-app pipeline by appending a mutually-exclusive route to # app_route, a parse transform, and its id to the clickhouse sink `inputs`. # # Structured fields go into the logs.raw `fields Map(String,String)` column — no # DDL change is needed (values are stringified; empties are compacted away). # # VM source-tag convention (the puppet-side vector rollout MUST follow it so # these transforms light up): file sources set `.file` (absolute log path); # journald sources set `.SYSLOG_IDENTIFIER` (falls back to `.program`/`.appname`). # # Durability model: JetStream (3d / 130 GiB, S2-compressed) is the SOLE # durability layer and the replay window. This # tier is stateless (no PVC, memory buffer). If ClickHouse is down the sink # blocks (when_full=block); back-pressure stops the source pulling, so unpulled # messages stay in JetStream and are redelivered. NB: Vector's NATS source has # no end-to-end acks (acks on receipt), so a pod killed mid-outage can lose the # in-memory buffer's worth of already-pulled events — accepted for a stateless, # autoscalable tier. data_dir: /vector-data-dir api: enabled: true address: 0.0.0.0:8686 sources: 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 transforms: # Stage 1: peel off Tier-1 per-app streams. Mutually exclusive conditions; # anything unmatched falls through to the generic `route` below. app_route: type: route inputs: - js_in route: # k8s: authentik SSO — structlog JSON on stdout. The `.postgres` container # is the authentik-namespace CNPG cluster; carve it out so it is claimed by # the single `cnpg_pg` route below (keeps app_route mutually exclusive). authentik: 'starts_with(to_string(.subject) ?? "", "logs.k8s.authentik.") && !ends_with(to_string(.subject) ?? "", ".postgres")' # k8s: Traefik ingress — JSON access logs (requires logs.access.format=json, # flipped in the traefik-system overlay values in this same change). traefik: 'starts_with(to_string(.subject) ?? "", "logs.k8s.traefik-system.")' # VM: Vault/OpenBao file audit device (/var/log/vault_audit.log), JSON. vault: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && contains(to_string(.file) ?? "", "vault_audit")' # VM: nginx combined access log (/var/log/nginx/_access.log). nginx_access: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && contains(to_string(.file) ?? "", "nginx") && ends_with(to_string(.file) ?? "", "access.log")' # VM: nginx error log (/var/log/nginx/_error.log). nginx_error: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && contains(to_string(.file) ?? "", "nginx") && ends_with(to_string(.file) ?? "", "error.log")' # VM: HAProxy httplog via journald. haproxy: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && ((to_string(.SYSLOG_IDENTIFIER) ?? "") == "haproxy" || (to_string(.program) ?? "") == "haproxy" || (to_string(.appname) ?? "") == "haproxy")' # VM: glauth LDAP — structuredlog (logrus) JSON. glauth: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && (contains(to_string(.file) ?? "", "glauth") || (to_string(.SYSLOG_IDENTIFIER) ?? "") == "glauth" || (to_string(.program) ?? "") == "glauth" || (to_string(.appname) ?? "") == "glauth")' # --- Tier-2 (stacks on #318) --- # BIND query logs, k8s + VM. k8s: any bind-* namespace (bind-internal DNS # servers, bind-system operator) — query logging enabled via `querylog yes` # in the BindCluster extraOptions in this change. VM: puppet-managed named # (file /var/log/named/*.log or journald `named`) — puppet-side enable is a # required follow-up (profiles/dns/server.pp). bind_query: 'starts_with(to_string(.subject) ?? "", "logs.k8s.bind") || (starts_with(to_string(.subject) ?? "", "logs.vm.") && (contains(to_string(.file) ?? "", "named") || (to_string(.SYSLOG_IDENTIFIER) ?? "") == "named" || (to_string(.program) ?? "") == "named" || (to_string(.appname) ?? "") == "named"))' # k8s: Rancher audit log — JSON, emitted by the `rancher-audit-log` sidecar # (auditLog.enabled level 1, already on in the cattle-system overlay). rancher_audit: 'starts_with(to_string(.subject) ?? "", "logs.k8s.cattle-system.rancher-audit-log")' # k8s: CNPG Postgres — ONE route for ALL clusters. The CNPG main container is # always named `postgres`, so logs.k8s..postgres uniquely identifies every # cluster across all namespaces (authentik/litellm/artifactapi/woodpecker/ # puppet/paperclip/grafana/netbox/gitea/encapi). Mutually exclusive because the # app routes above/below carve out `.postgres`. cnpg_pg: 'starts_with(to_string(.subject) ?? "", "logs.k8s.") && ends_with(to_string(.subject) ?? "", ".postgres")' # Gitea router/access logs. k8s: the new k8s gitea (ns gitea) with router + # access logging enabled in the overlay values in this change — carve out # `.postgres` (gitea-namespace CNPG). VM: puppet-managed gitea (file or # journald `gitea`) — puppet-side log-format enable is a follow-up. gitea: '(starts_with(to_string(.subject) ?? "", "logs.k8s.gitea.") && !ends_with(to_string(.subject) ?? "", ".postgres")) || (starts_with(to_string(.subject) ?? "", "logs.vm.") && (contains(to_string(.file) ?? "", "gitea") || (to_string(.SYSLOG_IDENTIFIER) ?? "") == "gitea" || (to_string(.program) ?? "") == "gitea" || (to_string(.appname) ?? "") == "gitea"))' # PuppetServer / PuppetDB. k8s: openvoxserver/openvoxdb stdout (ns puppet) — # carve out `.postgres` (puppet-namespace CNPG). VM file logs (multiline # logback + puppetserver-access.log) are a puppet-side vector concern (the # multiline join must happen at the edge) — follow-up. puppet: 'starts_with(to_string(.subject) ?? "", "logs.k8s.puppet.") && !ends_with(to_string(.subject) ?? "", ".postgres")' # k8s: LiteLLM request logs — JSON once JSON_LOGS=True (flipped in the litellm # env in this change). Carve out `.postgres` (litellm-namespace CNPG). litellm: 'starts_with(to_string(.subject) ?? "", "logs.k8s.litellm.") && !ends_with(to_string(.subject) ?? "", ".postgres")' # VM: Postfix maillog — journald (SYSLOG_IDENTIFIER postfix/*) or file maillog. postfix: 'starts_with(to_string(.subject) ?? "", "logs.vm.") && (starts_with(to_string(.SYSLOG_IDENTIFIER) ?? "", "postfix") || starts_with(to_string(.program) ?? "", "postfix") || starts_with(to_string(.appname) ?? "", "postfix") || contains(to_string(.file) ?? "", "maillog"))' # Stage 2: generic catch-all for everything app_route did not claim. route: type: route inputs: - app_route._unmatched route: k8s: 'starts_with(to_string(.subject) ?? "", "logs.k8s.")' vm: 'starts_with(to_string(.subject) ?? "", "logs.vm.")' k8s_shape: type: remap inputs: - route.k8s source: | ts = .timestamp || now() node = to_string(.kubernetes.pod_node_name || "") ?? "" ns = to_string(.kubernetes.pod_namespace || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" msg = to_string(.message || "") ?? "" lbls = object(.kubernetes.pod_labels) ?? {} . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": ns, "pod": pod, "container": container, "stream": strm, "severity": "", "message": msg, "labels": lbls, "fields": {} } vm_shape: type: remap inputs: - route.vm source: | ts = .timestamp || .ts || now() host = to_string(.host || .hostname || "") ?? "" msg = to_string(.message || .msg || "") ?? "" sev = to_string(.severity || .level || "") ?? "" role = to_string(.role || "") ?? "" lbls = {} if role != "" { lbls = {"role": role} } . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "", "severity": sev, "message": msg, "labels": lbls, "fields": {} } # --- Tier-1 per-app parse transforms (each emits the full logs.raw shape) --- # Authentik SSO (k8s, ns authentik) — structlog JSON on stdout. # LIVE NOW: authentik pods already stream to logs.k8s.authentik.*. authentik_parse: type: remap inputs: - app_route.authentik source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.timestamp || .timestamp || now() user = "" if is_string(ev.user) { user = to_string(ev.user) ?? "" } else if is_object(ev.user) { user = to_string(ev.user.username) ?? "" } fields = compact({ "event": to_string(ev.event) ?? "", "action": to_string(ev.action) ?? "", "user": user, "client_ip": to_string(ev.client_ip) ?? "", "result": to_string(ev.result) ?? "", "logger": to_string(ev.logger) ?? "" }, string: true) sev = to_string(ev.level) ?? "" msg = raw if ev.event != null { msg = to_string(ev.event) ?? raw } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": "authentik", "pod": pod, "container": container, "stream": strm, "severity": sev, "message": msg, "labels": {"app": "authentik"}, "fields": fields } # Traefik ingress (k8s, ns traefik-system) — JSON access logs. Non-access # traefik lines (app logs) simply parse to no access fields and keep .message. # geoip on client_ip is a PREREQUISITE (no enrichment table yet — see PR note). traefik_parse: type: remap inputs: - app_route.traefik source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.StartUTC || ev.time || .timestamp || now() status = "" if ev.DownstreamStatus != null { status = to_string(ev.DownstreamStatus) ?? "" } dur_ns = to_int(ev.Duration) ?? 0 dur_ms = "" if dur_ns > 0 { dur_ms = to_string(dur_ns / 1000000) } method = to_string(ev.RequestMethod) ?? "" path = to_string(ev.RequestPath) ?? "" fields = compact({ "route": to_string(ev.RouterName) ?? "", "service": to_string(ev.ServiceName) ?? "", "method": method, "path": path, "host": to_string(ev.RequestHost) ?? "", "status": status, "duration_ms": dur_ms, "client_ip": to_string(ev.ClientHost) ?? "", "protocol": to_string(ev.RequestProtocol) ?? "" }, string: true) msg = raw if method != "" { msg = method + " " + path + " " + status } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": "traefik-system", "pod": pod, "container": container, "stream": strm, "severity": "", "message": msg, "labels": {"app": "traefik"}, "fields": fields } # Vault/OpenBao file audit device (VM, /var/log/vault_audit.log) — JSON, one # object per request/response. AWAITING VM VECTOR (in-cluster vault is quiet; # lights up when the puppet vector rollout ships logs.vm.* with .file set). vault_parse: type: remap inputs: - app_route.vault source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.time || .timestamp || .ts || now() auth = object(ev.auth) ?? {} req = object(ev.request) ?? {} fields = compact({ "type": to_string(ev.type) ?? "", "display_name": to_string(auth.display_name) ?? "", "operation": to_string(req.operation) ?? "", "path": to_string(req.path) ?? "", "remote_address": to_string(req.remote_address) ?? "", "error": to_string(ev.error) ?? "" }, string: true) op = to_string(req.operation) ?? "" pth = to_string(req.path) ?? "" msg = raw if op != "" || pth != "" { msg = op + " " + pth } . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "", "severity": "", "message": msg, "labels": {"app": "vault"}, "fields": fields } # nginx access log (VM) — combined/CLF + optional trailing request_time. # AWAITING VM VECTOR. geoip on client_ip is a PREREQUISITE (see PR note). nginx_access_parse: type: remap inputs: - app_route.nginx_access source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() m = parse_regex(raw, r'^(?P\S+) \S+ (?P\S+) \[(?P[^\]]+)\] "(?P\S+) (?P\S+) (?P[^"]*)" (?P\d{3}) (?P\d+|-) "(?P[^"]*)" "(?P[^"]*)"(?: (?P[\d.]+))?') ?? {} fields = compact({ "client_ip": to_string(m.client_ip), "method": to_string(m.method), "path": to_string(m.path), "status": to_string(m.status), "bytes": to_string(m.bytes), "referer": to_string(m.referer), "user_agent": to_string(m.user_agent), "request_time": to_string(m.request_time) }, string: true) . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "access", "severity": "", "message": raw, "labels": {"app": "nginx", "log_type": "access"}, "fields": fields } # nginx error log (VM). AWAITING VM VECTOR. nginx_error_parse: type: remap inputs: - app_route.nginx_error source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() m = parse_regex(raw, r'^(?P\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[(?P\w+)\] (?P\d+)#(?P\d+): (?:\*(?P\d+) )?(?P.*)$') ?? {} c = parse_regex(raw, r'client: (?P[0-9a-fA-F:.]+)') ?? {} lvl = to_string(m.level) err = to_string(m.err) fields = compact({ "level": lvl, "pid": to_string(m.pid), "cid": to_string(m.cid), "client_ip": to_string(c.client_ip), "error": err }, string: true) msg = raw if err != "" { msg = err } . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "error", "severity": lvl, "message": msg, "labels": {"app": "nginx", "log_type": "error"}, "fields": fields } # HAProxy httplog (VM, journald). AWAITING VM VECTOR. # httplog: %ci:%cp [%tr] %ft %b/%s %Tq/%Tw/%Tc/%Tr/%Tt %ST %B %CC %CS %tsc # %ac/%fc/%bc/%sc/%rc %sq/%bq {hdrs} "%r" haproxy_parse: type: remap inputs: - app_route.haproxy source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() m = parse_regex(raw, r'(?P\d{1,3}(?:\.\d{1,3}){3}):(?P\d+) \[(?P[^\]]+)\] (?P\S+) (?P[^/ ]+)/(?P\S+) (?P-?\d+)/(?P-?\d+)/(?P-?\d+)/(?P-?\d+)/(?P[+-]?\d+) (?P\d{3}) (?P\d+) \S+ \S+ (?P\S{4}) (?P\d+)/(?P\d+)/(?P\d+)/(?P\d+)/(?P\d+) (?P\d+)/(?P\d+)') ?? {} fields = compact({ "client_ip": to_string(m.client_ip), "frontend": to_string(m.frontend), "backend": to_string(m.backend), "server": to_string(m.server), "tq": to_string(m.tq), "tw": to_string(m.tw), "tc": to_string(m.tc), "tr": to_string(m.tr), "tt": to_string(m.tt), "termination_state": to_string(m.termination_state), "retries": to_string(m.retries), "status": to_string(m.status), "bytes": to_string(m.bytes) }, string: true) . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "", "severity": "", "message": raw, "labels": {"app": "haproxy"}, "fields": fields } # glauth LDAP (VM) — structuredlog (logrus) JSON. AWAITING VM VECTOR. glauth_parse: type: remap inputs: - app_route.glauth source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.time || .timestamp || .ts || now() binddn = to_string(ev.bindDN) ?? "" if binddn == "" { binddn = to_string(ev.binddn) ?? "" } remote = to_string(ev.src) ?? "" if remote == "" { remote = to_string(ev.remoteAddr) ?? "" } lvl = to_string(ev.level) ?? "" gmsg = to_string(ev.msg) ?? "" success = "false" if contains(downcase(gmsg), "success") || (lvl == "info" && contains(downcase(gmsg), "bind")) { success = "true" } fields = compact({ "bindDN": binddn, "remote": remote, "success": success, "level": lvl, "msg": gmsg }, string: true) msg = gmsg if msg == "" { msg = raw } . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "", "severity": lvl, "message": msg, "labels": {"app": "glauth"}, "fields": fields } # --- Tier-2 per-app parse transforms (stacks on #318) --- # BIND query logs (k8s bind-* namespaces + VM named). LIVE on k8s once the # `querylog yes` extraOptions (this change) roll out; VM AWAITS the puppet-side # enable (profiles/dns/server.pp). rcode is NOT present in standard query-log # lines (that needs response logging / dnstap) — extracted only if a # response-style `status:` line is seen. Non-query lines keep .message. bind_query_parse: type: remap inputs: - app_route.bind_query source: | subj = to_string(.subject) ?? "" is_k8s = starts_with(subj, "logs.k8s.") raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() node = "" ns = "" pod = "" container = "" strm = "" hostv = "" src = "vm" if is_k8s { src = "k8s" node = to_string(.kubernetes.pod_node_name || "") ?? "" ns = to_string(.kubernetes.pod_namespace || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" hostv = node } else { hostv = to_string(.host || .hostname || "") ?? "" } m = parse_regex(raw, r'client\s+(?:@\S+\s+)?(?P[0-9a-fA-F:.]+)#(?P\d+)(?:\s+\([^)]*\))?:\s+(?:view\s+(?P\S+):\s+)?query:\s+(?P\S+)\s+(?P\S+)\s+(?P\S+)(?:\s+(?P\S+))?') ?? {} rc = parse_regex(raw, r'status:\s+(?P\w+)') ?? {} fields = compact({ "client_ip": to_string(m.client_ip), "qname": to_string(m.qname), "qtype": to_string(m.qtype), "qclass": to_string(m.qclass), "view": to_string(m.view), "flags": to_string(m.flags), "rcode": to_string(rc.rcode) }, string: true) qn = to_string(m.qname) msg = raw if qn != "" { msg = "query " + qn + " " + to_string(m.qtype) } . = { "timestamp": ts, "host": hostv, "source": src, "namespace": ns, "pod": pod, "container": container, "stream": strm, "severity": "", "message": msg, "labels": {"app": "bind"}, "fields": fields } # Rancher audit log (k8s, cattle-system rancher-audit-log sidecar) — JSON, # auditLog level 1 (already enabled in the overlay). LIVE NOW. rancher_audit_parse: type: remap inputs: - app_route.rancher_audit source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.requestTimestamp || ev.time || .timestamp || now() user = "" if is_object(ev.user) { user = to_string(ev.user.name) ?? "" } else if is_string(ev.user) { user = to_string(ev.user) ?? "" } verb = to_string(ev.method) ?? "" if verb == "" { verb = to_string(ev.verb) ?? "" } uri = to_string(ev.requestURI) ?? "" if uri == "" { uri = to_string(ev.uri) ?? "" } status = "" if ev.responseCode != null { status = to_string(ev.responseCode) ?? "" } if status == "" && is_object(ev.responseStatus) { status = to_string(ev.responseStatus.code) ?? "" } fields = compact({ "user": user, "verb": verb, "uri": uri, "status": status, "auditID": to_string(ev.auditID) ?? "", "remote_addr": to_string(ev.remoteAddr) ?? "" }, string: true) msg = raw if verb != "" || uri != "" { msg = verb + " " + uri + " " + status } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": "cattle-system", "pod": pod, "container": container, "stream": strm, "severity": "", "message": msg, "labels": {"app": "rancher", "log_type": "audit"}, "fields": fields } # CNPG Postgres — ONE transform for ALL clusters (10 namespaces). The instance # manager wraps postgres logs as JSON on stdout; the postgres CSV columns nest # under `.record` (logger == "postgres"). Non-postgres lines (instance-manager # operator logs) keep .message and set no PG fields. LIVE NOW. cnpg_pg_parse: type: remap inputs: - app_route.cnpg_pg source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" ns = to_string(.kubernetes.pod_namespace || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" cluster = to_string(.kubernetes.pod_labels."cnpg.io/cluster" || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = .timestamp || now() rec = object(ev.record) ?? {} logger = to_string(ev.logger) ?? "" sev = "" pgmsg = "" fields = {} if logger == "postgres" { sev = to_string(rec.error_severity) ?? "" pgmsg = to_string(rec.message) ?? "" dm = parse_regex(pgmsg, r'duration:\s+(?P[0-9.]+)\s+ms') ?? {} fields = compact({ "error_severity": sev, "message": pgmsg, "query": to_string(rec.query) ?? "", "duration_ms": to_string(dm.ms), "user": to_string(rec.user_name) ?? "", "database": to_string(rec.database_name) ?? "" }, string: true) } lbls = {"app": "cnpg"} if cluster != "" { lbls = {"app": "cnpg", "cluster": cluster} } msg = raw if pgmsg != "" { msg = pgmsg } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": ns, "pod": pod, "container": container, "stream": strm, "severity": sev, "message": msg, "labels": lbls, "fields": fields } # Gitea router/access logs (k8s gitea + VM gitea). Router "completed" lines give # method/path/status/latency; NCSA access lines give method/path/status/user. # k8s LIVE once the overlay log config (this change) rolls out; VM AWAITS the # puppet-side log-format enable. gitea_parse: type: remap inputs: - app_route.gitea source: | subj = to_string(.subject) ?? "" is_k8s = starts_with(subj, "logs.k8s.") raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() node = "" ns = "" pod = "" container = "" strm = "" hostv = "" src = "vm" if is_k8s { src = "k8s" node = to_string(.kubernetes.pod_node_name || "") ?? "" ns = to_string(.kubernetes.pod_namespace || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" hostv = node } else { hostv = to_string(.host || .hostname || "") ?? "" } r = parse_regex(raw, r'completed (?P\S+) (?P\S+) for (?P\S+), (?P\d{3}) [^ ]+ in (?P[0-9.]+\w+)') ?? {} a = parse_regex(raw, r'^(?P\S+) \S+ (?P\S+) \[[^\]]+\] "(?P\S+) (?P\S+) [^"]*" (?P\d{3})') ?? {} method = to_string(r.method) if method == "" { method = to_string(a.method) } path = to_string(r.path) if path == "" { path = to_string(a.path) } status = to_string(r.status) if status == "" { status = to_string(a.status) } user = to_string(a.user) if user == "-" { user = "" } fields = compact({ "method": method, "path": path, "status": status, "latency": to_string(r.latency), "user": user, "client_ip": to_string(a.client_ip) }, string: true) msg = raw if method != "" { msg = method + " " + path + " " + status } . = { "timestamp": ts, "host": hostv, "source": src, "namespace": ns, "pod": pod, "container": container, "stream": strm, "severity": "", "message": msg, "labels": {"app": "gitea"}, "fields": fields } # PuppetServer / PuppetDB (k8s openvoxserver/openvoxdb stdout, ns puppet). Per # line logback parse (level/logger/message + node) and an access-log line # (method/status/node) where present. VM multiline stacktrace join + # puppetserver-access.log are a puppet-side edge concern (follow-up). LIVE NOW. puppet_parse: type: remap inputs: - app_route.puppet source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" ts = .timestamp || now() lb = parse_regex(raw, r'^(?P\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}[,.]\d+)\s+(?P[A-Z]+)\s+\[(?P[^\]]*)\]\s+\[(?P[^\]]*)\]\s+(?P.*)$') ?? {} ac = parse_regex(raw, r'^(?P\S+) \S+ \S+ \[[^\]]+\] "(?P\S+) (?P\S+) [^"]*" (?P\d{3})') ?? {} nd = parse_regex(raw, r'(?:catalog for|for node)\s+(?P[a-zA-Z0-9._-]+\.[a-zA-Z0-9._-]+)') ?? {} lvl = to_string(lb.level) pmsg = to_string(lb.msg) err = "" if lvl == "ERROR" { err = pmsg } fields = compact({ "level": lvl, "logger": to_string(lb.logger), "node": to_string(nd.node), "method": to_string(ac.method), "status": to_string(ac.status), "path": to_string(ac.path), "client_ip": to_string(ac.client_ip), "error": err }, string: true) msg = raw if pmsg != "" { msg = pmsg } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": "puppet", "pod": pod, "container": container, "stream": strm, "severity": lvl, "message": msg, "labels": {"app": "puppet"}, "fields": fields } # LiteLLM request logs (k8s) — JSON once JSON_LOGS=True (flipped in the litellm # env this change). parse_json -> model/tokens/latency/key/status (best-effort # against litellm's JSON schema); non-JSON lines keep .message. Field keys light # up once the env flip rolls out. litellm_parse: type: remap inputs: - app_route.litellm source: | node = to_string(.kubernetes.pod_node_name || "") ?? "" pod = to_string(.kubernetes.pod_name || "") ?? "" container = to_string(.kubernetes.container_name || "") ?? "" strm = to_string(.stream || "") ?? "" raw = to_string(.message || "") ?? "" ev = object(parse_json(raw) ?? {}) ?? {} ts = ev.timestamp || .timestamp || now() sev = to_string(ev.level) ?? "" lmsg = to_string(ev.message) ?? "" fields = compact({ "model": to_string(ev.model) ?? "", "tokens": to_string(ev.total_tokens) ?? "", "latency": to_string(ev.response_time) ?? "", "key": to_string(ev.api_key) ?? "", "status": to_string(ev.status) ?? "", "user": to_string(ev.user) ?? "" }, string: true) msg = raw if lmsg != "" { msg = lmsg } . = { "timestamp": ts, "host": node, "source": "k8s", "namespace": "litellm", "pod": pod, "container": container, "stream": strm, "severity": sev, "message": msg, "labels": {"app": "litellm"}, "fields": fields } # Postfix maillog (VM) — best-effort PER-LINE parse (qid + from/to/status/relay/ # delay). Full qid-lifecycle correlation is a query-time GROUP BY qid in # ClickHouse, NOT a stateless-aggregator job (stitching the multi-line lifecycle # needs a stateful reduce). AWAITS VM VECTOR. postfix_parse: type: remap inputs: - app_route.postfix source: | host = to_string(.host || .hostname || "") ?? "" raw = to_string(.message || .msg || "") ?? "" ts = .timestamp || .ts || now() prog = to_string(.SYSLOG_IDENTIFIER || .program || .appname || "") ?? "" q = parse_regex(raw, r'^(?P[0-9A-F]{6,}):') ?? {} frm = parse_regex(raw, r'from=<(?P[^>]*)>') ?? {} rcpt = parse_regex(raw, r'to=<(?P[^>]*)>') ?? {} st = parse_regex(raw, r'status=(?P\w+)') ?? {} rel = parse_regex(raw, r'relay=(?P[^,]+)') ?? {} dly = parse_regex(raw, r'delay=(?P[0-9.]+)') ?? {} fields = compact({ "qid": to_string(q.qid), "from": to_string(frm.from), "to": to_string(rcpt.to), "status": to_string(st.status), "relay": to_string(rel.relay), "delay": to_string(dly.delay), "program": prog }, string: true) . = { "timestamp": ts, "host": host, "source": "vm", "namespace": "", "pod": "", "container": "", "stream": "", "severity": "", "message": raw, "labels": {"app": "postfix"}, "fields": fields } sinks: clickhouse: type: clickhouse inputs: - k8s_shape - vm_shape - authentik_parse - traefik_parse - vault_parse - nginx_access_parse - nginx_error_parse - haproxy_parse - glauth_parse - bind_query_parse - rancher_audit_parse - cnpg_pg_parse - gitea_parse - puppet_parse - litellm_parse - postfix_parse endpoint: http://clickhouse-logs.logging.svc.cluster.local:8123 database: logs table: raw skip_unknown_fields: true date_time_best_effort: true auth: strategy: basic user: "${CLICKHOUSE_USER}" password: "${CLICKHOUSE_PASSWORD}" batch: max_events: 500000 max_bytes: 134217728 timeout_secs: 10 # Stateless: in-memory buffer, block on full so back-pressure reaches the # JetStream pull source (which then stops acking). JetStream is durability. buffer: type: memory max_events: 2000 when_full: block healthcheck: enabled: true