feat(backend): log one wide event per request with the trace id

backend/.gitignore

@@ -173,3 +173,4 @@ cython_debug/
 
 # PyPI configuration file
 .pypirc
+.perf-benchmarks/

backend/pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "pyyaml>=6.0",
     "fastapi-sqlalchemy-monitor>=1.1.3",
     "cr-service[tracing]>=0.6.0",
+    "prometheus-fastapi-instrumentator>=7.0.0",
     "opentelemetry-instrumentation-sqlalchemy>=0.51b0",
 ]
 

backend/scripts/perf-compare.sh

@@ -10,10 +10,12 @@ set -euo pipefail
 BASE="${1:-origin/main}"
 BACKEND="$(cd "$(dirname "$0")/.." && pwd)"
 WORK="$(mktemp -d)"
-trap 'git -C "$BACKEND" worktree remove --force "$WORK/base" >/dev/null 2>&1 || true; rm -rf "$WORK"' EXIT
+# Inside this checkout, because pytest-benchmark crashes printing its machine_info warning for storage outside it.
+BENCHMARKS="$BACKEND/.perf-benchmarks/$(basename "$WORK")"
+trap 'git -C "$BACKEND" worktree remove --force "$WORK/base" >/dev/null 2>&1 || true; rm -rf "$WORK" "$BENCHMARKS"' EXIT
 
 export REF_PERF_SCALE="${REF_PERF_SCALE:-0.1}"
-STORAGE="file://$WORK/benchmarks"
+STORAGE="file://$BENCHMARKS"
 
 git -C "$BACKEND" worktree add --quiet --detach "$WORK/base" "$BASE"
 

backend/src/ref_backend/builder.py

@@ -1,21 +1,24 @@
-from dataclasses import asdict
+from collections.abc import Awaitable, Callable
 from importlib.metadata import version
 
 import sentry_sdk
-from cr_service import metrics, tracing
+from cr_service import context, metrics, tracing
 from cr_service.health import router as health_router
 from cr_service.info import ServiceInfo
-from fastapi import FastAPI
+from cr_service.middleware import WideEventMiddleware
+from fastapi import FastAPI, Request
 from fastapi.routing import APIRoute
 from fastapi_sqlalchemy_monitor import AlchemyStatistics, SQLAlchemyMonitor
 from fastapi_sqlalchemy_monitor.action import Action, ConditionalAction, WarnMaxTotalInvocation
 from loguru import logger
+from prometheus_fastapi_instrumentator import routing
+from sqlalchemy import Engine
 from starlette import status
 from starlette.exceptions import HTTPException
 from starlette.middleware.cors import CORSMiddleware
 from starlette.responses import Response
 from starlette.staticfiles import StaticFiles
-from starlette.types import Scope
+from starlette.types import ASGIApp, Receive, Scope, Send
 
 from climate_ref.config import Config
 from climate_ref.database import Database
@@ -45,6 +48,25 @@ def custom_generate_unique_id(route: APIRoute) -> str:
     return f"{route.tags[0]}-{route.name}" if route.tags else route.name
 
 
+class RequestQueryMonitor(SQLAlchemyMonitor):
+    """Runs the actions even when the endpoint raises, so a failed request keeps its query statistics."""
+
+    def __init__(self, app: ASGIApp, *, engine: Engine, actions: list[Action]) -> None:
+        super().__init__(app, engine=engine, actions=actions)
+        self.actions = actions
+
+    async def _dispatch(
+        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
+    ) -> Response:
+        statistics = AlchemyStatistics()
+        self.request_context.set(statistics)
+        try:
+            return await call_next(request)
+        finally:
+            for action in self.actions:
+                action.handle(statistics)
+
+
 class SlowQueryMonitor(ConditionalAction):
     def __init__(self, threshold_ms: float):
         self.threshold_ms = threshold_ms
@@ -64,12 +86,27 @@ def _handle(self, statistics: AlchemyStatistics) -> None:
                 )
 
 
-class LogStatistics(Action):
-    """Action that logs current statistics."""
+class BindQueryStatistics(Action):
+    """Put the request's query count and time on its wide event."""
 
     def handle(self, statistics: AlchemyStatistics) -> None:
