---
# 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/<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")'

      # --- 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.<ns>.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<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
      }

  # --- 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<client_ip>[0-9a-fA-F:.]+)#(?P<port>\d+)(?:\s+\([^)]*\))?:\s+(?:view\s+(?P<view>\S+):\s+)?query:\s+(?P<qname>\S+)\s+(?P<qclass>\S+)\s+(?P<qtype>\S+)(?:\s+(?P<flags>\S+))?') ?? {}
      rc = parse_regex(raw, r'status:\s+(?P<rcode>\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<ms>[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<method>\S+) (?P<path>\S+) for (?P<client>\S+), (?P<status>\d{3}) [^ ]+ in (?P<latency>[0-9.]+\w+)') ?? {}
      a = parse_regex(raw, r'^(?P<client_ip>\S+) \S+ (?P<user>\S+) \[[^\]]+\] "(?P<method>\S+) (?P<path>\S+) [^"]*" (?P<status>\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<ts>\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}[,.]\d+)\s+(?P<level>[A-Z]+)\s+\[(?P<thread>[^\]]*)\]\s+\[(?P<logger>[^\]]*)\]\s+(?P<msg>.*)$') ?? {}
      ac = parse_regex(raw, r'^(?P<client_ip>\S+) \S+ \S+ \[[^\]]+\] "(?P<method>\S+) (?P<path>\S+) [^"]*" (?P<status>\d{3})') ?? {}
      nd = parse_regex(raw, r'(?:catalog for|for node)\s+(?P<node>[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<qid>[0-9A-F]{6,}):') ?? {}
      frm = parse_regex(raw, r'from=<(?P<from>[^>]*)>') ?? {}
      rcpt = parse_regex(raw, r'to=<(?P<to>[^>]*)>') ?? {}
      st = parse_regex(raw, r'status=(?P<status>\w+)') ?? {}
      rel = parse_regex(raw, r'relay=(?P<relay>[^,]+)') ?? {}
      dly = parse_regex(raw, r'delay=(?P<delay>[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
