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


# Field names whose values never reach a log line. Keys are compared
# lowercased with "-" folded to "_", so header names (X-Api-Key) match too.
# Fragments match anywhere in the key (db_password, refresh_token); the exact
# names are too short or too specific to match as fragments ("pan" would
# catch "company").
_REDACT_FRAGMENTS = ("password", "passwd", "secret", "token", "authorization", "cookie", "api_key", "holder_name")
_REDACT_EXACT = {"pan", "card_number", "cvv", "cvc", "holder_name"}
_REDACTED = "[REDACTED]"


def _is_sensitive(key) -> bool:
    k = str(key).lower().replace("-", "_")
    return k in _REDACT_EXACT or any(f in k for f in _REDACT_FRAGMENTS)


def _scrub(value):
    if isinstance(value, dict):
        return {k: _REDACTED if _is_sensitive(k) else _scrub(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_scrub(v) for v in value]
    if isinstance(value, tuple):
        # Stays a tuple: format_exc_info only trusts an exc_info that is one.
        return tuple(_scrub(v) for v in value)
    return value


def _redact(logger, method_name, event_dict):
    """Mask denylisted fields, at any depth (a logged headers dict included).

    This works on field names only, so it is only as good as the call sites:
    a value interpolated into the `event` string is invisible to it.
    """
    return _scrub(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,
        _redact,
    ]

    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)