spakky-opentelemetry

OpenTelemetry SDK 브릿지 — Spakky 트레이싱을 OpenTelemetry로 연결

브릿지

spakky.plugins.opentelemetry.bridge

LogContextBridge — optional trace-to-logging context synchronization.

LogContextBridge(binder=None)

Synchronizes TraceContext fields into LogContext.

When an ILogContextBinder is available (i.e., spakky-logging is installed and registered), sync() binds the current trace_id and span_id into the structured logging context. When no binder is available, all operations are no-ops.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/bridge.py

def __init__(self, binder: ILogContextBinder | None = None) -> None:
    self.__binder = binder

sync()

Bind current TraceContext's trace_id/span_id into LogContext.

If no ILogContextBinder was injected, this is a no-op. If no TraceContext is active, unbinds trace fields.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/bridge.py

def sync(self) -> None:
    """Bind current TraceContext's trace_id/span_id into LogContext.

    If no ``ILogContextBinder`` was injected, this is a no-op.
    If no TraceContext is active, unbinds trace fields.
    """
    if self.__binder is None:
        return
    ctx = TraceContext.get()
    if ctx is not None:
        self.__binder.bind(trace_id=ctx.trace_id, span_id=ctx.span_id)
    else:
        self.__binder.unbind("trace_id", "span_id")

options: show_root_heading: false

Propagator

spakky.plugins.opentelemetry.propagator

OTelTracePropagator — OpenTelemetry SDK bridge for ITracePropagator.

OTelTracePropagator()

Bases: ITracePropagator

Bridges Spakky's TraceContext with OpenTelemetry's context propagation.

Converts between Spakky's contextvars-based TraceContext and OTel's Context system using NonRecordingSpan for lightweight propagation.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/propagator.py

def __init__(self) -> None:
    self._propagator = TraceContextTextMapPropagator()

inject(carrier)

Read the ambient TraceContext and write it into the carrier.

Converts Spakky's TraceContext to an OTel Context, then delegates to the W3C TraceContext propagator for header serialization.

Parameters:

Name	Type	Description	Default
carrier	dict[str, str]	Mutable header dictionary.	required

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/propagator.py

@override
def inject(self, carrier: dict[str, str]) -> None:
    """Read the ambient TraceContext and write it into the carrier.

    Converts Spakky's TraceContext to an OTel Context, then delegates
    to the W3C TraceContext propagator for header serialization.

    Args:
        carrier: Mutable header dictionary.
    """
    ctx = TraceContext.get()
    if ctx is None:
        return
    otel_ctx = self._to_otel_context(ctx)
    self._propagator.inject(carrier, context=otel_ctx)

extract(carrier)

Reconstruct a TraceContext from the carrier.

Delegates to the OTel propagator for header parsing, then converts the resulting OTel SpanContext back to Spakky's TraceContext.

Parameters:

Name	Type	Description	Default
carrier	dict[str, str]	Read-only header dictionary.	required

Returns:

Type	Description
TraceContext | None	The restored TraceContext, or None if headers are missing/invalid.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/propagator.py

@override
def extract(self, carrier: dict[str, str]) -> TraceContext | None:
    """Reconstruct a TraceContext from the carrier.

    Delegates to the OTel propagator for header parsing, then converts
    the resulting OTel SpanContext back to Spakky's TraceContext.

    Args:
        carrier: Read-only header dictionary.

    Returns:
        The restored TraceContext, or None if headers are missing/invalid.
    """
    otel_ctx = self._propagator.extract(carrier)
    span = trace.get_current_span(otel_ctx)
    span_context = span.get_span_context()
    if not span_context.is_valid:
        return None
    return TraceContext(
        trace_id=format(span_context.trace_id, "032x"),
        span_id=format(span_context.span_id, "016x"),
        trace_flags=span_context.trace_flags,
    )

fields()

Return the header field names used by this propagator.

Returns:

Type	Description
list[str]	List of header names (traceparent, tracestate).

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/propagator.py

@override
def fields(self) -> list[str]:
    """Return the header field names used by this propagator.

    Returns:
        List of header names (traceparent, tracestate).
    """
    return list(self._propagator.fields)

options: show_root_heading: false

후처리기

spakky.plugins.opentelemetry.post_processor

Post-processor that configures OpenTelemetry SDK and replaces W3CTracePropagator.

OTelSetupPostProcessor()

Bases: IPostProcessor, IContainerAware

Initializes OTel TracerProvider and replaces W3CTracePropagator.

On first post_process call, configures the global TracerProvider with the user's chosen exporter. For every pod that is an instance of W3CTracePropagator, returns an OTelTracePropagator instead.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/post_processor.py

def __init__(self) -> None:
    super().__init__()
    self.__configured = False

post_process(pod)

Configure TracerProvider on first call; replace W3CTracePropagator.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance being processed.	required

Returns:

Type	Description
object	OTelTracePropagator if pod is W3CTracePropagator, else pod unchanged.

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Configure TracerProvider on first call; replace W3CTracePropagator.

    Args:
        pod: The Pod instance being processed.

    Returns:
        OTelTracePropagator if pod is W3CTracePropagator, else pod unchanged.
    """
    if not self.__configured:
        self.__configured = True
        self._configure_tracer_provider()
    if isinstance(pod, W3CTracePropagator):
        return OTelTracePropagator()
    return pod

options: show_root_heading: false

설정

spakky.plugins.opentelemetry.config

OpenTelemetry plugin configuration.

ExporterType

Bases: StrEnum

Supported span exporter types.

OpenTelemetryConfig()

Bases: BaseSettings

Configuration for the OpenTelemetry SDK bridge.

Attributes:

Name	Type	Description
service_name	str	OTel service name for resource identification.
exporter_type	ExporterType	Span exporter backend (otlp, console, none).
exporter_endpoint	str	OTLP collector endpoint URL.
sample_rate	Annotated[float, Field(ge=0.0, le=1.0)]	Trace sampling rate (0.0 to 1.0).

Source code in plugins/spakky-opentelemetry/src/spakky/plugins/opentelemetry/config.py

def __init__(self) -> None:
    super().__init__()

options: show_root_heading: false

에러

spakky.plugins.opentelemetry.error

Error classes for the spakky-opentelemetry package.

AbstractSpakkyOpenTelemetryError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for all OpenTelemetry plugin errors.

UnsupportedExporterTypeError

Bases: AbstractSpakkyOpenTelemetryError

Raised when the configured exporter type is not supported.

options: show_root_heading: false