perf(logging): redact and sample access logs

.env.example

@@ -103,6 +103,11 @@ UVICORN_PORT = 8000
 # LOG_MAX_BYTES = 10485760
 # LOG_LEVEL = "INFO"
 
+## Successful requests to high-volume routes are sampled; errors and slow requests are always logged.
+# ACCESS_LOG_SUCCESS_SAMPLE_RATE = 0.01
+# ACCESS_LOG_SLOW_MS = 1000
+# ACCESS_LOG_SAMPLED_ROUTES = "/api/user/{username},/api/user/by-username/{username},/api/user/by-id/{user_id}"
+
 ## JWT access token lifetime in minutes.
 # JWT_ACCESS_TOKEN_EXPIRE_MINUTES = 1440
 

app/app_factory.py

@@ -8,7 +8,7 @@
 from sqlalchemy.exc import DBAPIError
 
 from app.lifecycle import on_shutdown, on_startup
-from app.middlewares import setup_middleware
+from app.middlewares import safe_request_target, setup_middleware
 from app.nats import is_multi_worker, require_nats_if_multiworker
 from app.nats.message import MessageTopic
 from app.nats.router import router
@@ -28,7 +28,8 @@ async def _ignore_worker_sync_message(_: dict):
 async def database_operational_error_handler(request: Request, exc: DBAPIError):
     orig = getattr(exc, "orig", None)
     error_summary = f"{type(orig).__name__}: {orig}" if orig else type(exc).__name__
-    logger.warning(f"Database unavailable while handling {request.method} {request.url.path}: {error_summary}")
+    _, request_target = safe_request_target(request.scope)
+    logger.warning(f"Database unavailable while handling {request.method} {request_target}: {error_summary}")
     return JSONResponse(
         status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
         content={"detail": "Database temporarily unavailable"},

app/middlewares/__init__.py

@@ -3,9 +3,9 @@
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 
 from app.utils.logger import get_logger
-from config import cors_settings, server_settings
+from config import cors_settings, logging_settings, server_settings
 
-from .request_logging import RequestProcessTimeLoggingMiddleware
+from .request_logging import RequestProcessTimeLoggingMiddleware, safe_request_target as safe_request_target
 
 
 def setup_middleware(app: FastAPI):
@@ -18,4 +18,10 @@ def setup_middleware(app: FastAPI):
     )
     if server_settings.proxy_headers:
         app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=server_settings.forwarded_allow_ips)
-    app.add_middleware(RequestProcessTimeLoggingMiddleware, access_logger=get_logger("uvicorn.access"))
+    app.add_middleware(
+        RequestProcessTimeLoggingMiddleware,
+        access_logger=get_logger("uvicorn.access"),
+        success_sample_rate=logging_settings.access_log_success_sample_rate,
+        slow_request_ms=logging_settings.access_log_slow_ms,
+        sampled_routes=logging_settings.access_log_sampled_routes,
+    )

app/middlewares/request_logging.py

@@ -1,14 +1,44 @@
 import logging
+import random
 from time import perf_counter
 
 from h11 import LocalProtocolError
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 
+def safe_request_target(scope: Scope) -> tuple[str, str]:
+    """Return a route template without path parameters or query contents."""
+    route = scope.get("route")
+    route_path = getattr(route, "path", None)
+    if not isinstance(route_path, str) or not route_path:
+        route_path = "/<unmatched>"
+
+    query_bytes = scope.get("query_string", b"")
+    request_target = f"{route_path}?<redacted>" if query_bytes else route_path
+    return route_path, request_target
+
+
 class RequestProcessTimeLoggingMiddleware:
-    def __init__(self, app: ASGIApp, access_logger: logging.Logger):
+    def __init__(
+        self,
+        app: ASGIApp,
+        access_logger: logging.Logger,
+        success_sample_rate: float = 1.0,
+        slow_request_ms: float = 1000,
+        sampled_routes: frozenset[str] = frozenset(),
+    ):
         self.app = app
         self.access_logger = access_logger
