hold-slayer/core/logging_config.py

"""
Logging configuration — human-readable text or structured JSON.

Hold Slayer's logs are shipped to Loki by the host's Alloy agent, which reads
the container's stdout. Text logs arrive there as an opaque blob: filtering on
a status code or a call ID means regex over a formatted string. JSON lines
arrive as queryable fields.

Two things about this are less obvious than they look, and both are the reason
this module exists instead of a `format=` argument on `basicConfig`:

1. **Uvicorn brings its own handlers.** `uvicorn.config.LOGGING_CONFIG` attaches
   a `StreamHandler` to `uvicorn` and `uvicorn.access` with `propagate: False`,
   so those records never reach the root logger's formatter. Configuring only
   the root would leave the access log — the highest-volume, most useful stream
   — as plain colourised text next to our JSON. `configure_logging` reaches into
   those two loggers explicitly.

2. **The access record's payload is in `record.args`, not the message.** Uvicorn
   logs access lines as a 5-tuple `(client_addr, method, full_path, http_version,
   status_code)` and lets its `AccessFormatter` interpolate them. Formatting the
   message would throw that structure away and force Loki to parse it back out,
   so `JSONFormatter` unpacks the tuple into real fields.

Text mode stays the default: it is what a developer wants on a terminal, and a
JSON-only logger makes local debugging worse. Production opts in via
`LOG_FORMAT=json`.
"""

import datetime as _dt
import json
import logging
import sys
from typing import Any

# LogRecord attributes that are either already represented in our output or are
# formatting machinery. Anything on a record that is *not* here is treated as a
# caller-supplied `extra=` field and promoted into the JSON object.
_RESERVED = frozenset(
    {
        "args",
        "asctime",
        # Uvicorn passes an ANSI-colourised duplicate of the message as
        # `extra={"color_message": ...}` for its own formatter to prefer. Left
        # unfiltered, generic extra-promotion copies escape codes into every
        # startup line — the same unreadable-in-Grafana problem as the lab's
        # Asterisk logs.
        "color_message",
        "created",
        "exc_info",
        "exc_text",
        "filename",
        "funcName",
        "levelname",
        "levelno",
        "lineno",
        "module",
        "msecs",
        "message",
        "msg",
        "name",
        "pathname",
        "process",
        "processName",
        "relativeCreated",
        "stack_info",
        "taskName",
        "thread",
        "threadName",
    }
)

# Uvicorn's own access-log tuple, in order.
_ACCESS_FIELDS = ("client_addr", "method", "path", "http_version", "status_code")

TEXT_FORMAT = "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s"


class JSONFormatter(logging.Formatter):
    """Render a LogRecord as a single-line JSON object."""

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            # RFC 3339 in UTC. `logging`'s default asctime is local-time and
            # date-less, which makes correlating with Loki's own timestamps
            # unnecessarily hard.
            "ts": _dt.datetime.fromtimestamp(record.created, tz=_dt.UTC).isoformat(
                timespec="milliseconds"
            ),
            "level": record.levelname,
            "logger": record.name,
        }

        if record.name == "uvicorn.access" and isinstance(record.args, tuple):
            payload.update(_access_fields(record))
        else:
            payload["msg"] = record.getMessage()

        # Thread name matters here in a way it doesn't in a single-context app:
        # this process runs the asyncio loop, the Sippy ED thread, and PJSUA2
        # worker threads, and "which context logged this" is usually the first
        # question when debugging a call.
        if record.threadName and record.threadName != "MainThread":
            payload["thread"] = record.threadName

        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        if record.stack_info:
            payload["stack"] = self.formatStack(record.stack_info)

        for key, value in record.__dict__.items():
            if key not in _RESERVED and not key.startswith("_"):
                payload[key] = _safe(value)

        return json.dumps(payload, default=str, separators=(",", ":"))


def _access_fields(record: logging.LogRecord) -> dict[str, Any]:
    """Unpack uvicorn's access-log arg tuple into named fields.

    Falls back to the interpolated message if uvicorn ever changes the tuple's
    shape — a log line with a slightly wrong shape beats an exception inside the
    logging path taking out the request.
    """
    args = record.args
    if not isinstance(args, tuple) or len(args) != len(_ACCESS_FIELDS):
        return {"msg": record.getMessage()}

    fields: dict[str, Any] = dict(zip(_ACCESS_FIELDS, args))
    try:
        fields["status_code"] = int(fields["status_code"])
    except (TypeError, ValueError):
        pass
    return fields


def _safe(value: Any) -> Any:
    """Keep JSON-native types; stringify everything else."""
    if isinstance(value, (str, int, float, bool, type(None))):
        return value
    return str(value)


def configure_logging(log_format: str, log_level: str) -> None:
    """Install the root and uvicorn log handlers.

    Idempotent: existing root handlers are removed first, so calling this after
    uvicorn has configured itself replaces its formatting rather than adding a
    second stream (which is how you get every line twice).
    """
    level = getattr(logging, log_level.upper(), logging.INFO)
    use_json = log_format.lower() == "json"

    formatter: logging.Formatter = (
        JSONFormatter() if use_json else logging.Formatter(TEXT_FORMAT, datefmt="%H:%M:%S")
    )

    handler = logging.StreamHandler(stream=sys.stdout)
    handler.setFormatter(formatter)

    root = logging.getLogger()
    for existing in root.handlers[:]:
        root.removeHandler(existing)
    root.addHandler(handler)
    root.setLevel(level)

    # Uvicorn sets `propagate = False` on these and attaches its own colourised
    # handlers, so they must be redirected explicitly or they bypass everything
    # above. Clearing the handlers and re-enabling propagation routes them
    # through the root handler like any other logger.
    for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
        uv = logging.getLogger(name)
        uv.handlers.clear()
        uv.propagate = True
        uv.setLevel(level)