Merge pull request #48 from trick77/feat/splunk-friendly-logs

feat(logging): Splunk-friendly JSON logs + request correlation

Author: trick77

AGENTS.md

@@ -48,6 +48,26 @@ If `docker ps` fails, ask the user to start OrbStack.
 - Tests use real Postgres via testcontainers, never SQLite. The `client` fixture in `tests/conftest.py` depends on `session_factory` which truncates tables per test.
 - `.pre-commit-config.yaml` runs ruff + pyright + uv-lock-check; expect CI to enforce the same.
 
+## Logging & Splunk
+
+- **One JSON object per line, on stdout.** OpenShift's Splunk Connect for Kubernetes (SCK) tails the container log; Splunk auto-extracts fields via `KV_MODE=json` for sourcetype `riptide:collector:json` (set as pod annotation in `openshift/collector/deployment.yaml`).
+- **Stdlib loggers (uvicorn, sqlalchemy, alembic) are bridged through structlog.** Do NOT add separate logging handlers or re-init `logging.basicConfig` — `configure_logging()` in `logging_config.py` is the single entry point.
+- **Splunk-reserved field names are forbidden as kwargs**: `source`, `sourcetype`, `host`, `index`, `time`, `_time`, `_raw`, `event`. The CI vendor field is `ci_system` (not `source`); the structlog event name lives in `msg` (renamed from `event`); severity lives in `log_level` (renamed from `level`). A runtime processor (`_strip_reserved`) namespaces accidental reserved kwargs under `splunk_<name>` as a safety net — do not rely on it; pick the right name from the start.
+- **Webhook handlers emit exactly one `msg=webhook_processed` log per request** with required fields `webhook_source ∈ {bitbucket,pipeline,argocd,noergler}`, `outcome ∈ {accepted,deduped,ignored,skipped}`, `delivery_id`, `team`. Source-specific fields go alongside (e.g. `app`, `revision`, `phase` for argocd). Include `delivery_id` even on `ignored`/`skipped` paths so triage has a key.
+- **`outcome=deduped`** is detected via `RETURNING delivery_id` on the `INSERT ... ON CONFLICT DO NOTHING` — a `None` scalar means the row already existed. Preserve this when adding new sources.
+- **Persist failures**: wrap the `async with session_factory()` block in `try/except Exception: logger.exception("webhook_persist_failed", ...); raise`. Never swallow.
+- **Access log** is emitted by the `access_log` middleware in `main.py` as `msg=http_request` with `request_id`, `method`, `path`, `status_code`, `duration_ms`. `request_id` is bound to contextvars so any log within the request inherits it. `/health` and `/ready` are silenced; uvicorn.access is set to WARNING (do not lower it).
+- **Splunk `props.conf` snippet** (owned by platform team, kept here for reference):
+  ```
+  [riptide:collector:json]
+  SHOULD_LINEMERGE = false
+  LINE_BREAKER     = ([\r\n]+)
+  KV_MODE          = json
+  TIME_PREFIX      = "timestamp":\s*"
+  TIME_FORMAT      = %Y-%m-%dT%H:%M:%S.%6NZ
+  TRUNCATE         = 0
+  ```
+
 ## OpenShift layout
 
 `openshift/` is **suite-level**, structured per-component. The collector lives in `openshift/collector/`. When adding a new component:

openshift/collector/configmap-app.yaml

@@ -4,6 +4,8 @@ metadata:
   name: riptide-collector-app
 data:
   RIPTIDE_LOG_LEVEL: "INFO"
+  # Stamped on every log record as `env`. Override per overlay (intg / prod).
+  RIPTIDE_ENV: "prod"
   RIPTIDE_CONFIG_PATH: "/etc/riptide-collector/riptide.json"
   RIPTIDE_TEAM_KEYS_PATH: "/etc/riptide-collector-team-keys/team-keys.json"
   RIPTIDE_CONFIG_RELOAD_SECONDS: "30"

openshift/collector/deployment.yaml

@@ -18,6 +18,12 @@ spec:
       labels:
         app.kubernetes.io/name: riptide-collector
         app.kubernetes.io/component: collector
+      annotations:
+        # Splunk Connect for Kubernetes reads these per-pod annotations.
+        # The sourcetype must have `KV_MODE=json` configured in props.conf
+        # so Splunk auto-extracts our JSON log fields. The index annotation
+        # can be overridden per overlay.
+        splunk.com/sourcetype: "riptide:collector:json"
     spec:
       securityContext:
         runAsNonRoot: true

src/riptide_collector/auth.py

