From da4a66046a00d8073a1539391a9085cf61b66e66 Mon Sep 17 00:00:00 2001 From: Ben Vincent Date: Sat, 8 Aug 2026 22:12:46 +1000 Subject: [PATCH] Add Tier-2 per-app Vector transform pipelines (structured logs) (#320) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: extend the Tier-1 survey with 7 more high-value log sources so they parse into logs.raw columns/.fields for real querying instead of the generic catch-all. **Stacks on #318 — merge after it.** How: - 7 mutually-exclusive app_route conditions + parse transforms into the ClickHouse sink: **bind_query** (k8s bind-* + VM named), **rancher_audit** (cattle-system sidecar JSON), **cnpg_pg** (ONE transform for all 10 CNPG clusters via the `.postgres` container), **gitea** (router+access, k8s+VM), **puppet** (openvoxserver/openvoxdb logback + access), **litellm** (JSON request logs), **postfix** (per-line maillog). - Carve `.postgres` out of the Tier-1 authentik route + new puppet/gitea/litellm routes so the single cnpg_pg route claims every CNPG pod without double-insert (keeps app_route mutually exclusive). Catch-all intact. - Companion k8s flips in this PR: litellm `JSON_LOGS=True`; bind `querylog yes` on both bind-internal BindClusters; gitea router+access logging to stdout. Rancher auditLog was already on. - 15 new `vector test` cases (routing + field extraction + authentik-postgres→cnpg exclusivity proof); all 35 green (vector 0.57). Fields go into the existing `fields Map(String,String)` — no DDL change. Puppet-side follow-ups (out of scope for argocd): enable named query logging (profiles/dns/server.pp); ship the VM vector rollout with `.file`/`.SYSLOG_IDENTIFIER` tags for named/gitea/puppetserver(+multiline logback join)/postfix maillog. https://claude.ai/code/session_015ur3i7D2azsMAWTSVABApv --------- Co-authored-by: Ben Vincent Reviewed-on: https://git.unkin.net/unkin/argocd-apps/pulls/320 Co-authored-by: Ben Vincent Co-committed-by: Ben Vincent --- .../bind-internal/authoritative/cluster.yaml | 3 + .../base/bind-internal/resolvers/cluster.yaml | 5 + apps/base/litellm/kustomization.yaml | 3 + .../base/logging/vector/aggregator-tests.yaml | 313 +++++++++++++ apps/base/logging/vector/aggregator.yaml | 417 +++++++++++++++++- apps/overlays/au-syd1/gitea/values.yaml | 9 + 6 files changed, 748 insertions(+), 2 deletions(-) diff --git a/apps/base/bind-internal/authoritative/cluster.yaml b/apps/base/bind-internal/authoritative/cluster.yaml index de2a627..ef884f1 100644 --- a/apps/base/bind-internal/authoritative/cluster.yaml +++ b/apps/base/bind-internal/authoritative/cluster.yaml @@ -18,6 +18,9 @@ spec: # without it every dynamic update is "denied due to allow-query". extraOptions: - "allow-query { localhost; auth-acl-main; 10.42.0.0/16; }" + # Enable query logging for the Tier-2 vector bind_query pipeline (see the + # resolvers cluster for the routing rationale). + - "querylog yes" service: type: LoadBalancer externalTrafficPolicy: Local diff --git a/apps/base/bind-internal/resolvers/cluster.yaml b/apps/base/bind-internal/resolvers/cluster.yaml index 161a7bd..bda2308 100644 --- a/apps/base/bind-internal/resolvers/cluster.yaml +++ b/apps/base/bind-internal/resolvers/cluster.yaml @@ -30,6 +30,11 @@ spec: # (incl. k8s.syd1.au.unkin.net); 18.198.in-addr.arpa covers every reverse zone. extraOptions: - "validate-except { unkin.net; 18.198.in-addr.arpa; consul; }" + # Enable query logging so the Tier-2 vector bind_query pipeline can parse + # client/qname/qtype. Routes to the `queries` category which, with no explicit + # logging{} clause, follows the default category to the named foreground + # stderr channel -> pod stdout -> vector (subject logs.k8s.bind-internal.*). + - "querylog yes" resources: requests: cpu: 20m diff --git a/apps/base/litellm/kustomization.yaml b/apps/base/litellm/kustomization.yaml index 5a582cd..bc8e4b4 100644 --- a/apps/base/litellm/kustomization.yaml +++ b/apps/base/litellm/kustomization.yaml @@ -27,6 +27,9 @@ configMapGenerator: - name: litellm-env literals: - STORE_MODEL_IN_DB=True + # Emit structured JSON logs so the Tier-2 vector litellm pipeline can parse + # model/tokens/latency/key/status (logs.k8s.litellm.*). + - JSON_LOGS=True # Authentik OIDC SSO (generic). Client secret is injected from the # oauth-credentials Secret in the Deployment; endpoints match the other # apps (identity.unkin.net). PROXY_BASE_URL is required for SSO. diff --git a/apps/base/logging/vector/aggregator-tests.yaml b/apps/base/logging/vector/aggregator-tests.yaml index 4eeb794..deb2e9b 100644 --- a/apps/base/logging/vector/aggregator-tests.yaml +++ b/apps/base/logging/vector/aggregator-tests.yaml @@ -353,3 +353,316 @@ tests: assert_eq!(.fields.bindDN, "cn=admin,dc=example,dc=com") assert_eq!(.fields.remote, "192.0.2.40:1234") assert_eq!(.fields.success, "true") + + # ================= Tier-2 (stacks on #318) ================= + + # --- BIND query logs (k8s bind-* + VM named) --- + - name: bind_routes_k8s_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.bind-internal.named" + message: "routed" + outputs: + - extract_from: app_route.bind_query + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: bind_routes_vm_by_identifier + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.dns1_syd1" + SYSLOG_IDENTIFIER: "named" + message: "routed" + outputs: + - extract_from: app_route.bind_query + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: bind_parse_extracts_query + inputs: + - insert_at: bind_query_parse + type: log + log_fields: + subject: "logs.k8s.bind-internal.named" + kubernetes.pod_namespace: "bind-internal" + kubernetes.container_name: "named" + kubernetes.pod_node_name: "node-4" + message: '02-Aug-2026 00:00:00.123 client @0x7f 192.0.2.1#40426 (www.example.com): view internal: query: www.example.com IN A +E(0)K (198.18.200.7)' + outputs: + - extract_from: bind_query_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "bind-internal") + assert_eq!(.labels.app, "bind") + assert_eq!(.message, "query www.example.com A") + assert_eq!(.fields.client_ip, "192.0.2.1") + assert_eq!(.fields.qname, "www.example.com") + assert_eq!(.fields.qclass, "IN") + assert_eq!(.fields.qtype, "A") + assert_eq!(.fields.view, "internal") + + # --- Rancher audit (k8s, cattle-system sidecar) --- + - name: rancher_routes_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.cattle-system.rancher-audit-log" + message: "routed" + outputs: + - extract_from: app_route.rancher_audit + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: rancher_parse_extracts_audit + inputs: + - insert_at: rancher_audit_parse + type: log + log_fields: + subject: "logs.k8s.cattle-system.rancher-audit-log" + kubernetes.pod_namespace: "cattle-system" + kubernetes.container_name: "rancher-audit-log" + kubernetes.pod_node_name: "node-5" + message: '{"auditID":"abc-123","requestURI":"/v3/tokens","user":{"name":"u-alice","group":["admins"]},"method":"GET","remoteAddr":"10.42.0.9:1234","responseCode":200,"requestTimestamp":"2026-08-01T00:00:00Z"}' + outputs: + - extract_from: rancher_audit_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "cattle-system") + assert_eq!(.labels.app, "rancher") + assert_eq!(.labels.log_type, "audit") + assert_eq!(.message, "GET /v3/tokens 200") + assert_eq!(.fields.user, "u-alice") + assert_eq!(.fields.verb, "GET") + assert_eq!(.fields.uri, "/v3/tokens") + assert_eq!(.fields.status, "200") + + # --- CNPG Postgres (ONE transform, all clusters) --- + - name: cnpg_routes_by_postgres_container + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.litellm.postgres" + message: "routed" + outputs: + - extract_from: app_route.cnpg_pg + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + # mutual exclusivity: an app-namespace CNPG pod (authentik) is claimed by + # cnpg_pg, NOT the authentik app route (which now carves out .postgres). + - name: cnpg_authentik_postgres_routes_to_cnpg + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.authentik.postgres" + message: "routed" + outputs: + - extract_from: app_route.cnpg_pg + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: cnpg_parse_extracts_record + inputs: + - insert_at: cnpg_pg_parse + type: log + log_fields: + subject: "logs.k8s.litellm.postgres" + kubernetes.pod_namespace: "litellm" + kubernetes.container_name: "postgres" + kubernetes.pod_node_name: "node-6" + kubernetes.pod_labels."cnpg.io/cluster": "litellm-postgres" + message: '{"level":"info","ts":"2026-08-01T00:00:00Z","logger":"postgres","msg":"record","record":{"user_name":"litellm","database_name":"litellm","error_severity":"LOG","message":"duration: 12.345 ms statement: SELECT 1","query":""}}' + outputs: + - extract_from: cnpg_pg_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "litellm") + assert_eq!(.severity, "LOG") + assert_eq!(.labels.app, "cnpg") + assert_eq!(.labels.cluster, "litellm-postgres") + assert_eq!(.fields.error_severity, "LOG") + assert_eq!(.fields.duration_ms, "12.345") + assert_eq!(.fields.user, "litellm") + assert_eq!(.fields.database, "litellm") + + # --- Gitea router/access (k8s + VM) --- + - name: gitea_routes_k8s_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.gitea.gitea" + message: "routed" + outputs: + - extract_from: app_route.gitea + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: gitea_parse_router_line + inputs: + - insert_at: gitea_parse + type: log + log_fields: + subject: "logs.k8s.gitea.gitea" + kubernetes.pod_namespace: "gitea" + kubernetes.container_name: "gitea" + kubernetes.pod_node_name: "node-7" + message: '2026/08/01 00:00:00 .../router.go:100:func() [I] router: completed GET /user/login for 10.0.0.1:0, 200 OK in 12.3ms @ web/base.go:1' + outputs: + - extract_from: gitea_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "gitea") + assert_eq!(.labels.app, "gitea") + assert_eq!(.message, "GET /user/login 200") + assert_eq!(.fields.method, "GET") + assert_eq!(.fields.path, "/user/login") + assert_eq!(.fields.status, "200") + assert_eq!(.fields.latency, "12.3ms") + - name: gitea_parse_access_line + inputs: + - insert_at: gitea_parse + type: log + log_fields: + subject: "logs.k8s.gitea.gitea" + kubernetes.pod_namespace: "gitea" + kubernetes.container_name: "gitea" + message: '10.0.0.5 - alice [01/Aug/2026:00:00:00 +0000] "POST /repo/foo HTTP/1.1" 201 512 "-" "git/2.0"' + outputs: + - extract_from: gitea_parse + conditions: + - type: vrl + source: | + assert_eq!(.fields.method, "POST") + assert_eq!(.fields.path, "/repo/foo") + assert_eq!(.fields.status, "201") + assert_eq!(.fields.user, "alice") + assert_eq!(.fields.client_ip, "10.0.0.5") + + # --- PuppetServer / PuppetDB (k8s stdout) --- + - name: puppet_routes_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.puppet.puppetserver" + message: "routed" + outputs: + - extract_from: app_route.puppet + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: puppet_parse_logback_line + inputs: + - insert_at: puppet_parse + type: log + log_fields: + subject: "logs.k8s.puppet.puppetserver" + kubernetes.pod_namespace: "puppet" + kubernetes.container_name: "puppetserver" + kubernetes.pod_node_name: "node-8" + message: '2026-08-01 00:00:00,123 INFO [qtp123-45] [puppetserver] Compiled catalog for web01.unkin.net in environment production in 1.23 seconds' + outputs: + - extract_from: puppet_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "puppet") + assert_eq!(.severity, "INFO") + assert_eq!(.labels.app, "puppet") + assert_eq!(.fields.level, "INFO") + assert_eq!(.fields.logger, "puppetserver") + assert_eq!(.fields.node, "web01.unkin.net") + + # --- LiteLLM request logs (k8s JSON) --- + - name: litellm_routes_by_subject + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.k8s.litellm.litellm" + message: "routed" + outputs: + - extract_from: app_route.litellm + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: litellm_parse_extracts_request + inputs: + - insert_at: litellm_parse + type: log + log_fields: + subject: "logs.k8s.litellm.litellm" + kubernetes.pod_namespace: "litellm" + kubernetes.container_name: "litellm" + kubernetes.pod_node_name: "node-9" + message: '{"message":"Request completed","level":"info","model":"gpt-4o","total_tokens":1234,"response_time":0.532,"api_key":"sk-abc","status":"success","timestamp":"2026-08-01T00:00:00Z"}' + outputs: + - extract_from: litellm_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "k8s") + assert_eq!(.namespace, "litellm") + assert_eq!(.severity, "info") + assert_eq!(.message, "Request completed") + assert_eq!(.labels.app, "litellm") + assert_eq!(.fields.model, "gpt-4o") + assert_eq!(.fields.tokens, "1234") + assert_eq!(.fields.latency, "0.532") + assert_eq!(.fields.key, "sk-abc") + assert_eq!(.fields.status, "success") + + # --- Postfix maillog (VM, per-line best-effort) --- + - name: postfix_routes_by_identifier + inputs: + - insert_at: app_route + type: log + log_fields: + subject: "logs.vm.mail1_syd1" + SYSLOG_IDENTIFIER: "postfix/qmgr" + message: "routed" + outputs: + - extract_from: app_route.postfix + conditions: + - type: vrl + source: 'assert_eq!(.message, "routed")' + - name: postfix_parse_extracts_line + inputs: + - insert_at: postfix_parse + type: log + log_fields: + subject: "logs.vm.mail1_syd1" + host: "mail1" + SYSLOG_IDENTIFIER: "postfix/smtp" + message: 'ABC123DEF: to=, relay=mx.example.com[1.2.3.4]:25, delay=1.2, delays=0.1/0/0.5/0.6, dsn=2.0.0, status=sent (250 OK)' + outputs: + - extract_from: postfix_parse + conditions: + - type: vrl + source: | + assert_eq!(.source, "vm") + assert_eq!(.host, "mail1") + assert_eq!(.labels.app, "postfix") + assert_eq!(.fields.qid, "ABC123DEF") + assert_eq!(.fields.to, "rcpt@example.com") + assert_eq!(.fields.relay, "mx.example.com[1.2.3.4]:25") + assert_eq!(.fields.delay, "1.2") + assert_eq!(.fields.status, "sent") + assert_eq!(.fields.program, "postfix/smtp") diff --git a/apps/base/logging/vector/aggregator.yaml b/apps/base/logging/vector/aggregator.yaml index 83e37ed..4e445a3 100644 --- a/apps/base/logging/vector/aggregator.yaml +++ b/apps/base/logging/vector/aggregator.yaml @@ -62,8 +62,10 @@ transforms: inputs: - js_in route: - # k8s: authentik SSO — structlog JSON on stdout. - authentik: 'starts_with(to_string(.subject) ?? "", "logs.k8s.authentik.")' + # 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.")' @@ -78,6 +80,38 @@ transforms: # 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 @@ -451,6 +485,378 @@ transforms: "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 @@ -464,6 +870,13 @@ sinks: - 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 diff --git a/apps/overlays/au-syd1/gitea/values.yaml b/apps/overlays/au-syd1/gitea/values.yaml index 4ccef3a..baee19e 100644 --- a/apps/overlays/au-syd1/gitea/values.yaml +++ b/apps/overlays/au-syd1/gitea/values.yaml @@ -161,6 +161,15 @@ gitea: actions: ENABLED: false + # Router + NCSA access logs to stdout so the Tier-2 vector gitea pipeline can + # parse method/path/status/latency/user (subject logs.k8s.gitea.*). + log: + LEVEL: Info + MODE: console + ROUTER: console + ENABLE_ACCESS_LOG: true + ACCESS: console + repository: DEFAULT_BRANCH: main DEFAULT_PRIVATE: last