tessl/pypi-opentelemetry-api
OpenTelemetry Python API providing core abstractions for distributed tracing, metrics collection, and context propagation
—
Pending
tracing.mddocs/
Distributed Tracing
Comprehensive distributed tracing capabilities for creating and managing spans across services. OpenTelemetry tracing provides the foundation for distributed system observability with support for span relationships, attributes, events, and W3C Trace Context propagation.
Capabilities
Tracer Creation and Management
Get tracers for creating spans with proper instrumentation scope identification.
def get_tracer(
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
tracer_provider: Optional[TracerProvider] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Tracer:
"""
Returns a Tracer for use by the given instrumentation library.
Parameters:
- instrumenting_module_name: The name of the instrumentation scope
- instrumenting_library_version: Optional version of the instrumentation library
- tracer_provider: Optional specific TracerProvider to use
- schema_url: Optional Schema URL of the emitted telemetry
- attributes: Optional attributes of the emitted telemetry
Returns:
Tracer instance for creating spans
"""
def get_tracer_provider() -> TracerProvider:
"""Gets the current global TracerProvider object."""
def set_tracer_provider(tracer_provider: TracerProvider) -> None:
"""
Sets the current global TracerProvider object.
This can only be done once, a warning will be logged if any further attempt is made.
"""Span Creation and Lifecycle
Create spans to represent operations in distributed traces with full lifecycle management.
class Tracer(ABC):
def start_span(
self,
name: str,
context: Optional[Context] = None,
kind: SpanKind = SpanKind.INTERNAL,
attributes: Attributes = None,
links: Optional[Sequence[Link]] = None,
start_time: Optional[int] = None,
record_exception: bool = True,
set_status_on_exception: bool = True,
) -> Span:
"""
Starts a span.
Parameters:
- name: The name of the span to be created
- context: An optional Context containing the span's parent
- kind: The span's kind (relationship to parent)
- attributes: The span's attributes
- links: Links span to other spans
- start_time: Sets the start time of a span
- record_exception: Whether to record exceptions as error events
- set_status_on_exception: Whether to set span status on exceptions
Returns:
The newly-created span
"""
def start_as_current_span(
self,
name: str,
context: Optional[Context] = None,
kind: SpanKind = SpanKind.INTERNAL,
attributes: Attributes = None,
links: Optional[Sequence[Link]] = None,
start_time: Optional[int] = None,
record_exception: bool = True,
set_status_on_exception: bool = True,
end_on_exit: bool = True,
) -> Iterator[Span]:
"""
Context manager for creating a new span and set it as the current span.
Parameters:
- name: The name of the span to be created
- context: An optional Context containing the span's parent
- kind: The span's kind (relationship to parent)
- attributes: The span's attributes
- links: Links span to other spans
- start_time: Sets the start time of a span
- record_exception: Whether to record exceptions as error events
- set_status_on_exception: Whether to set span status on exceptions
- end_on_exit: Whether to end the span when leaving the context manager
Yields:
The newly-created span
"""Span Operations and Metadata
Manage span data including attributes, events, status, and relationships.
class Span(ABC):
def get_span_context(self) -> SpanContext:
"""Returns the SpanContext for this span."""
def set_attribute(self, key: str, value: AttributeValue) -> None:
"""Sets a single attribute on the span."""
def set_attributes(self, attributes: Attributes) -> None:
"""Sets multiple attributes on the span."""
def add_event(
self,
name: str,
attributes: Attributes = None,
timestamp: Optional[int] = None,
) -> None:
"""
Adds an event to the span.
Parameters:
- name: Name of the event
- attributes: Event attributes
- timestamp: Optional timestamp of the event
"""
def update_name(self, name: str) -> None:
"""Updates the span name."""
def set_status(self, status: Status) -> None:
"""Sets the span status."""
def record_exception(
self,
exception: BaseException,
attributes: Attributes = None,
timestamp: Optional[int] = None,
escaped: bool = False,
) -> None:
"""
Records an exception as a span event.
Parameters:
- exception: The exception to record
- attributes: Additional attributes for the exception event
- timestamp: Optional timestamp of the exception
- escaped: Whether the exception escaped the span scope
"""
def is_recording(self) -> bool:
"""Returns True if the span is recording data."""
def end(self, end_time: Optional[int] = None) -> None:
"""Ends the span."""Span Context and Propagation
Access span context for propagation and parent-child relationships.
class SpanContext:
"""Immutable span state for propagation."""
@property
def trace_id(self) -> int:
"""The trace ID of the span context."""
@property
def span_id(self) -> int:
"""The span ID of the span context."""
@property
def trace_flags(self) -> TraceFlags:
"""The trace flags of the span context."""
@property
def trace_state(self) -> TraceState:
"""The trace state of the span context."""
@property
def is_valid(self) -> bool:
"""Returns True if the span context is valid."""
def get_current_span(context: Optional[Context] = None) -> Span:
"""Returns the current span from the context."""
def set_span_in_context(span: Span, context: Optional[Context] = None) -> Context:
"""Returns a new context with the span set."""
def use_span(
span: Span,
end_on_exit: bool = False,
record_exception: bool = True,
set_status_on_exception: bool = True,
) -> Iterator[Span]:
"""
Takes a non-active span and activates it in the current context.
Parameters:
- span: The span that should be activated
- end_on_exit: Whether to end the span when leaving the context
- record_exception: Whether to record exceptions as error events
- set_status_on_exception: Whether to set span status on exceptions
Yields:
The activated span
"""Links and Relationships
Create relationships between spans across different traces.
class Link:
"""A link to a Span. The attributes of a Link are immutable."""
def __init__(
self,
context: SpanContext,
attributes: Attributes = None,
) -> None:
"""
Parameters:
- context: SpanContext of the Span to link to
- attributes: Link's attributes
"""
@property
def context(self) -> SpanContext:
"""Returns the SpanContext of the linked span."""
@property
def attributes(self) -> Attributes:
"""Returns the link's attributes."""
@property
def dropped_attributes(self) -> int:
"""Returns the number of dropped attributes."""Status and Error Handling
Manage span status and error conditions.
class Status:
"""Represents the status of a span."""
def __init__(
self,
status_code: StatusCode = StatusCode.UNSET,
description: Optional[str] = None,
) -> None:
"""
Parameters:
- status_code: The status code
- description: Optional description of the status
"""
@property
def status_code(self) -> StatusCode:
"""Returns the status code."""
@property
def description(self) -> Optional[str]:
"""Returns the status description."""
class StatusCode(Enum):
"""Represents the canonical status of a span."""
UNSET = 0 # Default status
OK = 1 # Operation completed successfully
ERROR = 2 # Operation contains an errorProvider Implementations
Provider classes for different deployment scenarios.
class TracerProvider(ABC):
"""Abstract base class for tracer providers."""
def get_tracer(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Tracer:
"""Returns a Tracer for use by the given instrumentation library."""
class NoOpTracerProvider(TracerProvider):
"""The default TracerProvider, used when no implementation is available."""
def get_tracer(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Tracer:
"""Returns a no-op tracer."""
class ProxyTracerProvider(TracerProvider):
"""Proxy tracer provider for late binding of real providers."""
def get_tracer(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Tracer:
"""Returns a tracer from the real provider or proxy tracer."""Usage Examples
Basic Span Creation
from opentelemetry import trace
# Get a tracer
tracer = trace.get_tracer(__name__)
# Create a simple span
with tracer.start_as_current_span("my-operation") as span:
span.set_attribute("user.id", "12345")
span.add_event("Processing started")
# Do work
result = process_data()
span.set_attribute("result.count", len(result))
span.add_event("Processing completed")Parent-Child Span Relationships
from opentelemetry import trace
from opentelemetry.trace import SpanKind
tracer = trace.get_tracer(__name__)
# Parent span representing an HTTP request
with tracer.start_as_current_span("handle-request", kind=SpanKind.SERVER) as parent:
parent.set_attribute("http.method", "GET")
parent.set_attribute("http.url", "/api/users")
# Child span for database operation
with tracer.start_as_current_span("query-database", kind=SpanKind.CLIENT) as child:
child.set_attribute("db.system", "postgresql")
child.set_attribute("db.statement", "SELECT * FROM users")
# Another child span for cache check
with tracer.start_as_current_span("check-cache") as cache_span:
cache_span.set_attribute("cache.hit", True)Error Handling and Status
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("risky-operation") as span:
try:
result = perform_risky_operation()
span.set_status(Status(StatusCode.OK))
except ValueError as e:
span.record_exception(e)
span.set_status(Status(StatusCode.ERROR, str(e)))
span.set_attribute("error.type", "ValueError")
raiseConstants
# Default values
DEFAULT_TRACE_OPTIONS: TraceFlags
DEFAULT_TRACE_STATE: TraceState
# Invalid constants for no-op scenarios
INVALID_SPAN: Span
INVALID_SPAN_CONTEXT: SpanContext
INVALID_SPAN_ID: int
INVALID_TRACE_ID: int
# Utility functions
def format_span_id(span_id: int) -> str:
"""Format span ID as hex string."""
def format_trace_id(trace_id: int) -> str:
"""Format trace ID as hex string."""Install with Tessl CLI
npx tessl i tessl/pypi-opentelemetry-api