from opentelemetry import trace
from contextlib import contextmanager
[docs]
class TraceUtility:
"""
A small utility wrapper around OpenTelemetry tracing, modeled after MetricUtility.
Usage:
# Explicit start/end
utility = TraceUtility("my_tracer")
span = utility.create_span("operation", attributes={"deploymentId": "dep-123", "region": "us-ashburn-1"})
utility.add_event(span, "started", attributes={"phase": "begin"})
# ... do work ...
utility.end_span(span)
# Context manager that attaches to current trace
with utility.start_as_current_span("operation", attributes={"deploymentId": "dep-123", "region": "us-ashburn-1"}) as span:
utility.add_event(span, "started", attributes={"phase": "begin"})
# ... do work ...
"""
def __init__(self, tracer_name: str):
self.tracer = trace.get_tracer(tracer_name)
def _sanitize_attributes(self, attributes: dict) -> dict:
"""
Prepare attributes for OTLP export:
- drop None values (OTLP rejects None)
- allow scalars: str, bool, int, float
- for lists/tuples, keep only supported scalar items
- fallback to str(v) for unsupported objects
"""
allowed_scalars = (str, bool, int, float)
sanitized = {}
for k, v in attributes.items():
if v is None:
continue
try:
if isinstance(v, (list, tuple)):
filtered = [item for item in v if isinstance(item, allowed_scalars)]
if filtered:
type_set = {type(item) for item in filtered}
if len(type_set) > 1:
# OTel arrays must be homogeneous types; coerce mixed types to strings
sanitized[k] = [str(item) for item in filtered]
else:
sanitized[k] = filtered
continue
if isinstance(v, allowed_scalars):
sanitized[k] = v
else:
sanitized[k] = str(v)
except Exception:
# skip problematic attributes entirely
continue
return sanitized
def _set_attributes(self, span, attributes: dict):
if not attributes:
return
sanitized = self._sanitize_attributes(attributes)
for k, v in sanitized.items():
try:
span.set_attribute(k, v)
except Exception:
# best-effort only
pass
[docs]
def create_span(self, name: str, attributes: dict | None = None):
"""
Create (start) a span with the given name and optional attributes.
The caller is responsible for ending the span (utility.end_span(span) or span.end()).
"""
span = self.tracer.start_span(name)
if attributes:
self._set_attributes(span, attributes)
return span
[docs]
@contextmanager
def start_as_current_span(self, name: str, attributes: dict | None = None):
"""
Start a span as the current span (context manager).
This guarantees that if a parent span is active (e.g., from LangGraph/LangChain instrumentation),
the new span will be a child of that active span, appending to the current trace.
"""
with self.tracer.start_as_current_span(name) as span:
if attributes:
self._set_attributes(span, attributes)
yield span
[docs]
def add_event(self, span, name: str, attributes: dict | None = None):
"""
Add an event to a given span with optional attributes.
"""
if attributes:
span.add_event(name, attributes=self._sanitize_attributes(attributes))
else:
span.add_event(name)
[docs]
def end_span(self, span):
"""
End the given span safely; avoid double-ending when used with context managers.
"""
try:
# If the span is already ended, most SDKs report is_recording() as False
if hasattr(span, "is_recording") and not span.is_recording():
return
except Exception:
pass
try:
span.end()
except Exception:
pass
# Example usage
if __name__ == "__main__":
# NOTE: This example assumes a tracer provider has been configured elsewhere.
utility = TraceUtility("test_tracer")
span = utility.create_span("test_span", attributes={"deploymentId": "dep-123", "region": "us-ashburn-1"})
utility.add_event(span, "example_event", attributes={"step": 1})
utility.end_span(span)
# Context manager example (will attach as a child if a current span exists)
with utility.start_as_current_span("test_span_ctx", attributes={"deploymentId": "dep-123", "region": "us-ashburn-1"}) as s:
# minimal body to avoid syntax errors during import-time parsing
pass