-        if statistics.total_invocations > 0:
-            logger.info(asdict(statistics))
+        context.bind(
+            db_queries=statistics.total_invocations,
+            db_time_ms=round(statistics.total_invocation_time_ms, 2),
+        )
+
+
+class BindRouteMiddleware:
+    """Put the route template on the wide event, named as the `handler` label in /metrics names it."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] == "http":
+            # None for a path no route matches.
+            context.bind(route=routing.get_route_name(Request(scope), should_include_root_path=False))
+        await self.app(scope, receive, send)
 
 
 class SPAStaticFiles(StaticFiles):
@@ -125,10 +162,10 @@ def build_app(settings: Settings, ref_config: Config, database: Database) -> Fas
         },
     )
     app.add_middleware(
-        SQLAlchemyMonitor,
+        RequestQueryMonitor,
         engine=database._engine,
         actions=[
-            LogStatistics(),
+            BindQueryStatistics(),
             WarnMaxTotalInvocation(max_invocations=10),
             SlowQueryMonitor(threshold_ms=100),
         ],
@@ -151,6 +188,9 @@ def build_app(settings: Settings, ref_config: Config, database: Database) -> Fas
         api_max_age=settings.API_CACHE_MAX_AGE,
         results_max_age=settings.RESULTS_CACHE_MAX_AGE,
     )
+    # Outermost, so the wide event times the whole stack and sees the query statistics.
+    app.add_middleware(BindRouteMiddleware)
+    app.add_middleware(WideEventMiddleware)
 
     # Registered before the static mount, so the SPA fallback does not swallow the probes or /metrics.
     app.include_router(health_router)

backend/src/ref_backend/log.py

@@ -1,7 +1,22 @@
+import inspect
+import json
 import logging
+import sys
+import traceback
+from datetime import UTC
+from typing import TYPE_CHECKING, Any
 
+from cr_service.context import get_context
+from cr_service.redact import redact_fields
+from cr_service.tracing import current_trace_context
 from loguru import logger
 
+if TYPE_CHECKING:
+    from loguru import Message, Record
+
+
+_RESERVED_LOG_RECORD_KEYS = frozenset(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
+
 
 class InterceptHandler(logging.Handler):
     def emit(self, record: logging.LogRecord) -> None:
@@ -12,15 +27,58 @@ def emit(self, record: logging.LogRecord) -> None:
             level = record.levelno  # type: ignore[assignment]
 
         # Find caller to get correct stack depth
-        frame, depth = logging.currentframe(), 2
-        while frame.f_back and frame.f_code.co_filename == logging.__file__:
+        frame, depth = inspect.currentframe(), 0
+        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
             frame = frame.f_back
             depth += 1
 
-        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+        # Carries `extra` across, which is where cr-service puts the wide event's fields.
+        fields = {
+            key: value for key, value in record.__dict__.items() if key not in _RESERVED_LOG_RECORD_KEYS
+        }
+        logger.bind(**fields).opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+
+
+def _add_request_context(record: "Record") -> None:
+    record["extra"] = redact_fields({**current_trace_context(), **get_context(), **record["extra"]})
+
 
+def _json_sink(message: "Message") -> None:
+    """Write one compact JSON object per line, in the shape cr-service services log."""
+    record = message.record
+    payload: dict[str, Any] = {
+        "ts": record["time"].astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z"),
+        "level": record["level"].name.lower(),
+        "logger": record["name"],
+        "message": record["message"],
+        **record["extra"],
+    }
+    if record["exception"]:
+        payload["exc"] = "".join(traceback.format_exception(*record["exception"]))
+    # Compact separators, because Grafana matches `"trace_id":"<id>"` to link a line to its trace.
+    sys.stdout.write(json.dumps(payload, default=str, separators=(",", ":")) + "\n")
+    sys.stdout.flush()
+
+
+def _text_format(record: "Record") -> str:
+    """Loguru's default line, followed by the bound fields as ``key=value`` pairs."""
+    fields = " ".join(f"{key}={value!r}" for key, value in record["extra"].items() if key != "_fields")
+    record["extra"]["_fields"] = f" | {fields}" if fields else ""
+    return (
+        "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | "
+        "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
+        "{extra[_fields]}\n{exception}"
+    )
+
+
+def setup_logging(log_level: str, json_logs: bool = False) -> None:
+    logger.configure(patcher=_add_request_context)
+    logger.remove()
+    if json_logs:
+        logger.add(_json_sink, level=log_level.upper())
+    else:
+        logger.add(sys.stderr, level=log_level.upper(), format=_text_format)
 
