"""Structured logging configuration using structlog. This module provides centralized logging configuration for the entire application. It supports both development (colored, readable) and production (JSON) output formats. Usage: from mcp_hangar.logging_config import setup_logging, get_logger # At application startup setup_logging(level="INFO", json_format=True) # In any module logger = get_logger(__name__) logger.info("event_name", key="value", count=42) Configuration via environment variables: MCP_LOG_FIELD_LENGTH_LIMIT: Longest string value a log record carries, in characters (default: 2048). Longer ones are cut after redaction and end with ``TRUNCATION_MARKER``. """ from __future__ import annotations from collections.abc import Mapping, MutableMapping, Sequence import functools import logging import os import sys import threading import time from typing import Any, cast import structlog from structlog.types import Processor # Make the pre-`setup_logging` window safe. structlog's out-of-the-box factory # is `PrintLoggerFactory()`, which writes to **stdout** -- and on the stdio # transport stdout is the JSON-RPC stream, so a single line emitted before # `setup_logging()` runs corrupts the session and the client dies on a parse # error. That is not hypothetical: `gc.py` logged "watchdog package not installed" # at import time, i.e. while bootstrap was still importing modules (#563; the # line itself is gone since #1236, the window is not). # # Configuring this at import of *this* module closes the window for every logger # in the package, since `get_logger` lives here and cannot be reached without # importing it first. `setup_logging()` reconfigures on top later; caching stays # off so a logger bound during the early window is not frozen to this factory. structlog.configure( logger_factory=structlog.PrintLoggerFactory(file=sys.stderr), cache_logger_on_first_use=False, ) def _add_service_context(_logger: Any, _method_name: str, event_dict: MutableMapping[str, Any]) -> Mapping[str, Any]: """Add service-level context to all log entries.""" event_dict.setdefault("service", "mcp-hangar") return event_dict def _add_trace_context(_logger: Any, _method_name: str, event_dict: MutableMapping[str, Any]) -> Mapping[str, Any]: """Correlate logs with traces: add trace_id/span_id when inside an OTel span. A no-op when tracing is unavailable, uninitialized, or there is no active span. Imported lazily to avoid a circular import with the tracing module (which imports this one for its logger). Never lets a tracing error break a log call. """ try: from .observability.tracing import get_current_span_id, get_current_trace_id trace_id = get_current_trace_id() if trace_id: event_dict["trace_id"] = trace_id span_id = get_current_span_id() if span_id: event_dict["span_id"] = span_id except Exception: # noqa: BLE001 -- fault-barrier: logging must not fail on tracing pass return event_dict def _sanitize_sensitive_data( _logger: Any, _method_name: str, event_dict: MutableMapping[str, Any] ) -> Mapping[str, Any]: """Redact sensitive fields from log output.""" sensitive_keys = { "password", "secret", "token", "api_key", "authorization", "credential", } def redact(obj: Any, depth: int = 0) -> Any: if depth > 5: # Prevent infinite recursion return obj if isinstance(obj, dict): return {k: "[REDACTED]" if k.lower() in sensitive_keys else redact(v, depth + 1) for k, v in obj.items()} if isinstance(obj, list): return [redact(item, depth + 1) for item in obj] return obj return cast(dict[str, Any], redact(event_dict)) def _redact_secret_values(_logger: Any, _method_name: str, event_dict: MutableMapping[str, Any]) -> Mapping[str, Any]: """Scrub secret *values* (tokens, keys, JWTs) from every string in the record. Complements ``_sanitize_sensitive_data`` (which redacts by key name only): this catches a secret embedded in the message string or under a non-matching key, using the shared builtin-pattern redactor. Long-string redaction is off, so it only rewrites recognizable token shapes. """ from .redactor import get_default_redactor redactor = get_default_redactor() def scrub(obj: Any, depth: int = 0) -> Any: if depth > 5: return obj if isinstance(obj, str): return redactor.redact(obj) if isinstance(obj, dict): return {k: scrub(v, depth + 1) for k, v in obj.items()} if isinstance(obj, list): return [scrub(item, depth + 1) for item in obj] return obj return cast(dict[str, Any], scrub(event_dict)) #: How a cut value ends; the number is how many characters were removed. TRUNCATION_MARKER = "…[truncated {}]" #: Hangar's bound on one string value in a log record, in characters. LOG_FIELD_LENGTH_LIMIT = 2048 LOG_FIELD_LENGTH_LIMIT_ENV = "MCP_LOG_FIELD_LENGTH_LIMIT" def truncate_text(value: str, limit: int) -> str: """``value`` if it fits in ``limit`` characters, else cut to fit and ending with the marker. The marker counts toward the limit, so the result is never longer than ``limit``. A limit too short to hold the marker cuts to the limit without one. """ if len(value) <= limit: return value keep = limit - len(TRUNCATION_MARKER.format(len(value))) # room left beside the widest marker it can need if keep <= 0: return value[:limit] return value[:keep] + TRUNCATION_MARKER.format(len(value) - keep) def env_length_limit(name: str) -> int | None: """The positive integer in environment variable ``name``; None when it is unset or empty. Any other value is ignored too, with one warning per distinct value. """ raw = os.environ.get(name, "").strip() return _parse_length_limit(name, raw) if raw else None @functools.lru_cache(maxsize=64) def _parse_length_limit(name: str, raw: str) -> int | None: try: value = int(raw) except ValueError: value = 0 if value > 0: return value get_logger(__name__).warning("length_limit_invalid", variable=name, value=raw, expected="a positive integer") return None def _truncate_long_values(limit: int) -> Processor: """A processor cutting every string in the record to ``limit`` characters, marker included. It must run after both redaction processors: a secret is redacted whole before a cut could leave a fragment too short for its pattern to match. It reaches as deep as they do, five levels. """ def cut(obj: Any, depth: int = 0) -> Any: if depth > 5: return obj if isinstance(obj, str): return truncate_text(obj, limit) if isinstance(obj, dict): return {k: cut(v, depth + 1) for k, v in obj.items()} if isinstance(obj, list): return [cut(item, depth + 1) for item in obj] return obj def truncate_long_values( _logger: Any, _method_name: str, event_dict: MutableMapping[str, Any] ) -> Mapping[str, Any]: return cast(dict[str, Any], cut(event_dict)) return truncate_long_values def _drop_color_message_key(_logger: Any, _method_name: str, event_dict: MutableMapping[str, Any]) -> Mapping[str, Any]: """Remove the color_message key that uvicorn adds.""" event_dict.pop("color_message", None) return event_dict def setup_logging( level: str = "INFO", json_format: bool = False, development: bool | None = None, log_file: str | None = None, ) -> None: """Configure structlog for the entire application. This function should be called once at application startup, before any logging occurs. Args: level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). json_format: If True, output logs as JSON (recommended for production). development: If True, use colored console output. Defaults to not json_format. log_file: Optional path to log file. If provided, logs will also be written to this file. Every string in a record is cut to MCP_LOG_FIELD_LENGTH_LIMIT characters (default ``LOG_FIELD_LENGTH_LIMIT``), read here, once. """ if development is None: development = not json_format field_limit = env_length_limit(LOG_FIELD_LENGTH_LIMIT_ENV) or LOG_FIELD_LENGTH_LIMIT # Shared processors for all log entries shared_processors: Sequence[Processor] = [ structlog.contextvars.merge_contextvars, structlog.stdlib.add_log_level, structlog.stdlib.add_logger_name, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), _add_service_context, _add_trace_context, _sanitize_sensitive_data, _redact_secret_values, _truncate_long_values(field_limit), # after redaction, never before _drop_color_message_key, structlog.processors.UnicodeDecoder(), ] if development: # Colored, readable output for development renderer: Processor = structlog.dev.ConsoleRenderer( colors=True, exception_formatter=structlog.dev.plain_traceback, ) else: # JSON output for production renderer = structlog.processors.JSONRenderer() # Configure structlog structlog.configure( processors=list(shared_processors) + [ structlog.stdlib.ProcessorFormatter.wrap_for_formatter, ], logger_factory=structlog.stdlib.LoggerFactory(), wrapper_class=structlog.stdlib.BoundLogger, cache_logger_on_first_use=True, ) # Create formatter for stdlib logging integration formatter = structlog.stdlib.ProcessorFormatter( foreign_pre_chain=list(shared_processors), processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, renderer, ], ) # Configure root logger root_logger = logging.getLogger() root_logger.handlers.clear() root_logger.setLevel(level.upper()) # Console handler (stderr for MCP compatibility) console_handler = logging.StreamHandler(sys.stderr) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # Optional file handler if log_file: try: from pathlib import Path Path(log_file).parent.mkdir(parents=True, exist_ok=True) file_handler = logging.FileHandler(log_file, mode="a", encoding="utf-8") # Always use JSON format for file logs file_formatter = structlog.stdlib.ProcessorFormatter( foreign_pre_chain=list(shared_processors), processors=[ structlog.stdlib.ProcessorFormatter.remove_processors_meta, structlog.processors.JSONRenderer(), ], ) file_handler.setFormatter(file_formatter) root_logger.addHandler(file_handler) except Exception as e: # noqa: BLE001 -- fault-barrier: file logging setup failure must not crash application root_logger.warning(f"Could not setup file logging: {e}") # Silence noisy third-party loggers or ensure they use structlog logging.getLogger("asyncio").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("httpcore").setLevel(logging.WARNING) # Uvicorn loggers - set higher level to suppress INFO messages logging.getLogger("uvicorn").setLevel(logging.WARNING) logging.getLogger("uvicorn.error").setLevel(logging.WARNING) logging.getLogger("uvicorn.access").setLevel(logging.WARNING) # MCP library - keep at INFO but it will be formatted by structlog # logging.getLogger("mcp").setLevel(logging.INFO) def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger: """Get a configured structlog logger. Args: name: Logger name (typically __name__). If None, returns root logger. Returns: A bound structlog logger with all configured processors. Example: logger = get_logger(__name__) logger.info("user_logged_in", user_id=123, ip="192.168.1.1") """ return cast(structlog.stdlib.BoundLogger, structlog.get_logger(name)) #: Last emission time per throttle key, for :func:`should_log_now`. _throttle_last_emitted: dict[str, float] = {} _throttle_lock = threading.Lock() #: Default quiet period for a throttled log line, in seconds. THROTTLE_INTERVAL_S = 60.0 def should_log_now(key: str, interval_s: float = THROTTLE_INTERVAL_S) -> bool: """Whether a throttled log line keyed by *key* may be emitted now. For conditions that are true on EVERY request while they last -- a front door denying for want of an identity, a projection that resolves to nothing -- where the first occurrence is the whole signal and the next thousand are noise that would bury it. Emits once per *key* per *interval_s*. Deliberately not a rate limiter: no token bucket, no burst allowance. The caller passes a key coarse enough to be bounded (a reason, or a reason plus a tenant), because an unbounded key set would leak memory here the same way an unbounded label set would blow up a metric. Returns: True when the caller should log, False when it should stay quiet. """ now = time.monotonic() with _throttle_lock: last = _throttle_last_emitted.get(key) if last is not None and (now - last) < interval_s: return False _throttle_last_emitted[key] = now return True def reset_log_throttle() -> None: """Forget every throttle key. For tests, which must not inherit a quiet period.""" with _throttle_lock: _throttle_last_emitted.clear() # Convenience aliases for common log levels def debug(event: str, **kwargs: Any) -> None: """Log a debug message.""" get_logger().debug(event, **kwargs) def info(event: str, **kwargs: Any) -> None: """Log an info message.""" get_logger().info(event, **kwargs) def warning(event: str, **kwargs: Any) -> None: """Log a warning message.""" get_logger().warning(event, **kwargs) def error(event: str, **kwargs: Any) -> None: """Log an error message.""" get_logger().error(event, **kwargs) def exception(event: str, **kwargs: Any) -> None: """Log an exception with traceback.""" get_logger().exception(event, **kwargs)