pixano_inference.observability

Observability: request-id propagation, structured logging, and Prometheus metrics.

Three concerns, all optional-at-runtime and cheap:

• Request id: :class:RequestContextMiddleware (a pure-ASGI middleware, so the id propagates via a ContextVar into endpoints and log records) tags every request with an X-Request-ID — echoed from the client's header or freshly generated — exposes it on request.state.request_id and the response header, and makes it available to logs.
• Logging: :func:configure_logging installs a dictConfig whose formatter includes the current request id (plain or JSON).
• Metrics: :class:PrometheusMiddleware records request count / latency / in-flight gauges; :func:render_metrics renders the exposition for a /metrics scrape endpoint. Ray Serve's own replica/queue metrics are exported separately by Serve.

JsonLogFormatter

Bases: Formatter

Minimal structured JSON log formatter carrying the request id.

format(record)

Serialize the record to a single-line JSON object.

Source code in pixano_inference/observability.py

def format(self, record: logging.LogRecord) -> str:
    """Serialize the record to a single-line JSON object."""
    payload = {
        "time": self.formatTime(record),
        "level": record.levelname,
        "logger": record.name,
        "message": record.getMessage(),
        "requestId": getattr(record, "request_id", "-"),
    }
    if record.exc_info:
        payload["exc"] = self.formatException(record.exc_info)
    return json.dumps(payload, default=str)

PrometheusMiddleware(app)

Record request count, latency, and in-flight gauge per method and route template.

The path label is the matched route template (e.g. /v1/inference/detection), not the raw URL, so cardinality stays bounded; unmatched requests are labelled unmatched.

Source code in pixano_inference/observability.py

def __init__(self, app: Any) -> None:
    """Wrap the downstream ASGI *app*."""
    self.app = app

__call__(scope, receive, send) async

Time the request and update the metrics, tolerating a missing prometheus_client.

Source code in pixano_inference/observability.py

async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
    """Time the request and update the metrics, tolerating a missing prometheus_client."""
    if scope["type"] != "http" or not _PROM_AVAILABLE:
        await self.app(scope, receive, send)
        return

    method = scope.get("method", "GET")
    status_holder = {"code": 500}

    async def send_wrapper(message: Any) -> None:
        if message["type"] == "http.response.start":
            status_holder["code"] = message["status"]
        await send(message)

    _IN_PROGRESS.labels(method).inc()
    start = time.perf_counter()
    try:
        await self.app(scope, receive, send_wrapper)
    finally:
        elapsed = time.perf_counter() - start
        _IN_PROGRESS.labels(method).dec()
        route = scope.get("route")
        path = getattr(route, "path", None) or "unmatched"
        _REQUESTS.labels(method, path, str(status_holder["code"])).inc()
        _LATENCY.labels(method, path).observe(elapsed)

RequestContextMiddleware(app)

Assign each HTTP request an id and expose it on state, logs, and the response header.

Source code in pixano_inference/observability.py

def __init__(self, app: Any) -> None:
    """Wrap the downstream ASGI *app*."""
    self.app = app

__call__(scope, receive, send) async

Set the request-id ContextVar and inject the X-Request-ID response header.

Source code in pixano_inference/observability.py

async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
    """Set the request-id ContextVar and inject the ``X-Request-ID`` response header."""
    if scope["type"] != "http":
        await self.app(scope, receive, send)
        return

    incoming = dict(scope.get("headers") or {}).get(b"x-request-id")
    request_id = incoming.decode("latin-1") if incoming else uuid.uuid4().hex

    # Starlette's request.state reads from scope["state"], so this reaches route handlers.
    scope.setdefault("state", {})
    scope["state"]["request_id"] = request_id
    token = _request_id_ctx.set(request_id)

    async def send_wrapper(message: Any) -> None:
        if message["type"] == "http.response.start":
            headers = message.setdefault("headers", [])
            headers.append((REQUEST_ID_HEADER.encode("latin-1"), request_id.encode("latin-1")))
        await send(message)

    try:
        await self.app(scope, receive, send_wrapper)
    finally:
        _request_id_ctx.reset(token)

RequestIdFilter

Bases: Filter

Inject the current request id onto every log record as request_id.

filter(record)

Attach record.request_id from the ContextVar.

Source code in pixano_inference/observability.py

def filter(self, record: logging.LogRecord) -> bool:
    """Attach ``record.request_id`` from the ContextVar."""
    record.request_id = _request_id_ctx.get()
    return True

configure_logging(level='INFO', json_logs=False)

Configure root logging with a request-id-aware formatter.

Parameters:

Name	Type	Description	Default
level	str	Root log level name (e.g. "INFO").	'INFO'
json_logs	bool	Emit structured JSON logs instead of the plain text format.	False

Source code in pixano_inference/observability.py

def configure_logging(level: str = "INFO", json_logs: bool = False) -> None:
    """Configure root logging with a request-id-aware formatter.

    Args:
        level: Root log level name (e.g. ``"INFO"``).
        json_logs: Emit structured JSON logs instead of the plain text format.
    """
    plain_format = "%(asctime)s %(levelname)s [%(request_id)s] %(name)s: %(message)s"
    formatter: dict[str, Any] = {"()": f"{__name__}.JsonLogFormatter"} if json_logs else {"format": plain_format}
    logging.config.dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "filters": {"request_id": {"()": f"{__name__}.RequestIdFilter"}},
            "formatters": {"default": formatter},
            "handlers": {
                "default": {
                    "class": "logging.StreamHandler",
                    "formatter": "default",
                    "filters": ["request_id"],
                }
            },
            "root": {"level": level.upper(), "handlers": ["default"]},
        }
    )

get_request_id()

Return the current request id, or "-" outside of a request.

Source code in pixano_inference/observability.py

def get_request_id() -> str:
    """Return the current request id, or ``"-"`` outside of a request."""
    return _request_id_ctx.get()

install_observability_middleware(app)

Add the metrics and request-id middleware to app (request-id outermost).

add_middleware stacks last-added outermost, so adding metrics first then the request-id middleware makes the request id available to the metrics layer, the error handlers, and the response headers for every request.

Source code in pixano_inference/observability.py

def install_observability_middleware(app: Any) -> None:
    """Add the metrics and request-id middleware to *app* (request-id outermost).

    ``add_middleware`` stacks last-added outermost, so adding metrics first then the request-id
    middleware makes the request id available to the metrics layer, the error handlers, and the
    response headers for every request.
    """
    app.add_middleware(PrometheusMiddleware)
    app.add_middleware(RequestContextMiddleware)

render_metrics()

Return the Prometheus exposition body and its content type for a /metrics route.

Source code in pixano_inference/observability.py

def render_metrics() -> tuple[bytes, str]:
    """Return the Prometheus exposition body and its content type for a ``/metrics`` route."""
    if not _PROM_AVAILABLE:  # pragma: no cover
        return b"", "text/plain; charset=utf-8"
    return generate_latest(), CONTENT_TYPE_LATEST