-def setup_logging(log_level: str) -> None:
     # Remove existing handlers
     for handler in logging.root.handlers[:]:
         logging.root.removeHandler(handler)
@@ -31,7 +89,6 @@ def setup_logging(log_level: str) -> None:
     # List of loggers to intercept
     loggers = (
         "uvicorn",
-        "uvicorn.access",
         "uvicorn.error",
         "fastapi",
         "asyncio",
@@ -42,3 +99,8 @@ def setup_logging(log_level: str) -> None:
         logging_logger = logging.getLogger(logger_name)
         logging_logger.handlers = []
         logging_logger.propagate = True
+
+    # The wide event replaces uvicorn's access line.
+    access = logging.getLogger("uvicorn.access")
+    access.handlers = []
+    access.propagate = False

backend/src/ref_backend/main.py

@@ -32,7 +32,7 @@
 database = deps._get_database_dependency(settings, ref_config)
 provider_registry = get_provider_registry(ref_config, read_only=settings.REF_READ_ONLY_DATABASE)
 
-setup_logging(settings.LOG_LEVEL)
+setup_logging(settings.LOG_LEVEL, json_logs=settings.ENVIRONMENT != "local")
 app = build_app(settings, ref_config, database)
 
 

backend/tests/test_log.py

@@ -0,0 +1,79 @@
+import json
+import re
+import sys
+
+import pytest
+from loguru import logger
+from opentelemetry import trace
+from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags
+
+from ref_backend.log import setup_logging
+
+TRACE_ID = 0x4BF92F3577B34DA6A3CE929D0E0E4736
+SPAN_ID = 0x00F067AA0BA902B7
+# The derived field Grafana uses to link a Loki line to its Tempo trace.
+GRAFANA_TRACE_ID = re.compile(r'"trace_id":"(\w+)"')
+
+
+@pytest.fixture
+def restore_loguru():
+    yield
+    logger.remove()
+    logger.configure(patcher=None)
+    logger.add(sys.stderr)
+
+
+def test_json_lines_carry_the_active_trace(restore_loguru, capsys):
+    setup_logging("INFO", json_logs=True)
+    span = NonRecordingSpan(SpanContext(TRACE_ID, SPAN_ID, is_remote=False, trace_flags=TraceFlags(1)))
+
+    with trace.use_span(span):
+        logger.warning("Slow query detected")
+
+    line = capsys.readouterr().out.strip()
+    payload = json.loads(line)
+    assert payload["level"] == "warning"
+    assert payload["message"] == "Slow query detected"
+    assert payload["span_id"] == "00f067aa0ba902b7"
+    assert GRAFANA_TRACE_ID.search(line).group(1) == "4bf92f3577b34da6a3ce929d0e0e4736"
+
+
+def test_json_lines_outside_a_span_have_no_trace(restore_loguru, capsys):
+    setup_logging("INFO", json_logs=True)
+
+    logger.info("Starting up")
+
+    payload = json.loads(capsys.readouterr().out)
+    assert "trace_id" not in payload
+
+
+def test_json_lines_include_the_exception(restore_loguru, capsys):
+    setup_logging("INFO", json_logs=True)
+
+    try:
+        raise ValueError("boom")
+    except ValueError:
+        logger.exception("Request failed")
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["level"] == "error"
+    assert "ValueError: boom" in payload["exc"]
+
+
+def test_text_lines_show_the_bound_fields(restore_loguru, capsys):
+    setup_logging("INFO")
+
+    logger.bind(event="http_request", status=404, route="/api/v1/executions/{group_id}").info("http_request")
+
+    line = capsys.readouterr().err
+    assert "status=404" in line
+    assert "route='/api/v1/executions/{group_id}'" in line
+
+
+def test_secret_fields_are_masked(restore_loguru, capsys):
+    setup_logging("INFO", json_logs=True)
+
+    logger.bind(api_key="sk-live-123").info("Calling upstream")
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["api_key"] != "sk-live-123"