Skip to main content

telemetry.bootstrap

OpenTelemetry bootstrap and instrumentation helpers for exp.arolariu.ro.

TelemetryDependencies Objects​

@dataclass(frozen=True, slots=True)
class TelemetryDependencies()

Container for lazily imported OpenTelemetry and Azure Monitor symbols.

TelemetryRuntime Objects​

@dataclass(slots=True)
class TelemetryRuntime()

Represents the active telemetry runtime for the current process.

TraceContextSnapshot Objects​

@dataclass(frozen=True, slots=True)
class TraceContextSnapshot()

Formatted trace/span identifiers for the current execution context.

SafeOtelFormatter Objects​

class SafeOtelFormatter(logging.Formatter)

Formatter that tolerates missing OpenTelemetry correlation fields.

initialize_telemetry​

def initialize_telemetry(app: FastAPI) -> TelemetryRuntime

Initialize OpenTelemetry providers and instrument the FastAPI app.

shutdown_telemetry​

def shutdown_telemetry() -> None

Flush and shut down the current telemetry runtime.

reset_telemetry_state​

def reset_telemetry_state() -> None

Reset telemetry module state for tests without touching global OTel SDK state.

start_span​

@contextmanager
def start_span(name: str,
               *,
               instrumentation_scope: str,
               attributes: AttributeMap | None = None) -> Iterator[Any | None]

Start a manual span when telemetry is active, otherwise yield None.

set_current_span_attributes​

def set_current_span_attributes(attributes: AttributeMap) -> None

Attach attributes to the current span when one is active.

record_current_span_exception​

def record_current_span_exception(exception: Exception) -> None

Record an exception on the current span when one is active.

get_current_trace_context​

def get_current_trace_context() -> TraceContextSnapshot | None

Return formatted trace/span IDs for the current execution context.

record_config_load_metric​

def record_config_load_metric(*, source: str, outcome: str,
                              duration_ms: float) -> None

Record metrics for one config-load operation.

record_auth_decision_metric​

def record_auth_decision_metric(*, flow: str, outcome: str,
                                resolved_target: str | None) -> None

Record one authorization decision for metrics backends.

record_config_delivery_metric​

def record_config_delivery_metric(*, endpoint_name: str, target: str,
                                  value_count: int) -> None

Record metrics for one successful config-serving response.

// was this page useful?