tessl/pypi-opencensus
A comprehensive observability framework providing distributed tracing, metrics collection, and statistics gathering capabilities for Python applications.
—
Pending
tracing.mddocs/
Distributed Tracing
Core tracing functionality that forms the foundation of OpenCensus observability. Provides distributed tracing capabilities including span creation, context management, sampling strategies, and trace export to various backends.
Capabilities
Tracer Management
The Tracer class coordinates span creation, sampling decisions, and trace export. It serves as the main entry point for creating and managing distributed traces.
class Tracer:
"""
Main tracing coordinator that manages span creation and export.
Parameters:
- span_context: SpanContext, existing trace context
- sampler: Sampler, sampling strategy (default: probability sampler)
- exporter: Exporter, trace export destination
- propagator: Propagator, context propagation format
"""
def __init__(self, span_context=None, sampler=None, exporter=None, propagator=None): ...
def span(self, name='span'):
"""
Create a new span as a context manager.
Parameters:
- name: str, span name/operation name
Returns:
Span context manager
"""
def should_sample(self):
"""
Determine if current trace should be sampled.
Returns:
bool: True if trace should be sampled
"""
def get_tracer(self):
"""Get the current tracer instance."""
def store_tracer(self):
"""Store tracer in execution context."""
def finish(self):
"""Finish the current trace and export data."""
def start_span(self, name='span'):
"""
Start a new span and set it as current span.
Parameters:
- name: str, span name/operation name
Returns:
Span: The created span
"""
def end_span(self):
"""
End the current span.
Update the span_id in SpanContext to the current span's parent span id;
Update the current span; Send the span to exporter.
"""
def current_span(self):
"""
Return the current span.
Returns:
Span: The current active span
"""
def add_attribute_to_current_span(self, attribute_key, attribute_value):
"""
Add attribute to current span.
Parameters:
- attribute_key: str, attribute name
- attribute_value: str, attribute value
"""
def trace_decorator(self):
"""
Decorator to trace a function.
Returns:
function: Decorator function that wraps the target function with tracing
"""Span Operations
Individual units of work in a distributed trace. Spans represent operations with timing information, attributes, annotations, and hierarchical relationships.
class Span:
"""
Individual trace span representing a unit of work.
Parameters:
- name: str, span name/operation name
- parent_span: Span, parent span for hierarchy
- attributes: dict, initial span attributes
- start_time: datetime, span start time
- end_time: datetime, span end time (for completed spans)
- span_id: str, unique span identifier
- stack_trace: StackTrace, call stack capture
- annotations: list, time-based annotations
- message_events: list, message/log events
- links: list, links to other spans
- status: Status, span completion status
- same_process_as_parent_span: bool, process relationship
- context_tracer: Tracer, associated tracer
- span_kind: SpanKind, span type (client, server, etc.)
"""
def __init__(self, name, parent_span=None, attributes=None, start_time=None,
end_time=None, span_id=None, stack_trace=None, annotations=None,
message_events=None, links=None, status=None,
same_process_as_parent_span=None, context_tracer=None,
span_kind=SpanKind.UNSPECIFIED): ...
def add_attribute(self, key, value):
"""
Add key-value attribute to span.
Parameters:
- key: str, attribute name
- value: str/int/float/bool, attribute value
"""
def add_annotation(self, description, **attrs):
"""
Add time-based annotation to span.
Parameters:
- description: str, annotation description
- **attrs: additional annotation attributes
Note: Timestamp is automatically set to current UTC time
"""
def add_message_event(self, message_event):
"""
Add message/log event to span.
Parameters:
- message_event: MessageEvent, structured message
"""
def add_link(self, link):
"""
Add link to another span.
Parameters:
- link: Link, span link object
"""
def set_status(self, status):
"""
Set span completion status.
Parameters:
- status: Status, gRPC-compatible status
"""
def start(self):
"""Start the span timing."""
def finish(self):
"""End the span and record completion time."""Span Context
Represents the current trace context including trace ID, span ID, and trace options for context propagation across service boundaries.
class SpanContext:
"""
Current trace context for propagation.
Parameters:
- trace_id: str, unique trace identifier (32 hex chars)
- span_id: str, unique span identifier (16 hex chars)
- trace_options: TraceOptions, trace flags and options
- from_header: bool, whether context came from header
- tracestate: str, W3C tracestate for vendor data
"""
def __init__(self, trace_id=None, span_id=None, trace_options=None,
tracestate=None, from_header=False): ...
@property
def trace_id(self):
"""str: Unique trace identifier"""
@property
def span_id(self):
"""str: Unique span identifier"""
@property
def trace_options(self):
"""TraceOptions: Trace configuration flags"""
@property
def from_header(self):
"""bool: Whether context originated from header"""
@property
def tracestate(self):
"""str: W3C tracestate vendor data"""Sampling Strategies
Control which traces are collected and exported to manage performance impact and storage costs.
class Sampler:
"""Base class for sampling strategies."""
def should_sample(self, span_context):
"""
Determine if span should be sampled.
Parameters:
- span_context: SpanContext, current trace context
Returns:
bool: True if span should be sampled
"""
class AlwaysOnSampler(Sampler):
"""Samples every trace (100% sampling rate)."""
class AlwaysOffSampler(Sampler):
"""Samples no traces (0% sampling rate)."""
class ProbabilitySampler(Sampler):
"""
Fixed probability sampling.
Parameters:
- rate: float, sampling rate between 0.0 and 1.0
"""
def __init__(self, rate=None): ...Status and Error Handling
gRPC-compatible status codes for representing span completion states and error conditions.
class Status:
"""
gRPC-compatible span completion status.
Parameters:
- code: int, status code (0=OK, non-zero=error)
- message: str, status message
- details: str, additional status details
"""
def __init__(self, code, message=None, details=None): ...
@property
def canonical_code(self):
"""int: Canonical status code"""
@property
def description(self):
"""str: Status description"""
@property
def is_ok(self):
"""bool: True if status indicates success"""
def format_status_json(self):
"""dict: Status as JSON-serializable dictionary"""
@classmethod
def from_exception(cls, exc):
"""
Create status from Python exception.
Parameters:
- exc: Exception, Python exception object
Returns:
Status: Status representing the exception
"""
@classmethod
def as_ok(cls):
"""
Create OK status.
Returns:
Status: Success status
"""Bounded Collections
Memory-efficient collections for span data with automatic size limits to prevent memory issues in long-running traces.
class BoundedList:
"""
Append-only list with fixed maximum size.
Parameters:
- maxlen: int, maximum number of items
"""
def __init__(self, maxlen): ...
def append(self, item):
"""Add item, dropping oldest if at capacity."""
def extend(self, items):
"""Add multiple items, maintaining size limit."""
class BoundedDict:
"""
Dictionary with fixed maximum capacity.
Parameters:
- maxlen: int, maximum number of key-value pairs
"""
def __init__(self, maxlen): ...
def __setitem__(self, key, value):
"""Set value, removing oldest entries if at capacity."""
def __getitem__(self, key):
"""Get value by key."""Usage Examples
Basic Tracing
from opencensus.trace.tracer import Tracer
from opencensus.trace.samplers import ProbabilitySampler
from opencensus.trace import print_exporter
# Create tracer with 50% sampling
tracer = Tracer(
sampler=ProbabilitySampler(rate=0.5),
exporter=print_exporter.PrintExporter()
)
# Use span as context manager
with tracer.span('database_query') as span:
span.add_attribute('query', 'SELECT * FROM users')
span.add_attribute('rows_returned', 42)
# Simulate database work
result = query_database()
span.add_annotation('Query completed successfully')
span.set_status(Status.as_ok())Manual Span Management
from opencensus.trace.tracer import Tracer
from opencensus.trace.span import Span
tracer = Tracer()
# Create and start span manually
span = tracer.span('http_request')
span.start()
try:
span.add_attribute('http.method', 'GET')
span.add_attribute('http.url', 'https://api.example.com/users')
# Make HTTP request
response = make_request()
span.add_attribute('http.status_code', response.status_code)
span.set_status(Status.as_ok())
except Exception as e:
span.set_status(Status.from_exception(e))
raise
finally:
span.finish()Nested Spans
from opencensus.trace.tracer import Tracer
tracer = Tracer()
with tracer.span('process_order') as parent_span:
parent_span.add_attribute('order_id', '12345')
# Child span for validation
with tracer.span('validate_order') as child_span:
child_span.add_attribute('validation_rules', 5)
validate_order_data()
# Child span for payment
with tracer.span('process_payment') as child_span:
child_span.add_attribute('payment_method', 'credit_card')
process_payment()
parent_span.add_annotation('Order processing completed')Install with Tessl CLI
npx tessl i tessl/pypi-opencensus