diff --git a/backend/.gitignore b/backend/.gitignore index 258ec27e..0b47bf0e 100644 --- a/backend/.gitignore +++ b/backend/.gitignore @@ -173,3 +173,4 @@ cython_debug/ # PyPI configuration file .pypirc +.perf-benchmarks/ diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 75a78040..0347db20 100644 --- a/backend/pyproject.toml +++ b/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", ] diff --git a/backend/scripts/perf-compare.sh b/backend/scripts/perf-compare.sh index a90cd263..ac77acae 100755 --- a/backend/scripts/perf-compare.sh +++ b/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" diff --git a/backend/src/ref_backend/builder.py b/backend/src/ref_backend/builder.py index 9b327e2c..dcaff5ac 100644 --- a/backend/src/ref_backend/builder.py +++ b/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) diff --git a/backend/src/ref_backend/log.py b/backend/src/ref_backend/log.py index b0593af0..d409b367 100644 --- a/backend/src/ref_backend/log.py +++ b/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":""` 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 ( + "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | " + "{name}:{function}:{line} - {message}" + "{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 diff --git a/backend/src/ref_backend/main.py b/backend/src/ref_backend/main.py index 32529d14..c8327103 100644 --- a/backend/src/ref_backend/main.py +++ b/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) diff --git a/backend/tests/test_log.py b/backend/tests/test_log.py new file mode 100644 index 00000000..9043a24c --- /dev/null +++ b/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" diff --git a/backend/tests/test_tracing.py b/backend/tests/test_tracing.py index 25e3c64f..d5b989c5 100644 --- a/backend/tests/test_tracing.py +++ b/backend/tests/test_tracing.py @@ -1,8 +1,19 @@ +import json +import sys + import pytest +import sqlalchemy +from cr_service.middleware import WideEventMiddleware +from fastapi import FastAPI +from loguru import logger from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from starlette.testclient import TestClient + +from ref_backend.builder import BindQueryStatistics, RequestQueryMonitor +from ref_backend.log import setup_logging TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736" @@ -28,3 +39,58 @@ def test_request_span_continues_the_browser_trace(exporter, client): assert spans assert trace.format_trace_id(spans[0].context.trace_id) == TRACE_ID assert spans[0].attributes["http.route"] == "/api/v1/utils/about" + + +@pytest.fixture +def json_logs(capsys): + setup_logging("INFO", json_logs=True) + capsys.readouterr() + yield lambda: [json.loads(line) for line in capsys.readouterr().out.splitlines()] + logger.remove() + logger.configure(patcher=None) + logger.add(sys.stderr) + + +def test_each_request_logs_one_wide_event(exporter, client, json_logs): + client.get("/api/v1/diagnostics/", headers={"traceparent": f"00-{TRACE_ID}-00f067aa0ba902b7-01"}) + lines = json_logs() + + events = [line for line in lines if line.get("event") == "http_request"] + assert len(events) == 1 + event = events[0] + assert event["method"] == "GET" + assert event["route"] == "/api/v1/diagnostics/" + assert event["status"] == 200 + assert event["duration_ms"] > 0 + assert event["db_queries"] > 0 + assert event["trace_id"] == TRACE_ID + assert all(line["level"] != "debug" for line in lines) + assert not any("query_stats" in line["message"] for line in lines) + + +def test_wide_event_names_the_route_when_no_endpoint_runs(client, json_logs): + client.post("/api/v1/diagnostics/") + + (event,) = [line for line in json_logs() if line.get("event") == "http_request"] + assert event["status"] == 405 + assert event["route"] == "/api/v1/diagnostics/" + + +def test_failed_request_keeps_its_query_statistics(json_logs): + engine = sqlalchemy.create_engine("sqlite://") + app = FastAPI() + + @app.get("/boom") + def boom(): + with engine.connect() as connection: + connection.execute(sqlalchemy.text("SELECT 1")) + raise RuntimeError("boom") + + app.add_middleware(RequestQueryMonitor, engine=engine, actions=[BindQueryStatistics()]) + app.add_middleware(WideEventMiddleware) + + TestClient(app, raise_server_exceptions=False).get("/boom") + + (event,) = [line for line in json_logs() if line.get("event") == "http_request"] + assert event["status"] == 500 + assert event["db_queries"] == 1 diff --git a/backend/uv.lock b/backend/uv.lock index 95ac593e..30f9a3a1 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -4123,6 +4123,7 @@ dependencies = [ { name = "httpx" }, { name = "loguru" }, { name = "opentelemetry-instrumentation-sqlalchemy" }, + { name = "prometheus-fastapi-instrumentator" }, { name = "psycopg", extra = ["binary"] }, { name = "pydantic" }, { name = "pydantic-settings" }, @@ -4153,6 +4154,7 @@ requires-dist = [ { name = "httpx", specifier = ">=0.27" }, { name = "loguru" }, { name = "opentelemetry-instrumentation-sqlalchemy", specifier = ">=0.51b0" }, + { name = "prometheus-fastapi-instrumentator", specifier = ">=7.0.0" }, { name = "psycopg", extras = ["binary"], specifier = ">=3.1.13,<4.0.0" }, { name = "pydantic", specifier = ">2.0" }, { name = "pydantic-settings", specifier = ">=2.13.1,<3.0.0" }, diff --git a/changelog/153.improvement.md b/changelog/153.improvement.md new file mode 100644 index 00000000..f6d84636 --- /dev/null +++ b/changelog/153.improvement.md @@ -0,0 +1 @@ +Added one structured log line per API request, with its route, status, duration and trace id, and made `LOG_LEVEL` apply to all API logs.