Spaces:
Sleeping
Sleeping
| """Structured log creation and trace/span correlation. | |
| The signal-responsibility table in the plan preamble gives logs one job: | |
| "Explain a discrete decision, failure, or fallback" -- e.g. "collection lookup | |
| failed before compatibility fallback returned []". This module produces those | |
| records through stdlib ``logging`` (no new framework), attaching correlation | |
| fields via the ``extra=`` mechanism so a record can be joined to its trace/span | |
| and carries operation, pipeline version, consumer, status, and error type. | |
| Field names are ``otel_``-prefixed to avoid colliding with stdlib | |
| ``LogRecord`` reserved attributes (``message``, ``name``, ``args`` ...). | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from typing import Any | |
| from opentelemetry import trace | |
| from app.observability.sanitize import sanitize_exception_text | |
| # The dedicated logger for observability-emitted correlation records. Product | |
| # code keeps using its own ``logging.getLogger(__name__)`` loggers; this one is | |
| # only for the boundary's own error/fallback explanations. | |
| LOGGER_NAME = "app.observability" | |
| _logger = logging.getLogger(LOGGER_NAME) | |
| _INVALID_TRACE_ID = 0 | |
| _INVALID_SPAN_ID = 0 | |
| def trace_context_ids(span: trace.Span | None) -> tuple[str | None, str | None]: | |
| """Return ``(trace_id_hex, span_id_hex)`` for a span, or ``(None, None)``. | |
| A non-recording / invalid span (disabled mode) has an all-zero context and | |
| yields ``None`` rather than a misleading zero string. | |
| """ | |
| if span is None: | |
| return None, None | |
| ctx = span.get_span_context() | |
| if ctx is None or ctx.trace_id == _INVALID_TRACE_ID: | |
| return None, None | |
| span_id = format(ctx.span_id, "016x") if ctx.span_id != _INVALID_SPAN_ID else None | |
| return format(ctx.trace_id, "032x"), span_id | |
| def build_log_extra( | |
| *, | |
| operation: str, | |
| status: str, | |
| trace_id: str | None = None, | |
| span_id: str | None = None, | |
| pipeline_version: str | None = None, | |
| consumer: str | None = None, | |
| subsystem: str | None = None, | |
| error_type: str | None = None, | |
| diagnostic_smoke: bool | None = None, | |
| ) -> dict[str, Any]: | |
| """Assemble the correlation ``extra`` dict for a structured log record.""" | |
| extra = { | |
| "otel_operation": operation, | |
| "otel_status": status, | |
| "otel_trace_id": trace_id, | |
| "otel_span_id": span_id, | |
| "otel_pipeline_version": pipeline_version, | |
| "otel_consumer": consumer, | |
| "otel_subsystem": subsystem, | |
| "otel_error_type": error_type, | |
| } | |
| # Keep normal product records unchanged/absent. Only the explicit bounded | |
| # smoke marker is allowed through; this is not a general attributes escape | |
| # hatch for log records. | |
| if diagnostic_smoke is not None: | |
| extra["diagnostic.smoke"] = bool(diagnostic_smoke) | |
| return extra | |
| def emit_correlated_log( | |
| message: str, | |
| *, | |
| operation: str, | |
| status: str, | |
| level: int = logging.ERROR, | |
| trace_id: str | None = None, | |
| span_id: str | None = None, | |
| pipeline_version: str | None = None, | |
| consumer: str | None = None, | |
| subsystem: str | None = None, | |
| error_type: str | None = None, | |
| diagnostic_smoke: bool | None = None, | |
| logger: logging.Logger | None = None, | |
| ) -> None: | |
| """Emit one structured, trace-correlated log record. | |
| ``message`` is sanitized (paths stripped, bounded) so an error string that | |
| embeds a filesystem path never reaches an exported log body. | |
| """ | |
| log = logger or _logger | |
| log.log( | |
| level, | |
| sanitize_exception_text(str(message)), | |
| extra=build_log_extra( | |
| operation=operation, | |
| status=status, | |
| trace_id=trace_id, | |
| span_id=span_id, | |
| pipeline_version=pipeline_version, | |
| consumer=consumer, | |
| subsystem=subsystem, | |
| error_type=error_type, | |
| diagnostic_smoke=diagnostic_smoke, | |
| ), | |
| ) | |