+        self.success_sample_rate = min(max(success_sample_rate, 0), 1)
+        self.slow_request_ms = max(slow_request_ms, 0)
+        self.sampled_routes = sampled_routes
+
+    def _log_level(self, route_path: str, status_code: int, process_time_ms: float) -> int | None:
+        if status_code >= 400 or process_time_ms >= self.slow_request_ms:
+            return logging.INFO
+        if route_path not in self.sampled_routes:
+            return logging.INFO
+        return logging.INFO if random.random() < self.success_sample_rate else None
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope["type"] != "http":
@@ -43,21 +73,21 @@ async def send_wrapper(message: Message) -> None:
                 raise
         finally:
             process_time_ms = (perf_counter() - start_time) * 1000
-            path = scope.get("path", "")
-            query_bytes = scope.get("query_string", b"")
-            if query_bytes:
-                path = f"{path}?{query_bytes.decode(errors='replace')}"
+            route_path, request_target = safe_request_target(scope)
             http_version = scope.get("http_version", "1.1")
             client = scope.get("client")
             client_addr = client[0] if client else "-"
             method = scope.get("method", "-")
+            log_level = self._log_level(route_path, status_code, process_time_ms)
 
-            self.access_logger.info(
-                '%s - "%s %s HTTP/%s" %d',
-                client_addr,
-                method,
-                path,
-                http_version,
-                status_code,
-                extra={"process_time": f"{process_time_ms:.2f}ms"},
-            )
+            if log_level is not None:
+                self.access_logger.log(
+                    log_level,
+                    '%s - "%s %s HTTP/%s" %d',
+                    client_addr,
+                    method,
+                    request_target,
+                    http_version,
+                    status_code,
+                    extra={"process_time": f"{process_time_ms:.2f}ms"},
+                )

config.py

@@ -159,13 +159,32 @@ class LoggingSettings(EnvSettings):
     rotation_unit: str = Field(default="H", validation_alias="LOG_ROTATION_UNIT")
     max_bytes: int = Field(default=10485760, validation_alias="LOG_MAX_BYTES")
     level: str = Field(default="INFO", validation_alias="LOG_LEVEL")
+    access_log_success_sample_rate: float = Field(
+        default=0.01,
+        ge=0,
+        le=1,
+        validation_alias="ACCESS_LOG_SUCCESS_SAMPLE_RATE",
+    )
+    access_log_slow_ms: float = Field(default=1000, ge=0, validation_alias="ACCESS_LOG_SLOW_MS")
+    access_log_sampled_routes_raw: str = Field(
+        default=(
+            "/api/user/{username},"
+            "/api/user/by-username/{username},"
+            "/api/user/by-id/{user_id}"
+        ),
+        validation_alias="ACCESS_LOG_SAMPLED_ROUTES",
+    )
 
     @field_validator("level")
     @classmethod
     def normalize_level(cls, value: str) -> str:
         value = value.upper()
         return value if value in ("CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG") else "INFO"
 
+    @property
+    def access_log_sampled_routes(self) -> frozenset[str]:
+        return frozenset(route.strip() for route in self.access_log_sampled_routes_raw.split(",") if route.strip())
+
 
 class AuthSettings(EnvSettings):
     sudo_username: str = Field(default="", validation_alias="SUDO_USERNAME")

tests/test_app_factory.py

@@ -1,9 +1,12 @@
 import json
+from types import SimpleNamespace
+from unittest.mock import Mock
 
 import pytest
 from sqlalchemy.exc import DBAPIError, OperationalError
 from starlette.requests import Request
 
+from app import app_factory
 from app.app_factory import database_operational_error_handler
 
 
@@ -27,3 +30,26 @@ async def test_database_operational_error_handler_handles_dbapi_errors():
 
     assert response.status_code == 503
     assert json.loads(response.body) == {"detail": "Database temporarily unavailable"}
+
+
+@pytest.mark.asyncio
+async def test_database_operational_error_log_redacts_route_values(monkeypatch: pytest.MonkeyPatch):
+    request = Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "path": "/sub/secret-token",
+            "query_string": b"username=alice",
+            "headers": [],
+            "route": SimpleNamespace(path="/sub/{token}"),
+        }
+    )
+    warning = Mock()
+    monkeypatch.setattr(app_factory.logger, "warning", warning)
+
+    await database_operational_error_handler(request, OperationalError(None, None, Exception("connection failed")))
+
+    message = warning.call_args.args[0]
+    assert "/sub/{token}?<redacted>" in message
+    assert "secret-token" not in message
+    assert "alice" not in message

