core.otlp

Elegant OTLP integration for jobmon that prevents global log pollution.

Submodules

Attributes

`OTLP_AVAILABLE`
`JobmonOTLPStructlogHandler`
`OpenTelemetryLogFormatter`

Classes

`JobmonOTLPFormatter`	Formatter that adds OpenTelemetry span details to jobmon logs.
`JobmonOTLPLoggingHandler`	Universal OTLP logging handler with lazy initialization and attribute extraction.
`JobmonOTLPManager`	OTLP manager for shared trace and log resources.

Functions

`create_log_exporter`(→ Optional[Any])	Create a pre-configured log exporter for client applications.
`get_logger`(→ Optional[Any])	Get a logger from the shared provider.
`get_shared_logger_provider`(→ Optional[Any])	Get the shared logger provider, initializing if needed.
`initialize_jobmon_otlp`(→ JobmonOTLPManager)	Initialize OTLP for shared resources (traces and logs).
`otlp_flush_on_exit`(...)	Context manager that guarantees OTLP flush when exiting.
`validate_otlp_exporter_config`(→ list[str])	Validate OTLP exporter configuration and return list of issues.
`add_span_details_processor`(→ Dict[str, Any])	Structlog processor to add OpenTelemetry span details to log entries.
`get_current_span_details`(→ Tuple[Optional[str], ...)	Get details of the current OpenTelemetry span.

Package Contents

core.otlp.OTLP_AVAILABLE = True

class core.otlp.JobmonOTLPFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: logging.Formatter

Formatter that adds OpenTelemetry span details to jobmon logs.

Note: For handlers using _JobmonOTLPLoggingHandler, this formatter is not used for OTLP output (handler creates OTLPLogRecord directly). It only affects console/file output.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

format(record: logging.LogRecord) → str: Format log record with OpenTelemetry span details.

class core.otlp.JobmonOTLPLoggingHandler(level: int = logging.NOTSET, exporter: Any | Dict | None = None, logger_provider: Any | None = None)

Bases: logging.Handler

Universal OTLP logging handler with lazy initialization and attribute extraction.

This handler works with both stdlib logging and structlog. It extracts structured attributes from structlog’s thread-local event_dict when available, enabling rich OTLP exports with queryable fields.

Note: JobmonOTLPStructlogHandler is an alias for this class (kept for backward compatibility). Both names can be used interchangeably in configuration.

Supports flexible configuration patterns:

Inline dict configuration (server pattern):

handlers:
  otlp_logs:
    class: jobmon.core.otlp.JobmonOTLPLoggingHandler
    level: INFO
    exporter:
      module: opentelemetry.exporter.otlp.proto.grpc._log_exporter
      class: OTLPLogExporter
      endpoint: otelcol.dev.aks.scicomp.ihme.washington.edu:443
      options: [["grpc.max_send_message_length", 16777216]]
      max_export_batch_size: 8

Pre-configured exporter instance:

handler = JobmonOTLPLoggingHandler(exporter=my_exporter)

Direct use with logger_provider (for testing):

handler = JobmonOTLPLoggingHandler(logger_provider=my_provider)

Initialize with optional exporter config or pre-configured logger provider.

Parameters:

level – Logging level for this handler
exporter – Either a dict configuration or pre-configured OTLP exporter instance
logger_provider – Optional pre-configured logger provider (for testing/direct use)

close() → None

Tidy up any resources used by the handler.

This version removes the handler from an internal map of handlers, _handlers, which is used for handler lookup by name. Subclasses should ensure that this gets called from overridden close() methods.

emit(record: logging.LogRecord) → None: Emit log record to OTLP with extracted attributes.

core.otlp.JobmonOTLPStructlogHandler

class core.otlp.JobmonOTLPManager

OTLP manager for shared trace and log resources.

This manager handles: - Trace provider setup (for distributed tracing) - Logger provider setup (for OTLP log export) - Resource detection (shared across components) - Request instrumentation (shared utility)

All OTLP handlers should use the shared logger provider to avoid duplicate connections and log emissions.

Initialize the OTLP manager.

tracer_provider: Any | None = None

logger_provider: Any | None = None

classmethod get_instance() → JobmonOTLPManager: Get or create the singleton OTLP manager with thread safety.

initialize() → None: Initialize trace and log providers with jobmon resources.

get_tracer(name: str) → Any | None: Get a tracer for distributed tracing.

classmethod instrument_requests() → None: Instrument requests library for HTTP tracing.

shutdown() → None: Shutdown trace and log providers.

flush_and_shutdown() → None: Flush pending OTLP telemetry and shut down providers.

core.otlp.create_log_exporter(**kwargs: Any) → Any | None

Create a pre-configured log exporter for client applications.

This factory function creates exporters that can be passed to JobmonOTLPLoggingHandler for pure separation.

Parameters:: **kwargs – Exporter configuration (endpoint, headers, etc.)
Returns:: Pre-configured OTLP log exporter, or None if unavailable

Example:

exporter = create_log_exporter(
    endpoint="otelcol.dev.aks:443",
    max_batch_size=8
)
handler = JobmonOTLPLoggingHandler(exporter=exporter)

core.otlp.get_logger(name: str) → Any | None

Get a logger from the shared provider.

This is the cleanest way for handlers to get OTLP loggers without dealing with manager instances or initialization complexity.

Parameters:: name – Logger name (typically __name__)
Returns:: OTLP logger instance, or None if unavailable

core.otlp.get_shared_logger_provider() → Any | None

Get the shared logger provider, initializing if needed.

This function provides a clean interface for handlers to access the shared LoggerProvider without dealing with manager instances directly. Uses double-checked locking to prevent race conditions in multi-threaded environments like Kubernetes with multiple workers.

Returns:: The shared LoggerProvider instance, or None if unavailable

core.otlp.initialize_jobmon_otlp() → JobmonOTLPManager

Initialize OTLP for shared resources (traces and logs).

This creates shared TracerProvider and LoggerProvider instances that should be used by all OTLP handlers to avoid duplicate connections.

Returns:: The OTLP manager instance with shared providers

core.otlp.otlp_flush_on_exit() → Generator[JobmonOTLPManager | None, None, None]: Context manager that guarantees OTLP flush when exiting.

core.otlp.validate_otlp_exporter_config(config: Any, exporter_type: str = 'log') → list[str]

Validate OTLP exporter configuration and return list of issues.

Parameters:

config – Exporter configuration dictionary
exporter_type – Type of exporter (‘log’, ‘trace’, ‘metric’)

Returns:

List of validation error messages. Empty list if valid.

core.otlp.add_span_details_processor(logger: Any, method_name: str, event_dict: Dict[str, Any]) → Dict[str, Any]

Structlog processor to add OpenTelemetry span details to log entries.

Parameters:

logger – The logger instance (not used, but required by Structlog processor signature).
method_name – The logging method name (e.g., “info”, “debug”).
event_dict – The event dictionary representing the log entry.

Returns:

The modified event dictionary with OpenTelemetry span details added.

core.otlp.get_current_span_details() → Tuple[str | None, str | None, str | None]

Get details of the current OpenTelemetry span.

Returns:: Tuple of (span_id, trace_id, parent_span_id) as hex strings, or (None, None, None)

core.otlp.OpenTelemetryLogFormatter