feat(promtail): per-service log pipeline stages

infra/loki/promtail-config.yml

@@ -38,3 +38,131 @@ scrape_configs:
       - labels:
           http_method:
           http_status:
+
+      # --- Traefik ---
+      - match:
+          selector: '{container="traefik"}'
+          stages:
+            - json:
+                expressions:
+                  RouterName: RouterName
+                  ServiceName: ServiceName
+                  DownstreamStatus: DownstreamStatus
+                  RequestMethod: RequestMethod
+            - labels:
+                tf_router: RouterName
+                tf_service: ServiceName
+                http_status: DownstreamStatus
+                http_method: RequestMethod
+            - regex:
+                expression: '(?P<tf_event>TLS handshake|EntryPoint|router|certResolver|middleware|tls)'
+            - labels:
+                tf_event:
+
+      # --- Gitea ---
+      - match:
+          selector: '{container="git"}'
+          stages:
+            - regex:
+                expression: '(?P<gitea_event>login|logout|register|push|pull|branch|tag|issue|webhook|oauth|created|deleted|merged)'
+            - labels:
+                gitea_event:
+
+      # --- oauth2-proxy ---
+      - match:
+          selector: '{container="oauth2-proxy"}'
+          stages:
+            - regex:
+                expression: '(?P<oauth_event>AuthSuccess|AuthFailure|OAuthStart|OAuthCallback|OAuthError|Invalid|Expired|Authenticated|Forbidden)'
+            - labels:
+                oauth_event:
+
+      # --- Postgres ---
+      - match:
+          selector: '{container="postgres"}'
+          stages:
+            - regex:
+                expression: '(?P<pg_event>FATAL|ERROR|WARNING|checkpoint|autovacuum|connection authorized|connection received|statement)'
+            - labels:
+                pg_event:
+
+      # --- Trino ---
+      - match:
+          selector: '{container="trino"}'
+          stages:
+            - regex:
+                expression: '(?P<trino_event>QUERY_CREATED|QUERY_STARTED|QUERY_COMPLETED|QUERY_FAILED|SPLIT_COMPLETED)'
+            - regex:
+                expression: '\b(?P<query_id>[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\b'
+            - labels:
+                trino_event:
+                query_id:
+
+      # --- Nessie / Polaris (Quarkus) ---
+      - match:
+          selector: '{container=~"nessie|polaris"}'
+          stages:
+            - regex:
+                expression: '"(?P<http_method>GET|POST|PUT|DELETE) (?P<quarkus_route>/[^ ?]+)'
+            - labels:
+                quarkus_route:
+
+      # --- RustFS ---
+      - match:
+          selector: '{container="rustfs"}'
+          stages:
+            - regex:
+                expression: '(?P<s3_op>PutObject|GetObject|DeleteObject|ListObjectsV2|HeadObject|CompleteMultipart)'
+            - regex:
+                expression: 'bucket=(?P<s3_bucket>[A-Za-z0-9._-]+)'
+            - labels:
+                s3_op:
+                s3_bucket:
+
+      # --- API (FastAPI/stack) ---
+      - match:
+          selector: '{container="api"}'
+          stages:
+            - regex:
+                expression: '(?P<api_event>pipeline|ingest|health|error)'
+            - regex:
+                expression: 'pipeline=(?P<pipeline_name>[A-Za-z0-9_.-]+)'
+            - labels:
+                api_event:
+                pipeline_name:
+
+      # --- mail-poller ---
+      - match:
+          selector: '{container="mail-poller"}'
+          stages:
+            - regex:
+                expression: '(?P<mail_event>ingest|poll|error)'
+            - regex:
+                expression: 'count=(?P<mail_count>\d+)'
+            - labels:
+                mail_event:
+                mail_count:
+
+      # --- act-runner ---
+      - match:
+          selector: '{container="act-runner"}'
+          stages:
+            - regex:
+                expression: '(?P<ci_event>job started|job finished|job failed|runner registered)'
+            - regex:
+                expression: 'job_id=(?P<job_id>\S+)'
+            - labels:
+                ci_event:
+                job_id:
+
+      # --- cloudflared ---
+      - match:
+          selector: '{container="cloudflared"}'
+          stages:
+            - regex:
+                expression: '(?P<cf_event>connection|registered|unregistered|reconnect|quic|error|tunnel|origin)'
+            - regex:
+                expression: 'status=(?P<cf_origin_status>\d{3})'
+            - labels:
+                cf_event:
+                cf_origin_status:

tests/test_observability.py

@@ -101,3 +101,29 @@ class TestPromtailRewire:
         docker_jobs = [j for j in jobs if j.get("job_name") == "docker"]
         assert docker_jobs, "expected job_name=docker"
         assert "docker_sd_configs" in docker_jobs[0]
+
+
+class TestPerServiceStages:
+    """Each named container must have a matching pipeline stage that emits at
+    least one service-specific label."""
+
+    EXPECTED = {
+        "traefik": "tf_event",
+        "git": "gitea_event",
+        "oauth2-proxy": "oauth_event",
+        "postgres": "pg_event",
+        "trino": "trino_event",
+        "rustfs": "s3_op",
+        "api": "api_event",
+        "mail-poller": "mail_event",
+        "act-runner": "ci_event",
+        "cloudflared": "cf_event",
+    }
+
+    def test_each_service_has_a_match_stage(self):
+        cfg = yaml.safe_load((ROOT / "infra/loki/promtail-config.yml").read_text())
+        stages = cfg["scrape_configs"][0]["pipeline_stages"]
+        text = yaml.safe_dump(stages)
+        for container, label in self.EXPECTED.items():
+            assert f'container="{container}"' in text, f"no match stage for {container}"
+            assert label in text, f"no {label} extraction for {container}"