tests/test_request_logging.py

@@ -0,0 +1,131 @@
+import logging
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from app.middlewares import request_logging
+from app.middlewares.request_logging import RequestProcessTimeLoggingMiddleware
+
+
+def _scope(path: str, query: bytes = b"") -> dict:
+    return {
+        "type": "http",
+        "method": "GET",
+        "path": path,
+        "query_string": query,
+        "http_version": "1.1",
+        "client": ("203.0.113.1", 1234),
+    }
+
+
+async def _call_middleware(
+    scope: dict,
+    *,
+    route_path: str | None,
+    status_code: int = 200,
+    sample_rate: float = 1,
+    sampled_routes: frozenset[str] = frozenset(),
+) -> Mock:
+    async def app(inner_scope, receive, send):
+        if route_path is not None:
+            inner_scope["route"] = SimpleNamespace(path=route_path)
+        await send({"type": "http.response.start", "status": status_code, "headers": []})
+        await send({"type": "http.response.body", "body": b""})
+
+    async def receive():
+        return {"type": "http.request", "body": b""}
+
+    async def send(_):
+        return None
+
+    logger = Mock()
+    middleware = RequestProcessTimeLoggingMiddleware(
+        app,
+        logger,
+        success_sample_rate=sample_rate,
+        slow_request_ms=1000,
+        sampled_routes=sampled_routes,
+    )
+    await middleware(scope, receive, send)
+    return logger
+
+
+@pytest.mark.asyncio
+async def test_access_log_uses_route_template_and_redacts_query_values():
+    logger = await _call_middleware(
+        _scope("/sub/secret-token", b"usernames=alice&limit=100"),
+        route_path="/sub/{token}",
+    )
+
+    args = logger.log.call_args.args
+    assert args[0] == logging.INFO
+    assert args[4] == "/sub/{token}?<redacted>"
+    assert "secret-token" not in args
+    assert "alice" not in args
+
+
+@pytest.mark.asyncio
+async def test_unmatched_access_log_never_uses_raw_path():
+    logger = await _call_middleware(_scope("/secret/unknown-token"), route_path=None, status_code=404)
+
+    args = logger.log.call_args.args
+    assert args[0] == logging.INFO
+    assert args[4] == "/<unmatched>"
+    assert "unknown-token" not in args
+
+
+@pytest.mark.asyncio
+async def test_frequent_success_is_suppressed_when_not_sampled():
+    route = "/api/user/{username}"
+    logger = await _call_middleware(
+        _scope("/api/user/alice"),
+        route_path=route,
+        sample_rate=0,
+        sampled_routes=frozenset({route}),
+    )
+
+    logger.log.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_frequent_success_is_info_when_sampled():
+    route = "/api/user/{username}"
+    logger = await _call_middleware(
+        _scope("/api/user/alice"),
+        route_path=route,
+        sample_rate=1,
+        sampled_routes=frozenset({route}),
+    )
+
+    assert logger.log.call_args.args[0] == logging.INFO
+
+
+@pytest.mark.asyncio
+async def test_error_on_frequent_route_is_always_info():
+    route = "/api/user/{username}"
+    logger = await _call_middleware(
+        _scope("/api/user/alice"),
+        route_path=route,
+        status_code=500,
+        sample_rate=0,
+        sampled_routes=frozenset({route}),
+    )
+
+    assert logger.log.call_args.args[0] == logging.INFO
+
+
+@pytest.mark.asyncio
+async def test_slow_success_on_frequent_route_is_always_info(monkeypatch: pytest.MonkeyPatch):
+    times = iter((1.0, 2.1))
+    monkeypatch.setattr(request_logging, "perf_counter", lambda: next(times))
+    route = "/api/user/{username}"
+
+    logger = await _call_middleware(
+        _scope("/api/user/alice"),
+        route_path=route,
+        sample_rate=0,
+        sampled_routes=frozenset({route}),
+    )
+
+    assert logger.log.call_args.args[0] == logging.INFO