From 084e4549547ddcc8057dc4a44eec8711834ede81 Mon Sep 17 00:00:00 2001 From: Ben Vincent Date: Fri, 31 Jul 2026 22:17:47 +1000 Subject: [PATCH] Add Tier-1 per-app Vector transform pipelines (structured logs) Why: the logging aggregator wrote every event through the generic catch-all shape. The Tier-1 survey (session 15) picked six high-value log sources that warrant structured parsing into logs.raw columns/fields for real querying. How: - Two-stage routing in aggregator.yaml: app_route peels off the six Tier-1 streams by subject / VM source tag (mutually exclusive, no double-insert), everything else falls through app_route._unmatched to the unchanged generic k8s/vm shape catch-all. - Six parse transforms emit the full logs.raw shape plus structured .fields (Map(String,String), no DDL change): authentik (JSON), traefik (JSON access), vault audit (JSON), nginx access+error (regex), haproxy httplog (regex), glauth (JSON). - Flip traefik-system access logs to format: json (companion change) so traefik_parse has structured input. - 15 new vector test cases (routing + field extraction) in aggregator-tests.yaml. VM apps (vault-file/nginx/haproxy/glauth) light up when the puppet-side vector rollout ships logs.vm.* with .file / .SYSLOG_IDENTIFIER tags per the documented convention. authentik and traefik are live now. Claude-Session: https://claude.ai/code/session_015ur3i7D2azsMAWTSVABApv --- .../base/logging/vector/aggregator-tests.yaml | 292 ++++++++++++++ apps/base/logging/vector/aggregator.yaml | 364 +++++++++++++++++- .../traefik-system/values-external.yaml | 4 + .../traefik-system/values-internal.yaml | 4 + 4 files changed, 660 insertions(+), 4 deletions(-) diff --git a/apps/base/logging/vector/aggregator-tests.yaml b/apps/base/logging/vector/aggregator-tests.yaml index 8a194fa..4eeb794 100644 --- a/apps/base/logging/vector/aggregator-tests.yaml +++ b/apps/base/logging/vector/aggregator-tests.yaml @@ -61,3 +61,295 @@ tests: assert_eq!(.severity, "info") assert_eq!(.message, "sshd started") assert_eq!(.labels.role, "database") + + # --- catch-all preservation: an unclaimed k8s event still flows app_route -> + # generic route -> k8s_shape (proves the two-stage chain keeps the fallback) --- + - name: unclaimed_k8s_falls_through_to_generic + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.shop.web" + message: "plain app log" + outputs: + - extract_from: route.k8s + conditions: + - type: vrl + source: | + assert_eq!(.message, "plain app log") + + # --- Tier-1: Authentik SSO (k8s, LIVE NOW) --- + - name: authentik_routes_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.authentik.server" + message: "routed" + outputs: + - extract_from: app_route.authentik + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: authentik_parse_extracts_event + inputs: + - insert_at: authentik_parse + type: log + log_fields: + subject: "logs.k8s.authentik.server" + stream: "stdout" + kubernetes.pod_namespace: "authentik" + kubernetes.container_name: "server" + kubernetes.pod_node_name: "node-2" + message: '{"event":"login","action":"login","user":"alice","client_ip":"203.0.113.9","result":"success","level":"info","logger":"authentik.events","timestamp":"2026-07-27T00:00:00Z"}' + outputs: + - extract_from: authentik_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "authentik") + assert_eq!(.container, "server") + assert_eq!(.severity, "info") + assert_eq!(.message, "login") + assert_eq!(.labels.app, "authentik") + assert_eq!(.fields.event, "login") + assert_eq!(.fields.action, "login") + assert_eq!(.fields.user, "alice") + assert_eq!(.fields.client_ip, "203.0.113.9") + assert_eq!(.fields.result, "success") + + # --- Tier-1: Traefik ingress (k8s, JSON access logs) --- + - name: traefik_routes_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.traefik-system.traefik" + message: "routed" + outputs: + - extract_from: app_route.traefik + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: traefik_parse_extracts_access_fields + inputs: + - insert_at: traefik_parse + type: log + log_fields: + subject: "logs.k8s.traefik-system.traefik" + kubernetes.pod_namespace: "traefik-system" + kubernetes.container_name: "traefik" + kubernetes.pod_node_name: "node-3" + message: '{"RouterName":"web@kubernetes","ServiceName":"shop-svc@kubernetes","RequestMethod":"GET","RequestPath":"/api","RequestHost":"shop.example.net","RequestProtocol":"HTTP/1.1","DownstreamStatus":200,"Duration":5000000,"ClientHost":"203.0.113.5","StartUTC":"2026-07-27T00:00:00Z"}' + outputs: + - extract_from: traefik_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "traefik-system") + assert_eq!(.labels.app, "traefik") + assert_eq!(.message, "GET /api 200") + assert_eq!(.fields.route, "web@kubernetes") + assert_eq!(.fields.service, "shop-svc@kubernetes") + assert_eq!(.fields.method, "GET") + assert_eq!(.fields.path, "/api") + assert_eq!(.fields.host, "shop.example.net") + assert_eq!(.fields.status, "200") + assert_eq!(.fields.duration_ms, "5") + assert_eq!(.fields.client_ip, "203.0.113.5") + + # --- Tier-1: Vault/OpenBao file audit (VM, awaiting VM vector) --- + - name: vault_routes_by_file + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.vault1_syd1" + file: "/var/log/vault_audit.log" + message: "routed" + outputs: + - extract_from: app_route.vault + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: vault_parse_extracts_request + inputs: + - insert_at: vault_parse + type: log + log_fields: + subject: "logs.vm.vault1_syd1" + host: "vault1" + file: "/var/log/vault_audit.log" + message: '{"time":"2026-07-27T00:00:00Z","type":"response","auth":{"display_name":"token"},"request":{"operation":"read","path":"secret/data/app","remote_address":"10.0.0.9"},"error":""}' + outputs: + - extract_from: vault_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "vm") + assert_eq!(.host, "vault1") + assert_eq!(.labels.app, "vault") + assert_eq!(.message, "read secret/data/app") + assert_eq!(.fields.type, "response") + assert_eq!(.fields.display_name, "token") + assert_eq!(.fields.operation, "read") + assert_eq!(.fields.path, "secret/data/app") + assert_eq!(.fields.remote_address, "10.0.0.9") + + # --- Tier-1: nginx access (VM, awaiting VM vector) --- + - name: nginx_access_routes_by_file + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.web1_syd1" + file: "/var/log/nginx/shop_access.log" + message: "routed" + outputs: + - extract_from: app_route.nginx_access + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: nginx_access_parse_extracts_combined + inputs: + - insert_at: nginx_access_parse + type: log + log_fields: + subject: "logs.vm.web1_syd1" + host: "web1" + file: "/var/log/nginx/shop_access.log" + message: '192.0.2.10 - - [27/Jul/2026:00:00:00 +0000] "GET /index.html HTTP/1.1" 200 1024 "https://ref.example/" "Mozilla/5.0" 0.012' + outputs: + - extract_from: nginx_access_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "vm") + assert_eq!(.stream, "access") + assert_eq!(.labels.log_type, "access") + assert_eq!(.fields.client_ip, "192.0.2.10") + assert_eq!(.fields.method, "GET") + assert_eq!(.fields.path, "/index.html") + assert_eq!(.fields.status, "200") + assert_eq!(.fields.bytes, "1024") + assert_eq!(.fields.referer, "https://ref.example/") + assert_eq!(.fields.user_agent, "Mozilla/5.0") + assert_eq!(.fields.request_time, "0.012") + + # --- Tier-1: nginx error (VM, awaiting VM vector) --- + - name: nginx_error_routes_by_file + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.web1_syd1" + file: "/var/log/nginx/shop_error.log" + message: "routed" + outputs: + - extract_from: app_route.nginx_error + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: nginx_error_parse_extracts_fields + inputs: + - insert_at: nginx_error_parse + type: log + log_fields: + subject: "logs.vm.web1_syd1" + host: "web1" + file: "/var/log/nginx/shop_error.log" + message: '2026/07/27 00:00:00 [error] 1234#0: *5 open() "/var/www/x" failed (2: No such file or directory), client: 192.0.2.20, server: shop, request: "GET / HTTP/1.1", host: "shop"' + outputs: + - extract_from: nginx_error_parse + conditions: + - type: vrl + source: | + assert_eq!(.stream, "error") + assert_eq!(.severity, "error") + assert_eq!(.labels.log_type, "error") + assert_eq!(.fields.level, "error") + assert_eq!(.fields.pid, "1234") + assert_eq!(.fields.cid, "5") + assert_eq!(.fields.client_ip, "192.0.2.20") + + # --- Tier-1: HAProxy httplog (VM journald, awaiting VM vector) --- + - name: haproxy_routes_by_identifier + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.halb1_syd1" + SYSLOG_IDENTIFIER: "haproxy" + message: "routed" + outputs: + - extract_from: app_route.haproxy + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: haproxy_parse_extracts_timers + inputs: + - insert_at: haproxy_parse + type: log + log_fields: + subject: "logs.vm.halb1_syd1" + host: "halb1" + SYSLOG_IDENTIFIER: "haproxy" + message: '192.0.2.30:54321 [27/Jul/2026:00:00:00.123] fe_http be_app/app1 10/0/1/2/13 200 512 - - ---- 5/4/3/2/0 0/0 "GET /health HTTP/1.1"' + outputs: + - extract_from: haproxy_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "vm") + assert_eq!(.labels.app, "haproxy") + assert_eq!(.fields.client_ip, "192.0.2.30") + assert_eq!(.fields.frontend, "fe_http") + assert_eq!(.fields.backend, "be_app") + assert_eq!(.fields.server, "app1") + assert_eq!(.fields.tq, "10") + assert_eq!(.fields.tw, "0") + assert_eq!(.fields.tc, "1") + assert_eq!(.fields.tr, "2") + assert_eq!(.fields.tt, "13") + assert_eq!(.fields.termination_state, "----") + assert_eq!(.fields.retries, "0") + assert_eq!(.fields.status, "200") + assert_eq!(.fields.bytes, "512") + + # --- Tier-1: glauth LDAP (VM, awaiting VM vector) --- + - name: glauth_routes_by_identifier + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.ldap1_syd1" + SYSLOG_IDENTIFIER: "glauth" + message: "routed" + outputs: + - extract_from: app_route.glauth + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: glauth_parse_extracts_bind + inputs: + - insert_at: glauth_parse + type: log + log_fields: + subject: "logs.vm.ldap1_syd1" + host: "ldap1" + SYSLOG_IDENTIFIER: "glauth" + message: '{"level":"info","msg":"Bind success as user","bindDN":"cn=admin,dc=example,dc=com","src":"192.0.2.40:1234","time":"2026-07-27T00:00:00Z"}' + outputs: + - extract_from: glauth_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "vm") + assert_eq!(.host, "ldap1") + assert_eq!(.severity, "info") + assert_eq!(.labels.app, "glauth") + assert_eq!(.fields.bindDN, "cn=admin,dc=example,dc=com") + assert_eq!(.fields.remote, "192.0.2.40:1234") + assert_eq!(.fields.success, "true") diff --git a/apps/base/logging/vector/aggregator.yaml b/apps/base/logging/vector/aggregator.yaml index f22b849..83e37ed 100644 --- a/apps/base/logging/vector/aggregator.yaml +++ b/apps/base/logging/vector/aggregator.yaml @@ -3,9 +3,25 @@ # 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. +# 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 @@ -39,10 +55,34 @@ sources: codec: json transforms: - route: + # 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. + authentik: 'starts_with(to_string(.subject) ?? "", "logs.k8s.authentik.")' + # 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")' + + # 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.")' @@ -102,12 +142,328 @@ transforms: "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 + } + 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 endpoint: http://clickhouse-logs.logging.svc.cluster.local:8123 database: logs table: raw diff --git a/apps/overlays/au-syd1/traefik-system/values-external.yaml b/apps/overlays/au-syd1/traefik-system/values-external.yaml index 950a35f..ec830e3 100644 --- a/apps/overlays/au-syd1/traefik-system/values-external.yaml +++ b/apps/overlays/au-syd1/traefik-system/values-external.yaml @@ -32,6 +32,10 @@ providers: logs: access: enabled: true + # JSON access logs so the logging aggregator's traefik_parse transform can + # extract route/service/status/duration/host/client_ip structured fields + # (apps/base/logging/vector/aggregator.yaml). + format: json global: checkNewVersion: true diff --git a/apps/overlays/au-syd1/traefik-system/values-internal.yaml b/apps/overlays/au-syd1/traefik-system/values-internal.yaml index c119afc..865146f 100644 --- a/apps/overlays/au-syd1/traefik-system/values-internal.yaml +++ b/apps/overlays/au-syd1/traefik-system/values-internal.yaml @@ -32,6 +32,10 @@ providers: logs: access: enabled: true + # JSON access logs so the logging aggregator's traefik_parse transform can + # extract route/service/status/duration/host/client_ip structured fields + # (apps/base/logging/vector/aggregator.yaml). + format: json global: checkNewVersion: true