core.config.structlog_config ============================ .. py:module:: core.config.structlog_config .. autoapi-nested-parse:: Shared structlog configuration for all jobmon components. This module provides structlog configuration that enables: 1. Context variable merging (required for @bind_context decorator) 2. Basic stdlib metadata decoration (logger name, level) while deferring rendering/formatting to the host application 3. Optional Jobmon telemetry isolation and OTLP capture 4. Optional component identification in logs Attributes ---------- .. autoapisummary:: core.config.structlog_config._thread_local core.config.structlog_config._otlp_handler_count core.config.structlog_config._otlp_capture_lock Classes ------- .. autoapisummary:: core.config.structlog_config._NamedPrintLogger core.config.structlog_config._NamedPrintLoggerFactory Functions --------- .. autoapisummary:: core.config.structlog_config._store_event_dict_for_otlp core.config.structlog_config._forward_event_to_logging_handlers core.config.structlog_config._prune_event_dict_for_console core.config.structlog_config._ensure_logger_name core.config.structlog_config._wrap_print_logger_factory core.config.structlog_config._wrap_wrapper_class_for_otlp core.config.structlog_config._build_structlog_processor_chain core.config.structlog_config._extract_exc_info core.config.structlog_config.enable_structlog_otlp_capture core.config.structlog_config.disable_structlog_otlp_capture core.config.structlog_config.structlog_otlp_capture_enabled core.config.structlog_config.configure_structlog core.config.structlog_config._create_component_processor core.config.structlog_config.create_telemetry_isolation_processor core.config.structlog_config.configure_structlog_with_otlp core.config.structlog_config.prepend_jobmon_processors_to_existing_config core.config.structlog_config.is_structlog_configured core.config.structlog_config._iter_factory_candidates core.config.structlog_config._looks_like_direct_rendering_factory core.config.structlog_config._wrapper_indicates_direct_rendering core.config.structlog_config._uses_stdlib_integration core.config.structlog_config._processor_present core.config.structlog_config._has_isolation_processor Module Contents --------------- .. py:class:: _NamedPrintLogger(logger_name: str, base_logger: structlog._output.PrintLogger) Wrapper that adds a ``name`` attribute to ``PrintLogger``. .. py:attribute:: __slots__ :value: ('_logger', 'name') .. py:attribute:: _logger .. py:attribute:: name .. py:method:: __getattr__(item: str) -> Any .. py:method:: __repr__() -> str .. py:class:: _NamedPrintLoggerFactory(base_factory: structlog._output.PrintLoggerFactory) Factory that decorates ``PrintLogger`` instances with a name. .. py:attribute:: __jobmon_named_factory__ :value: True .. py:attribute:: _base_factory .. py:method:: __call__(*args: Any) -> _NamedPrintLogger .. py:data:: _thread_local .. py:data:: _otlp_handler_count :value: 0 .. py:data:: _otlp_capture_lock .. py:function:: _store_event_dict_for_otlp(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any] Store ``event_dict`` in thread-local storage for OTLP exporters. .. py:function:: _forward_event_to_logging_handlers(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any] Forward structlog events to stdlib handlers when hosts render directly. .. py:function:: _prune_event_dict_for_console(event_dict: Dict[str, Any]) -> Dict[str, Any] Strip Jobmon-bound metadata while leaving host formatting intact. Removes all keys starting with 'telemetry_' prefix along with helper fields added for OTLP forwarding (``logger`` and ``level``). This keeps console output clean while preserving the full context for OTLP exports. .. py:function:: _ensure_logger_name(logger: Any, method_name: str, event_dict: Dict[str, Any]) -> Dict[str, Any] Ensure the event dict contains a ``logger`` field. .. py:function:: _wrap_print_logger_factory(factory: Any) -> Any Decorate PrintLogger factories so produced loggers expose ``name``. .. py:function:: _wrap_wrapper_class_for_otlp(wrapper_class: Any) -> Any Ensure filtered log levels still execute processors for OTLP capture. .. py:function:: _build_structlog_processor_chain(*, uses_stdlib_integration: bool, component_name: Optional[str], telemetry_logger_prefixes: Optional[Sequence[str]], extra_processors: Iterable[Callable], include_store_for_otlp: bool, include_wrap_for_formatter: bool, include_telemetry_isolation: bool = True) -> List[Callable] Compose the processor chain shared across Jobmon structlog setup paths. :param uses_stdlib_integration: If True, configures for stdlib logging integration. If False, configures for direct rendering (e.g., PrintLogger). :param component_name: Optional component name to add to all logs. :param telemetry_logger_prefixes: Logger name prefixes for telemetry isolation. :param extra_processors: Additional processors to insert after telemetry isolation. :param include_store_for_otlp: Whether to include OTLP event storage processor. :param include_wrap_for_formatter: Whether to include ProcessorFormatter wrapper (for stdlib). :param include_telemetry_isolation: Whether to include telemetry isolation processor. .. py:function:: _extract_exc_info(event_dict: Dict[str, Any]) -> Optional[Any] Translate structlog exception metadata into logging exc_info tuple. .. py:function:: enable_structlog_otlp_capture() -> None Enable thread-local capture for OTLP handlers. .. py:function:: disable_structlog_otlp_capture() -> None Disable OTLP capture (used for tests or when handlers are removed). .. py:function:: structlog_otlp_capture_enabled() -> Iterator[None] Context manager to enable OTLP capture temporarily. Ensures the reference count is decremented even if an exception is raised. Particularly useful in tests that need to toggle capture around assertions. .. py:function:: configure_structlog(component_name: Optional[str] = None, *, extra_processors: Optional[Iterable[Callable]] = None) -> None Configure structlog for jobmon components. This function sets up structlog with processors that: 1. Merge context variables (from bind_contextvars and @bind_context decorator) 2. Add logger metadata (logger name, log level) 3. Optionally add a component field to the event_dict 4. Isolate Jobmon telemetry metadata to jobmon.* loggers 5. Capture the raw event_dict for OTLP handlers 6. Optionally include extra processors supplied by the caller IMPORTANT: This must be called before using the @bind_context decorator or any structlog.contextvars.bind_contextvars() calls. PROCESSOR CHAIN ORDER (after configuration): 1. merge_contextvars (Jobmon context) 2. component processor (optional, when component_name is provided) 3. filter_by_level (stdlib) 4. add_logger_name (stdlib) 5. add_log_level (stdlib) 6. telemetry isolation processor 7. _store_event_dict_for_otlp (Jobmon OTLP capture) 8. Extra processors supplied via ``extra_processors`` (if any) 9. ProcessorFormatter.wrap_for_formatter (stdlib - keeps stdlib handlers working) :param component_name: Component name to add to all logs (e.g., "distributor") :param extra_processors: Additional structlog processors to append after Jobmon's defaults (e.g., custom formatting or telemetry processors) .. rubric:: Example >>> # Basic usage >>> configure_structlog(component_name="distributor") .. py:function:: _create_component_processor(component_name: str) -> Callable Create a processor that adds component name to all log events. :param component_name: Name of the component (e.g., "distributor", "worker") :returns: Processor function that adds component field to event_dict .. py:function:: create_telemetry_isolation_processor(telemetry_prefixes: List[str]) -> Callable[[Any, str, Dict[str, Any]], Dict[str, Any]] Create a processor that isolates telemetry metadata to specific logger prefixes. :param telemetry_prefixes: List of logger name prefixes that should receive telemetry metadata (e.g., ["jobmon.", "myapp.telemetry"]). :returns: A structlog processor function that isolates telemetry metadata. .. py:function:: configure_structlog_with_otlp(component_name: str) -> None Configure structlog with OTLP trace integration if enabled. Checks JobmonConfig for telemetry.tracing settings and configures structlog with OpenTelemetry trace processors if enabled. This is the recommended function to call from component CLIs as it automatically handles OTLP configuration based on settings. :param component_name: Component name (e.g., "distributor", "worker") .. rubric:: Example >>> # In distributor CLI >>> configure_structlog_with_otlp(component_name="distributor") >>> # In worker CLI >>> configure_structlog_with_otlp(component_name="worker") .. py:function:: prepend_jobmon_processors_to_existing_config() -> None Prepend Jobmon processors to an existing structlog configuration. This is called when Jobmon is used as a library and the host application has already configured structlog. It intelligently prepends Jobmon's processors to the existing processor chain to enable telemetry context management and isolation while preserving the host application's final rendering. Adapts to the host app's logging architecture: - Stdlib integration adds merge_contextvars, filter_by_level, add_logger_name, and telemetry isolation - Direct rendering (like FHS) adds merge_contextvars and telemetry isolation - Host processors remain untouched so final rendering is preserved .. py:function:: is_structlog_configured() -> bool Check if structlog has been configured. :returns: True if structlog is configured, False otherwise .. rubric:: Example >>> if not is_structlog_configured(): ... configure_structlog(component_name="my_component") .. py:function:: _iter_factory_candidates(factory: Any) -> Iterator[Any] Yield the given factory and any nested factories exposed via common attrs. .. py:function:: _looks_like_direct_rendering_factory(factory: Any) -> bool Best-effort detection of structlog PrintLogger-based factories. .. py:function:: _wrapper_indicates_direct_rendering(wrapper_class: Any) -> bool Detect wrapper classes that imply direct rendering (e.g., PrintLogger). .. py:function:: _uses_stdlib_integration(logger_factory: Any, wrapper_class: Any) -> bool Best-effort detection of stdlib integration vs direct rendering. .. py:function:: _processor_present(processors: Iterable[Callable], processor: Callable) -> bool Return True when the exact processor function is already present. .. py:function:: _has_isolation_processor(processors: Iterable[Callable], prefixes: tuple[str, Ellipsis]) -> bool Check whether an isolation processor for the given prefixes already exists.