"""Logs, metrics and request context — what an operator needs to answer "what happened to this request / job, and is the system healthy?". Request context: every request gets an id — the caller's X-Request-ID if it sent a sane one (a proxy's, so the two logs line up), else a new one — returned on the response and carried by every log line written while the request runs, together with the tenant and user once the request has authenticated (get_auth_context binds them). A job run carries its job id and kind the same way. An unhandled error is logged with its traceback and answered with a 500 that names the request id, so a person can quote it. Logs: LOG_FORMAT=json writes one JSON object per line (for a log pipeline); "text" (the default) stays readable in a terminal. One access line per request: method, route, status, duration. Metrics (Prometheus, GET /metrics): requests by route and status and their latency; job runs by kind and outcome and their duration; and, read from the database at scrape time, jobs by queue and status and the age of the oldest due job. /metrics needs `Authorization: Bearer ` when METRICS_TOKEN is set, and is off in production until it is. """ import contextvars import json import logging import re import sys import time import uuid from datetime import datetime, timezone from typing import Any from prometheus_client import CONTENT_TYPE_LATEST, CollectorRegistry, Counter, Gauge, Histogram, generate_latest from app.core.config import settings # Fields every log line written in this context carries. A dict, so that a # dependency running deeper in the request (get_auth_context) can add to # the one the middleware set up and the middleware's own access line sees it. _context: contextvars.ContextVar[dict[str, Any] | None] = contextvars.ContextVar("ff_log_context", default=None) # The browser a request came from — where a sign-in is recorded from (app/crud/sessions.py). _client: contextvars.ContextVar[tuple[str | None, str | None]] = contextvars.ContextVar("ff_client", default=(None, None)) def client_info() -> tuple[str | None, str | None]: """(IP address, User-Agent) of the current request.""" return _client.get() _REQUEST_ID = re.compile(r"^[A-Za-z0-9._:-]{1,64}$") _QUIET_PATHS = {"/health", "/ready", "/metrics"} def bind(**fields: Any) -> None: """Adds fields (tenant_id, user_id, …) to the current request's or job's log context.""" ctx = _context.get() if ctx is not None: ctx.update({k: str(v) for k, v in fields.items() if v is not None}) def current(field: str) -> str | None: ctx = _context.get() return ctx.get(field) if ctx else None class context: """`with observability.context(job_id=…, job_kind=…):` — a log context for work outside a request.""" def __init__(self, **fields: Any): self._fields = {k: str(v) for k, v in fields.items() if v is not None} def __enter__(self) -> dict[str, Any]: self._token = _context.set(dict(self._fields)) return _context.get() def __exit__(self, *exc: Any) -> None: _context.reset(self._token) # ---- logging ------------------------------------------------------------------------------------------------ class _ContextFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: ctx = _context.get() or {} record.ff_context = ctx record.request_id = ctx.get("request_id", "-") return True class JsonFormatter(logging.Formatter): _SKIP = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"ff_context", "request_id", "message", "asctime"} def format(self, record: logging.LogRecord) -> str: out: dict[str, Any] = { "time": datetime.fromtimestamp(record.created, timezone.utc).isoformat(timespec="milliseconds"), "level": record.levelname, "logger": record.name, "message": record.getMessage(), **getattr(record, "ff_context", {}), } # Fields passed with logger.info(..., extra={...}). out.update({k: v for k, v in record.__dict__.items() if k not in self._SKIP and not k.startswith("_")}) trace_id = _trace_id() if trace_id: out["trace_id"] = trace_id if record.exc_info: out["exception"] = self.formatException(record.exc_info) return json.dumps(out, default=str) def _trace_id() -> str | None: try: from opentelemetry import trace except ImportError: return None span = trace.get_current_span().get_span_context() return format(span.trace_id, "032x") if span.is_valid else None _configured = False def configure_logging() -> None: """Root logging for the API and the worker, per LOG_FORMAT / LOG_LEVEL.""" global _configured if _configured: return _configured = True handler = logging.StreamHandler(sys.stdout) handler.addFilter(_ContextFilter()) if settings.log_format == "json": handler.setFormatter(JsonFormatter()) else: handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)-7s %(name)s [%(request_id)s] %(message)s")) root = logging.getLogger() root.handlers = [handler] root.setLevel(settings.log_level.upper()) # Uvicorn's own access log repeats ours without the context. logging.getLogger("uvicorn.access").disabled = True # Every outbound call at INFO is noise; failures still surface. logging.getLogger("httpx").setLevel(logging.WARNING) for name in ("uvicorn", "uvicorn.error"): logging.getLogger(name).handlers = [] logging.getLogger(name).propagate = True # ---- metrics ------------------------------------------------------------------------------------------------ REGISTRY = CollectorRegistry() HTTP_REQUESTS = Counter("flowfabric_http_requests", "HTTP requests handled", ["method", "route", "status"], registry=REGISTRY) HTTP_SECONDS = Histogram( "flowfabric_http_request_duration_seconds", "HTTP request duration", ["method", "route"], registry=REGISTRY, buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30), ) HTTP_IN_FLIGHT = Gauge("flowfabric_http_requests_in_flight", "HTTP requests being handled", registry=REGISTRY) JOB_RUNS = Counter("flowfabric_job_runs", "Job attempts finished, by outcome (done, retry, failed)", ["kind", "outcome"], registry=REGISTRY) JOB_SECONDS = Histogram( "flowfabric_job_duration_seconds", "Job attempt duration", ["kind"], registry=REGISTRY, buckets=(0.05, 0.1, 0.5, 1, 5, 15, 60, 300, 900, 3600), ) JOBS = Gauge("flowfabric_jobs", "Jobs in the queue table, by queue and status", ["queue", "status"], registry=REGISTRY) JOB_OLDEST_DUE = Gauge("flowfabric_job_oldest_due_seconds", "How long the oldest due job has been waiting", ["queue"], registry=REGISTRY) async def _refresh_job_gauges() -> None: from sqlalchemy import text from app.core.database import AsyncSessionLocal async with AsyncSessionLocal() as db: rows = (await db.execute(text("SELECT queue, status, count(*) FROM jobs GROUP BY 1, 2"))).all() oldest = ( await db.execute( text("SELECT queue, EXTRACT(EPOCH FROM now() - min(run_after)) FROM jobs WHERE status = 'queued' AND run_after <= now() GROUP BY 1") ) ).all() JOBS.clear() for queue, status, n in rows: JOBS.labels(queue, status).set(n) JOB_OLDEST_DUE.clear() for queue, seconds in oldest: JOB_OLDEST_DUE.labels(queue).set(float(seconds or 0)) async def metrics_text() -> tuple[bytes, str]: try: await _refresh_job_gauges() except Exception: # noqa: BLE001 - a down database still leaves the request metrics worth scraping logging.getLogger("flowfabric").warning("Couldn't read job counts for /metrics", exc_info=True) return generate_latest(REGISTRY), CONTENT_TYPE_LATEST def metrics_allowed(authorization: str | None) -> bool: if settings.metrics_token: return authorization == f"Bearer {settings.metrics_token}" return settings.environment != "production" # ---- the request middleware ------------------------------------------------------------------------------- def _route_of(scope: dict) -> str: """The route's template (/masters/{master_id}) rather than the path — so metrics don't get a label per record.""" route = scope.get("route") return getattr(route, "path", None) or "unmatched" class RequestContextMiddleware: """Pure ASGI (not BaseHTTPMiddleware): the request runs in this same task, so the context set here is the one every log line inside sees.""" def __init__(self, app): self.app = app self.log = logging.getLogger("flowfabric.access") async def __call__(self, scope, receive, send): if scope["type"] != "http": return await self.app(scope, receive, send) incoming = next((v.decode("latin-1") for k, v in scope.get("headers", []) if k == b"x-request-id"), "") request_id = incoming if _REQUEST_ID.match(incoming) else uuid.uuid4().hex token = _context.set({"request_id": request_id}) agent = next((v.decode("latin-1") for k, v in scope.get("headers", []) if k == b"user-agent"), None) client_token = _client.set(((scope.get("client") or (None,))[0], agent)) status = 500 started = time.perf_counter() response_started = False async def send_with_id(message): nonlocal status, response_started if message["type"] == "http.response.start": response_started = True status = message["status"] message.setdefault("headers", []) message["headers"] = [*message["headers"], (b"x-request-id", request_id.encode())] await send(message) HTTP_IN_FLIGHT.inc() try: await self.app(scope, receive, send_with_id) except Exception: logging.getLogger("flowfabric").exception("Unhandled error on %s %s", scope["method"], scope["path"]) if response_started: raise status = 500 body = json.dumps({"detail": f"Something went wrong on our side. Reference: {request_id}", "request_id": request_id}).encode() await send({ "type": "http.response.start", "status": 500, "headers": [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode()), (b"x-request-id", request_id.encode())], }) await send({"type": "http.response.body", "body": body}) finally: HTTP_IN_FLIGHT.dec() elapsed = time.perf_counter() - started route, method = _route_of(scope), scope["method"] HTTP_REQUESTS.labels(method, route, str(status)).inc() HTTP_SECONDS.labels(method, route).observe(elapsed) level = logging.DEBUG if scope["path"] in _QUIET_PATHS else logging.WARNING if status >= 500 else logging.INFO self.log.log( level, "%s %s %s %.0fms", method, scope["path"], status, elapsed * 1000, extra={"method": method, "route": route, "status": status, "duration_ms": round(elapsed * 1000, 1)}, ) _context.reset(token) _client.reset(client_token)