Skip to content

Tracing API Reference

Optional OpenTelemetry tracing integration (istos[otel]).

OpenTelemetry tracing integration (optional dependency).

TracingMiddleware

Creates an OpenTelemetry span per request, linked across hops.

Continues the caller's trace when the request carried a W3C traceparent (see :class:~istos.context.RequestEnvelope), and writes this hop's active traceparent back onto the request context so any outbound query/publish propagates it to the next hop — end-to-end distributed tracing over Zenoh.

Source code in src/istos/observability/tracing.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class TracingMiddleware:
    """Creates an OpenTelemetry span per request, linked across hops.

    Continues the caller's trace when the request carried a W3C ``traceparent``
    (see :class:`~istos.context.RequestEnvelope`), and writes this hop's active
    ``traceparent`` back onto the request context so any outbound query/publish
    propagates it to the next hop — end-to-end distributed tracing over Zenoh.
    """

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        if _tracer is None:
            return await call_next(scope)

        from opentelemetry.trace.propagation.tracecontext import (
            TraceContextTextMapPropagator,
        )

        from istos.context import get_request_context

        propagator = TraceContextTextMapPropagator()
        incoming = scope.context.traceparent
        parent = propagator.extract({"traceparent": incoming}) if incoming else None

        with _tracer.start_as_current_span(
            f"istos.{scope.operation}",
            context=parent,
            attributes={
                "istos.prefix": scope.prefix,
                "istos.operation": scope.operation,
                "istos.correlation_id": scope.context.correlation_id,
            },
        ):
            # Publish this span as the traceparent for downstream hops.
            carrier: dict[str, str] = {}
            propagator.inject(carrier)
            tp = carrier.get("traceparent")
            if tp:
                scope.context.traceparent = tp
                get_request_context().traceparent = tp
            return await call_next(scope)

configure_tracing(service_name='istos', endpoint=None)

Configure OpenTelemetry tracing. Returns True if configured, False if opentelemetry is not installed.

Source code in src/istos/observability/tracing.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def configure_tracing(
    service_name: str = "istos",
    endpoint: Optional[str] = None,
) -> bool:
    """
    Configure OpenTelemetry tracing. Returns True if configured, False if
    opentelemetry is not installed.
    """
    global _tracer
    try:
        from opentelemetry import trace
        from opentelemetry.sdk.trace import TracerProvider
        from opentelemetry.sdk.resources import Resource
        from opentelemetry.sdk.trace.export import BatchSpanProcessor
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
    except ImportError:
        return False

    resource = Resource.create({"service.name": service_name})
    provider = TracerProvider(resource=resource)
    if endpoint:
        exporter = OTLPSpanExporter(endpoint=endpoint)
        provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    _tracer = trace.get_tracer("istos")
    return True