@@ -125,7 +125,8 @@ async def verify_hmac_signature(
         signature_ok = _verify_hmac_sha256(verify_secret, raw, x_hub_signature)
         if secret is None or not signature_ok:
             logger.warning(
-                f"{source}_hmac_rejected",
+                "hmac_rejected",
+                webhook_source=source,
                 team=team,
                 has_secret=secret is not None,
                 has_signature=bool(x_hub_signature),

src/riptide_collector/logging_config.py

@@ -4,41 +4,91 @@
 
 import structlog
 
+from riptide_collector import __version__
 
-class _HealthCheckAccessFilter(logging.Filter):
-    _silent_paths = frozenset({"/health", "/ready"})
+# Field names Splunk treats as built-in input metadata. Using them as JSON
+# keys causes Splunk to silently overwrite our values with the forwarder's
+# (`source`, `host`, `index`, `sourcetype`, `time`/`_time`/`_raw`), or to
+# double-extract (`event`). Keep them off the wire.
+_SPLUNK_RESERVED = frozenset(
+    {"source", "sourcetype", "host", "index", "time", "_time", "_raw", "event"}
+)
 
-    def filter(self, record: logging.LogRecord) -> bool:
-        args = record.args
-        if isinstance(args, tuple) and len(args) >= 3:
-            path = args[2]
-            if isinstance(path, str) and path.split("?", 1)[0] in self._silent_paths:
-                return False
-        return True
+_SERVICE_NAME = "riptide-collector"
 
 
-def configure_logging(level: str = "INFO") -> None:
+def _make_service_metadata_processor(env: str):
+    def _add(_logger: Any, _name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
+        del _logger, _name
+        event_dict.setdefault("service", _SERVICE_NAME)
+        event_dict.setdefault("version", __version__)
+        event_dict.setdefault("env", env)
+        return event_dict
+
+    return _add
+
+
+def _rename_level(_logger: Any, _name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
+    del _logger, _name
+    if "level" in event_dict and "log_level" not in event_dict:
+        event_dict["log_level"] = event_dict.pop("level")
+    return event_dict
+
+
+def _strip_reserved(_logger: Any, _name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
+    del _logger, _name
+    # `msg` is set by EventRenamer, `log_level` by _rename_level — both
+    # already safe. This pass catches accidental kwargs (e.g. `source="jenkins"`)
+    # and namespaces them under `splunk_<name>` so the value survives without
+    # colliding with Splunk's built-in field of the same name.
+    for key in list(event_dict.keys()):
+        if key in _SPLUNK_RESERVED:
+            event_dict[f"splunk_{key}"] = event_dict.pop(key)
+    return event_dict
+
+
+def configure_logging(level: str = "INFO", env: str = "dev") -> None:
     log_level = getattr(logging, level.upper(), logging.INFO)
-    logging.basicConfig(
-        format="%(message)s",
-        stream=sys.stdout,
-        level=log_level,
-    )
-    logging.getLogger("uvicorn.access").addFilter(_HealthCheckAccessFilter())
 
-    processors: list[Any] = [
+    shared_processors: list[Any] = [
         structlog.contextvars.merge_contextvars,
         structlog.processors.add_log_level,
         structlog.processors.TimeStamper(fmt="iso", utc=True),
         structlog.processors.StackInfoRenderer(),
         structlog.processors.format_exc_info,
-        structlog.processors.JSONRenderer(),
+        _make_service_metadata_processor(env),
+        structlog.processors.EventRenamer("msg"),
+        _rename_level,
+        _strip_reserved,
     ]
 
+    # Bridge: route stdlib logs (uvicorn, sqlalchemy, alembic) through the
+    # same JSON pipeline so Splunk sees one schema only.
+    formatter = structlog.stdlib.ProcessorFormatter(
+        foreign_pre_chain=shared_processors,
+        processors=[
+            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
+            structlog.processors.JSONRenderer(),
+        ],
+    )
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(formatter)
+    root = logging.getLogger()
+    root.handlers = [handler]
+    root.setLevel(log_level)
+
+    # uvicorn's access log becomes redundant once our middleware emits
+    # http_request; keep only warnings/errors from it.
+    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
+    logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
+
     structlog.configure(
-        processors=processors,
+        processors=[
+            *shared_processors,
+            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
+        ],
         wrapper_class=structlog.make_filtering_bound_logger(log_level),
-        logger_factory=structlog.PrintLoggerFactory(),
+        logger_factory=structlog.stdlib.LoggerFactory(),
         cache_logger_on_first_use=True,
     )
 

src/riptide_collector/main.py

@@ -1,7 +1,10 @@
+import time
+import uuid
 from contextlib import asynccontextmanager
 from typing import Any
 
-from fastapi import FastAPI
+import structlog
+from fastapi import FastAPI, Request, Response
 
 from riptide_collector import __version__
 from riptide_collector.auth import make_hmac_dependency, make_team_bearer_dependency
@@ -14,6 +17,8 @@
 
 logger = get_logger(__name__)
 
+_SILENT_PATHS = frozenset({"/health", "/ready"})
+
 
 class StartupValidationError(RuntimeError):
     pass
@@ -37,7 +42,7 @@ def _cross_validate(config: RiptideConfigStore, team_keys: TeamKeysStore) -> Non
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     settings = settings or load_settings()
-    configure_logging(settings.log_level)
+    configure_logging(settings.log_level, env=settings.env)
 
     config = RiptideConfigStore(settings.config_path)
     team_keys = TeamKeysStore(settings.team_keys_path)
@@ -76,6 +81,34 @@ async def lifespan(_app: FastAPI):  # pyright: ignore[reportUnusedFunction]
         lifespan=lifespan,
     )
 
+    @app.middleware("http")
+    async def access_log(  # pyright: ignore[reportUnusedFunction]
+        request: Request, call_next: Any
+    ) -> Response:
+        # Liveness/readiness checks fire every few seconds; logging them
+        # buries real traffic in Splunk. Pass through unobserved.
+        if request.url.path in _SILENT_PATHS:
+            return await call_next(request)
+        request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
+        structlog.contextvars.bind_contextvars(
+            request_id=request_id,
+            method=request.method,
+            path=request.url.path,
+        )
+        started = time.perf_counter()
+        status_code = 500
+        try:
+            response: Response = await call_next(request)
+            status_code = response.status_code
+            return response
+        finally:
+            logger.info(
+                "http_request",
+                status_code=status_code,
+                duration_ms=round((time.perf_counter() - started) * 1000, 1),
+            )
+            structlog.contextvars.clear_contextvars()
+
     app.include_router(health.make_router(config, session_factory, team_keys, any_auth))
     app.include_router(bitbucket.make_router(config, session_factory, bitbucket_hmac))
     app.include_router(pipeline.make_router(session_factory, pipeline_auth))

src/riptide_collector/routers/argocd.py

@@ -35,56 +35,72 @@ async def argocd_webhook(  # pyright: ignore[reportUnusedFunction]
         revision = lower(event.revision)
         environment = parse_environment(event.destination_namespace)
 
+        # Compute the dedup key unconditionally so the `ignored` log line
+        # carries it too — Triage starts from `delivery_id`.
+        started_repr = event.started_at.isoformat() if event.started_at else "unknown"
+        delivery_id = (
+            f"{event.app_name}#{revision}#{started_repr}#{event.operation_phase or 'unknown'}"
+        )
+
         ignored_stages = config.get().environments.ignored_stages
         if environment is not None and environment in ignored_stages:
             logger.info(
-                "argocd_event_ignored",
+                "webhook_processed",
+                webhook_source="argocd",
+                outcome="ignored",
+                reason="stage_in_ignored_stages",
+                delivery_id=delivery_id,
                 app=event.app_name,
+                revision=revision,
+                phase=event.operation_phase,
                 environment=environment,
                 destination_namespace=event.destination_namespace,
                 team=caller_team,
             )
             return {"status": "ignored"}
 
-        # Stable dedup key: started_at is fixed for a sync attempt; phase varies
-        # across the lifecycle (Running → Succeeded/Failed) and SHOULD produce
-        # distinct rows. finished_at is excluded — it can drift between retries
-        # of the same phase and would cause duplicates.
-        started_repr = event.started_at.isoformat() if event.started_at else "unknown"
-        delivery_id = (
-            f"{event.app_name}#{revision}#{started_repr}#{event.operation_phase or 'unknown'}"
-        )
-
-        async with session_factory() as session:
-            stmt = (
-                pg_insert(ArgoCDEvent)
-                .values(
-                    delivery_id=delivery_id,
-                    app_name=event.app_name,
-                    revision=revision,
-                    sync_status=event.sync_status,
-                    operation_phase=event.operation_phase,
-                    started_at=event.started_at,
-                    finished_at=event.finished_at,
-                    occurred_at=event.finished_at or event.started_at or datetime.now(UTC),
-                    team=caller_team,
-                    destination_namespace=event.destination_namespace,
-                    environment=environment,
-                    payload=raw,
+        try:
+            async with session_factory() as session:
+                stmt = (
+                    pg_insert(ArgoCDEvent)
+                    .values(
+                        delivery_id=delivery_id,
+                        app_name=event.app_name,
+                        revision=revision,
+                        sync_status=event.sync_status,
+                        operation_phase=event.operation_phase,
+                        started_at=event.started_at,
+                        finished_at=event.finished_at,
+                        occurred_at=event.finished_at or event.started_at or datetime.now(UTC),
+                        team=caller_team,
+                        destination_namespace=event.destination_namespace,
+                        environment=environment,
+                        payload=raw,
+                    )
+                    .on_conflict_do_nothing(index_elements=["delivery_id"])
+                    .returning(ArgoCDEvent.delivery_id)
                 )
-                .on_conflict_do_nothing(index_elements=["delivery_id"])
+                inserted = (await session.execute(stmt)).scalar_one_or_none()
+                await session.commit()
+        except Exception:
+            logger.exception(
+                "webhook_persist_failed",
+                webhook_source="argocd",
+                delivery_id=delivery_id,
+                team=caller_team,
             )
-            await session.execute(stmt)
-            await session.commit()
+            raise
 
         logger.info(
-            "argocd_event_received",
+            "webhook_processed",
+            webhook_source="argocd",
+            outcome="accepted" if inserted is not None else "deduped",
             delivery_id=delivery_id,
             app=event.app_name,
             revision=revision,
             phase=event.operation_phase,
-            team=caller_team,
             environment=environment,
+            team=caller_team,
         )
         return {"status": "accepted"}