unkinben
da4a66046a
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 <neotheo@gmail.com> Reviewed-on: #320 Co-authored-by: Ben Vincent <ben@unkin.net> Co-committed-by: Ben Vincent <ben@unkin.net>
901 lines
34 KiB
YAML
901 lines
34 KiB
YAML
---
|
|
# 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
|