unkin/argocd-apps
3
0
Fork 0
Code Issues 7 Pull Requests 3 Actions Packages Projects Releases Wiki Activity
Files
4fc4aed3583349fd57902a8a4b1d475be4542770
argocd-apps/apps/base/logging/vector/aggregator-tests.yaml
T
unkinben 8d70149467 Add Tier-1 per-app Vector transform pipelines (structured logs) (#318) 
Why: the logging aggregator wrote every event through the generic catch-all shape. The Tier-1 survey picked six high-value log sources that warrant structured parsing into logs.raw columns/fields for real querying.

How:
- Two-stage routing in `aggregator.yaml`: `app_route` peels off the six Tier-1 streams by subject / VM source tag (mutually exclusive — no double-insert); everything else falls through `app_route._unmatched` to the unchanged generic k8s/vm catch-all.
- Six parse transforms emit the full `logs.raw` shape plus structured `.fields` (Map(String,String) — no DDL change): authentik (JSON), traefik (JSON access), vault audit (JSON), nginx access+error (regex), haproxy httplog (regex), glauth (JSON).
- Companion flip: traefik-system access logs to `format: json` (both overlays) so `traefik_parse` has structured input.
- 15 new `vector test` cases (routing + field extraction) in `aggregator-tests.yaml`; all green locally (vector 0.57).

Live now: authentik + traefik (k8s). Awaiting the puppet-side vector rollout (logs.vm.* with `.file`/`.SYSLOG_IDENTIFIER` tags per the documented convention): vault-file, nginx, haproxy, glauth — transforms are present and unit-tested so they light up automatically.

Note: geoip enrichment for nginx/traefik client IPs is a separate prerequisite — no enrichment table exists in the aggregator yet; these transforms extract `client_ip` ready for it.

https://claude.ai/code/session_015ur3i7D2azsMAWTSVABApv
Reviewed-on: #318
Co-authored-by: Ben Vincent <ben@unkin.net>
Co-committed-by: Ben Vincent <ben@unkin.net>
2026-08-01 00:22:32 +10:00

356 lines
13 KiB
YAML
Raw Blame History

---
# `vector test` unit tests for the aggregator transforms. Merged with
# aggregator.yaml in CI (.woodpecker/vector-test.yaml). This is the pattern the
# per-app parsing follow-ups extend: add a test per new transform here.
tests:
- name: subject_routes_k8s_vs_vm
inputs:
- insert_at: route
type: log
log_fields:
subject: "logs.k8s.shop.web"
message: "routed"
outputs:
- extract_from: route.k8s
conditions:
- type: vrl
source: |
assert_eq!(.message, "routed")
- name: k8s_log_is_normalised
inputs:
- insert_at: k8s_shape
type: log
log_fields:
message: "hello from pod"
stream: "stdout"
timestamp: "2026-07-27T00:00:00Z"
kubernetes.pod_name: "web-abc"
kubernetes.pod_namespace: "shop"
kubernetes.container_name: "web"
kubernetes.pod_node_name: "node-1"
outputs:
- extract_from: k8s_shape
conditions:
- type: vrl
source: |
assert_eq!(.source, "k8s")
assert_eq!(.namespace, "shop")
assert_eq!(.pod, "web-abc")
assert_eq!(.container, "web")
assert_eq!(.host, "node-1")
assert_eq!(.stream, "stdout")
assert_eq!(.message, "hello from pod")
- name: vm_log_is_normalised
inputs:
- insert_at: vm_shape
type: log
log_fields:
message: "sshd started"
host: "vm-db-1"
severity: "info"
role: "database"
outputs:
- extract_from: vm_shape
conditions:
- type: vrl
source: |
assert_eq!(.source, "vm")
assert_eq!(.host, "vm-db-1")
assert_eq!(.severity, "info")
assert_eq!(.message, "sshd started")
assert_eq!(.labels.role, "database")
# --- catch-all preservation: an unclaimed k8s event still flows app_route ->
# generic route -> k8s_shape (proves the two-stage chain keeps the fallback) ---
- name: unclaimed_k8s_falls_through_to_generic
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.k8s.shop.web"
message: "plain app log"
outputs:
- extract_from: route.k8s
conditions:
- type: vrl
source: |
assert_eq!(.message, "plain app log")
# --- Tier-1: Authentik SSO (k8s, LIVE NOW) ---
- name: authentik_routes_by_subject
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.k8s.authentik.server"
message: "routed"
outputs:
- extract_from: app_route.authentik
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: authentik_parse_extracts_event
inputs:
- insert_at: authentik_parse
type: log
log_fields:
subject: "logs.k8s.authentik.server"
stream: "stdout"
kubernetes.pod_namespace: "authentik"
kubernetes.container_name: "server"
kubernetes.pod_node_name: "node-2"
message: '{"event":"login","action":"login","user":"alice","client_ip":"203.0.113.9","result":"success","level":"info","logger":"authentik.events","timestamp":"2026-07-27T00:00:00Z"}'
outputs:
- extract_from: authentik_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "k8s")
assert_eq!(.namespace, "authentik")
assert_eq!(.container, "server")
assert_eq!(.severity, "info")
assert_eq!(.message, "login")
assert_eq!(.labels.app, "authentik")
assert_eq!(.fields.event, "login")
assert_eq!(.fields.action, "login")
assert_eq!(.fields.user, "alice")
assert_eq!(.fields.client_ip, "203.0.113.9")
assert_eq!(.fields.result, "success")
# --- Tier-1: Traefik ingress (k8s, JSON access logs) ---
- name: traefik_routes_by_subject
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.k8s.traefik-system.traefik"
message: "routed"
outputs:
- extract_from: app_route.traefik
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: traefik_parse_extracts_access_fields
inputs:
- insert_at: traefik_parse
type: log
log_fields:
subject: "logs.k8s.traefik-system.traefik"
kubernetes.pod_namespace: "traefik-system"
kubernetes.container_name: "traefik"
kubernetes.pod_node_name: "node-3"
message: '{"RouterName":"web@kubernetes","ServiceName":"shop-svc@kubernetes","RequestMethod":"GET","RequestPath":"/api","RequestHost":"shop.example.net","RequestProtocol":"HTTP/1.1","DownstreamStatus":200,"Duration":5000000,"ClientHost":"203.0.113.5","StartUTC":"2026-07-27T00:00:00Z"}'
outputs:
- extract_from: traefik_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "k8s")
assert_eq!(.namespace, "traefik-system")
assert_eq!(.labels.app, "traefik")
assert_eq!(.message, "GET /api 200")
assert_eq!(.fields.route, "web@kubernetes")
assert_eq!(.fields.service, "shop-svc@kubernetes")
assert_eq!(.fields.method, "GET")
assert_eq!(.fields.path, "/api")
assert_eq!(.fields.host, "shop.example.net")
assert_eq!(.fields.status, "200")
assert_eq!(.fields.duration_ms, "5")
assert_eq!(.fields.client_ip, "203.0.113.5")
# --- Tier-1: Vault/OpenBao file audit (VM, awaiting VM vector) ---
- name: vault_routes_by_file
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.vm.vault1_syd1"
file: "/var/log/vault_audit.log"
message: "routed"
outputs:
- extract_from: app_route.vault
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: vault_parse_extracts_request
inputs:
- insert_at: vault_parse
type: log
log_fields:
subject: "logs.vm.vault1_syd1"
host: "vault1"
file: "/var/log/vault_audit.log"
message: '{"time":"2026-07-27T00:00:00Z","type":"response","auth":{"display_name":"token"},"request":{"operation":"read","path":"secret/data/app","remote_address":"10.0.0.9"},"error":""}'
outputs:
- extract_from: vault_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "vm")
assert_eq!(.host, "vault1")
assert_eq!(.labels.app, "vault")
assert_eq!(.message, "read secret/data/app")
assert_eq!(.fields.type, "response")
assert_eq!(.fields.display_name, "token")
assert_eq!(.fields.operation, "read")
assert_eq!(.fields.path, "secret/data/app")
assert_eq!(.fields.remote_address, "10.0.0.9")
# --- Tier-1: nginx access (VM, awaiting VM vector) ---
- name: nginx_access_routes_by_file
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.vm.web1_syd1"
file: "/var/log/nginx/shop_access.log"
message: "routed"
outputs:
- extract_from: app_route.nginx_access
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: nginx_access_parse_extracts_combined
inputs:
- insert_at: nginx_access_parse
type: log
log_fields:
subject: "logs.vm.web1_syd1"
host: "web1"
file: "/var/log/nginx/shop_access.log"
message: '192.0.2.10 - - [27/Jul/2026:00:00:00 +0000] "GET /index.html HTTP/1.1" 200 1024 "https://ref.example/" "Mozilla/5.0" 0.012'
outputs:
- extract_from: nginx_access_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "vm")
assert_eq!(.stream, "access")
assert_eq!(.labels.log_type, "access")
assert_eq!(.fields.client_ip, "192.0.2.10")
assert_eq!(.fields.method, "GET")
assert_eq!(.fields.path, "/index.html")
assert_eq!(.fields.status, "200")
assert_eq!(.fields.bytes, "1024")
assert_eq!(.fields.referer, "https://ref.example/")
assert_eq!(.fields.user_agent, "Mozilla/5.0")
assert_eq!(.fields.request_time, "0.012")
# --- Tier-1: nginx error (VM, awaiting VM vector) ---
- name: nginx_error_routes_by_file
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.vm.web1_syd1"
file: "/var/log/nginx/shop_error.log"
message: "routed"
outputs:
- extract_from: app_route.nginx_error
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: nginx_error_parse_extracts_fields
inputs:
- insert_at: nginx_error_parse
type: log
log_fields:
subject: "logs.vm.web1_syd1"
host: "web1"
file: "/var/log/nginx/shop_error.log"
message: '2026/07/27 00:00:00 [error] 1234#0: *5 open() "/var/www/x" failed (2: No such file or directory), client: 192.0.2.20, server: shop, request: "GET / HTTP/1.1", host: "shop"'
outputs:
- extract_from: nginx_error_parse
conditions:
- type: vrl
source: |
assert_eq!(.stream, "error")
assert_eq!(.severity, "error")
assert_eq!(.labels.log_type, "error")
assert_eq!(.fields.level, "error")
assert_eq!(.fields.pid, "1234")
assert_eq!(.fields.cid, "5")
assert_eq!(.fields.client_ip, "192.0.2.20")
# --- Tier-1: HAProxy httplog (VM journald, awaiting VM vector) ---
- name: haproxy_routes_by_identifier
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.vm.halb1_syd1"
SYSLOG_IDENTIFIER: "haproxy"
message: "routed"
outputs:
- extract_from: app_route.haproxy
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: haproxy_parse_extracts_timers
inputs:
- insert_at: haproxy_parse
type: log
log_fields:
subject: "logs.vm.halb1_syd1"
host: "halb1"
SYSLOG_IDENTIFIER: "haproxy"
message: '192.0.2.30:54321 [27/Jul/2026:00:00:00.123] fe_http be_app/app1 10/0/1/2/13 200 512 - - ---- 5/4/3/2/0 0/0 "GET /health HTTP/1.1"'
outputs:
- extract_from: haproxy_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "vm")
assert_eq!(.labels.app, "haproxy")
assert_eq!(.fields.client_ip, "192.0.2.30")
assert_eq!(.fields.frontend, "fe_http")
assert_eq!(.fields.backend, "be_app")
assert_eq!(.fields.server, "app1")
assert_eq!(.fields.tq, "10")
assert_eq!(.fields.tw, "0")
assert_eq!(.fields.tc, "1")
assert_eq!(.fields.tr, "2")
assert_eq!(.fields.tt, "13")
assert_eq!(.fields.termination_state, "----")
assert_eq!(.fields.retries, "0")
assert_eq!(.fields.status, "200")
assert_eq!(.fields.bytes, "512")
# --- Tier-1: glauth LDAP (VM, awaiting VM vector) ---
- name: glauth_routes_by_identifier
inputs:
- insert_at: app_route
type: log
log_fields:
subject: "logs.vm.ldap1_syd1"
SYSLOG_IDENTIFIER: "glauth"
message: "routed"
outputs:
- extract_from: app_route.glauth
conditions:
- type: vrl
source: 'assert_eq!(.message, "routed")'
- name: glauth_parse_extracts_bind
inputs:
- insert_at: glauth_parse
type: log
log_fields:
subject: "logs.vm.ldap1_syd1"
host: "ldap1"
SYSLOG_IDENTIFIER: "glauth"
message: '{"level":"info","msg":"Bind success as user","bindDN":"cn=admin,dc=example,dc=com","src":"192.0.2.40:1234","time":"2026-07-27T00:00:00Z"}'
outputs:
- extract_from: glauth_parse
conditions:
- type: vrl
source: |
assert_eq!(.source, "vm")
assert_eq!(.host, "ldap1")
assert_eq!(.severity, "info")
assert_eq!(.labels.app, "glauth")
assert_eq!(.fields.bindDN, "cn=admin,dc=example,dc=com")
assert_eq!(.fields.remote, "192.0.2.40:1234")
assert_eq!(.fields.success, "true")
Reference in New Issue View Git Blame Copy Permalink