Spaces:
Sleeping
Sleeping
File size: 3,104 Bytes
830d137 feb1b1c 830d137 feb1b1c 830d137 feb1b1c 830d137 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
from __future__ import annotations
import logging
import sys
from collections.abc import Mapping
from typing import Final, TextIO
__all__: tuple[str, ...] = (
"get_logger",
"stage_metric",
)
_REPRO_SAFE_FORMAT: Final[str] = "%(levelname)s %(name)s %(message)s"
class _DeterministicFormatter(logging.Formatter):
"""A ``Formatter`` that never renders wall-clock time into a record."""
def formatTime( # noqa: N802 — overrides logging.Formatter's mixed-case API.
self, record: logging.LogRecord, datefmt: str | None = None
) -> str:
"""Return an empty string; repro-relevant lines carry no timestamp."""
return ""
class _DeterministicStreamHandler(logging.StreamHandler[TextIO]):
"""Marker subclass so :func:`get_logger` can detect prior setup idempotently."""
def get_logger(name: str, *, level: int = logging.INFO) -> logging.Logger:
"""Return a stdlib logger configured for deterministic, timestamp-free output.
Idempotent: calling this repeatedly for the same ``name`` reuses the
already-attached handler instead of stacking duplicate handlers, and
disables propagation to the root logger so output is not duplicated by a
caller's own root configuration.
Args:
name: The logger name, conventionally the calling module's stage id
(e.g. ``"R3"`` or ``"redstack.engines.scoring"``).
level: The minimum level this logger emits at.
Returns:
A configured :class:`logging.Logger`.
"""
logger = logging.getLogger(name)
logger.setLevel(level)
if not any(isinstance(h, _DeterministicStreamHandler) for h in logger.handlers):
handler = _DeterministicStreamHandler(sys.stderr)
handler.setFormatter(_DeterministicFormatter(_REPRO_SAFE_FORMAT))
logger.addHandler(handler)
logger.propagate = False
return logger
def _render_fields(fields: Mapping[str, object]) -> str:
"""Render structured fields as ``key=value`` pairs in ascending key order.
Sorting by key — rather than trusting call-site or dict insertion order —
is what makes the rendered line stable across runs and Python versions.
"""
return " ".join(f"{key}={fields[key]!r}" for key in sorted(fields))
def stage_metric(
logger: logging.Logger, stage: str, event: str, **fields: object
) -> None:
"""Emit one deterministic structured record: ``[stage] event key=value ...``.
This is the per-stage metric capture surface: stages report facts
(``stage_metric(log, "R6", "ranked", honeypot_count=0, size=100)``) rather
than hand-formatting strings, so every emitted line has a stable shape.
Args:
logger: A logger obtained from :func:`get_logger`.
stage: The stage identifier the metric belongs to (e.g. ``"R6"``).
event: A short, stable event name (e.g. ``"ranked"``, ``"started"``).
**fields: Arbitrary structured payload; rendered sorted by key.
"""
rendered = _render_fields(fields)
message = f"[{stage}] {event}" + (f" {rendered}" if rendered else "")
logger.info(message)
|