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

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")

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:
+# the ONLY ClickHouse writer.
-# insert a transform and append its id to the clickhouse sink `inputs` — no edge
+#
-# or VM rollout required.
+# 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/<vhost>_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/<vhost>_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<client_ip>\S+) \S+ (?P<user>\S+) \[(?P<time_local>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) (?P<protocol>[^"]*)" (?P<status>\d{3}) (?P<bytes>\d+|-) "(?P<referer>[^"]*)" "(?P<user_agent>[^"]*)"(?: (?P<request_time>[\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<time_local>\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) \[(?P<level>\w+)\] (?P<pid>\d+)#(?P<tid>\d+): (?:\*(?P<cid>\d+) )?(?P<err>.*)$') ?? {}
+      c = parse_regex(raw, r'client: (?P<client_ip>[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<client_ip>\d{1,3}(?:\.\d{1,3}){3}):(?P<client_port>\d+) \[(?P<accept_date>[^\]]+)\] (?P<frontend>\S+) (?P<backend>[^/ ]+)/(?P<server>\S+) (?P<tq>-?\d+)/(?P<tw>-?\d+)/(?P<tc>-?\d+)/(?P<tr>-?\d+)/(?P<tt>[+-]?\d+) (?P<status>\d{3}) (?P<bytes>\d+) \S+ \S+ (?P<termination_state>\S{4}) (?P<actconn>\d+)/(?P<feconn>\d+)/(?P<beconn>\d+)/(?P<srvconn>\d+)/(?P<retries>\d+) (?P<srv_queue>\d+)/(?P<backend_queue>\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

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

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