File size: 6,770 Bytes
28c12aa 700e896 28c12aa 700e896 28c12aa d40bd07 28c12aa d40bd07 28c12aa d40bd07 28c12aa d40bd07 | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """Structured logging: colored console renderer + the `step()` stage timer.
Logging standard for this project: every pipeline stage runs inside ``with step("<concern>.<action>") as
metrics:`` — it logs ``step.start`` then ``step.end`` with the elapsed seconds and whatever you attach via
``metrics["key"] = value``. No ``print()`` in the engine; no engine/library names in user-facing strings.
Configuration is read from the environment (so this module has no dependency on the settings layer yet):
``RERANKER_LOG_LEVEL`` (default ``INFO``), ``RERANKER_LOG`` (set ``0`` to silence), ``NO_COLOR`` to drop
ANSI. The stdlib ``import logging`` below is an ABSOLUTE import — Python resolves top-level imports against
``sys.path``, never this same-named sibling — so ``logging.INFO`` references the standard library correctly.
"""
from __future__ import annotations
import contextlib
import logging
import os
import sys
import time
from contextlib import contextmanager
from typing import Any
import structlog
from structlog.typing import EventDict
_log_configured = False
# this module's own functions to skip when locating the real caller (logging plumbing, not user code)
_CALLSITE_SKIP = {"step", "get_logger", "_callsite", "_configure_logging"}
_ANSI = {
"reset": "\033[0m", "bold": "\033[1m", "gray": "\033[90m", "red": "\033[31m",
"green": "\033[32m", "yellow": "\033[33m", "blue": "\033[94m", "magenta": "\033[35m", "cyan": "\033[36m",
}
_LEVEL_COLOR = {"debug": "blue", "info": "green", "warning": "yellow", "error": "red", "critical": "red"}
def _callsite(_logger: Any, _method_name: str, event_dict: EventDict) -> EventDict:
"""Add ``at`` (file:line) + ``func`` to each line, pointing at the code that opened the step."""
frame: Any = sys._getframe()
while frame is not None:
filename = frame.f_code.co_filename
func = frame.f_code.co_name
skip = (
"structlog" in filename
or "contextlib" in filename
or (os.path.basename(filename) == "logging.py" and func in _CALLSITE_SKIP)
)
if not skip:
event_dict["at"] = f"{os.path.basename(filename)}:{frame.f_lineno}"
event_dict["func"] = func
break
frame = frame.f_back
return event_dict
def _make_console_renderer(colors: bool):
"""Build a renderer: ``time [level] [step] [file:line func] event key=value …``."""
def paint(text: str, *styles: str) -> str:
if not colors:
return text
prefix = "".join(_ANSI[style] for style in styles)
return f"{prefix}{text}{_ANSI['reset']}"
def render(_logger: Any, _method: str, event_dict: EventDict) -> str:
extra = dict(event_dict)
timestamp = str(extra.pop("timestamp", ""))
level = str(extra.pop("level", "info"))
event = str(extra.pop("event", ""))
name = extra.pop("step", None) or extra.pop("logger", None)
extra.pop("logger", None)
location = f"{extra.pop('at', '')} {extra.pop('func', '')}".strip()
blocks = [paint(timestamp, "gray"), paint(f"[{level:<7}]", "bold", _LEVEL_COLOR.get(level, "green"))]
if name:
blocks.append(paint(f"[{name}]", "cyan"))
if location:
blocks.append(paint(f"[{location}]", "yellow"))
blocks.append(paint(event, "bold", "blue"))
header_line = " ".join(block for block in blocks if block)
pairs = " ".join(
f"{paint(key, 'magenta')}={paint(str(value), 'green')}" for key, value in extra.items()
)
return f"{header_line} {pairs}".rstrip()
return render
def _configure_logging(force: bool = False) -> None:
"""Configure structlog once from the environment (idempotent)."""
global _log_configured
if _log_configured and not force:
return
if os.environ.get("RERANKER_LOG", "1") == "0":
# Silence: filter at CRITICAL so info/warning/error are dropped (the engine never logs CRITICAL).
# NOTE: structlog only accepts standard levels here — CRITICAL+1 raises KeyError.
structlog.configure(
wrapper_class=structlog.make_filtering_bound_logger(logging.CRITICAL),
cache_logger_on_first_use=True,
)
_log_configured = True
return
with contextlib.suppress(Exception):
sys.stdout.reconfigure(encoding="utf-8", line_buffering=True) # type: ignore[union-attr]
colors = "NO_COLOR" not in os.environ
if colors:
with contextlib.suppress(Exception): # wrap stdout so ANSI works on legacy Windows consoles
import colorama
colorama.init()
level = getattr(logging, os.environ.get("RERANKER_LOG_LEVEL", "INFO").upper(), logging.INFO)
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="%H:%M:%S", utc=False),
_callsite,
_make_console_renderer(colors),
],
wrapper_class=structlog.make_filtering_bound_logger(level),
cache_logger_on_first_use=True,
)
_log_configured = True
def get_logger(name: str | None = None):
"""Return a configured structlog logger, with the concern ``name`` bound as ``logger=``."""
_configure_logging()
log = structlog.get_logger(name)
return log.bind(logger=name) if name else log
def _phase_banner(text: str) -> None:
"""A plain full-width divider line bracketing every phase, for grep/scroll tracing in a long log stream
(``grep -A20 "PHASE START: rank.explain"``). Bypasses the structured renderer on purpose — this is a visual
marker, not a parseable event — but still respects ``RERANKER_LOG=0`` silencing."""
if os.environ.get("RERANKER_LOG", "1") == "0":
return
print(f"--------------------------- {text} ---------------------------")
@contextmanager
def step(label: str, **context: Any):
"""Log ``step.start`` / ``step.end`` (+seconds, +metrics), bracketed by PHASE banners. Yields a mutable
metrics dict."""
_phase_banner(f"PHASE START: {label}")
log = get_logger("step").bind(step=label, **context)
log.info("step.start")
metrics: dict[str, Any] = {}
start = time.perf_counter()
try:
yield metrics
except Exception as exc:
log.error("step.fail", seconds=round(time.perf_counter() - start, 3), error=str(exc))
_phase_banner(f"PHASE FAILED: {label} ({round(time.perf_counter() - start, 3)}s)")
raise
else:
log.info("step.end", seconds=round(time.perf_counter() - start, 3), **metrics)
_phase_banner(f"PHASE END: {label} ({round(time.perf_counter() - start, 3)}s)")
|