LedgerLine/common/logging_config.py

import logging
import sys

import structlog
from opentelemetry import trace

# Loggers that install their own handlers (uvicorn's dictConfig runs before the
# app is imported). Their handlers are stripped so records propagate to the
# root handler below and come out as JSON like everything else.
# uvicorn.access is left alone on purpose: --no-access-log works by setting its
# propagate=False, and re-enabling propagation would turn access logs back on.
_FOREIGN_LOGGERS = ("uvicorn", "uvicorn.error")


def _add_service(service_name: str):
    def processor(logger, method_name, event_dict):
        event_dict["service"] = service_name
        return event_dict

    return processor


def _add_trace_context(logger, method_name, event_dict):
    """Stamp the active span's IDs on the record, as lowercase hex.

    The name `trace_id` is load-bearing: Grafana's Loki derived field matches
    the case-sensitive regex trace[_-]?id["=: ]+([a-fA-F0-9]+), so OTel's own
    `otelTraceID` would never become a link. Outside a span (startup, the
    consumers' idle loop) the context is invalid and nothing is added, rather
    than an all-zeros ID that points at no trace.
    """
    ctx = trace.get_current_span().get_span_context()
    if ctx.is_valid:
        event_dict["trace_id"] = trace.format_trace_id(ctx.trace_id)
        event_dict["span_id"] = trace.format_span_id(ctx.span_id)
    return event_dict


def configure_logging(service_name: str) -> structlog.stdlib.BoundLogger:
    """One JSON object per line on stdout, for structlog and stdlib loggers alike.

    `shared` runs for every record, whichever API emitted it: structlog calls
    pass through it via structlog.configure(), stdlib records (uvicorn, the
    consumers' logging.getLogger(...)) via the formatter's foreign_pre_chain.
    """
    shared = [
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        _add_service(service_name),
        _add_trace_context,
    ]

    structlog.configure(
        processors=[*shared, structlog.stdlib.ProcessorFormatter.wrap_for_formatter],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        foreign_pre_chain=shared,
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),
        ],
    )
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)

    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(logging.INFO)

    for name in _FOREIGN_LOGGERS:
        foreign = logging.getLogger(name)
        foreign.handlers = []
        foreign.propagate = True

    return structlog.get_logger(service_name)