Context Propagation

This section contains snippets demonstrating OpenTelemetry context propagation in Python, both within the same process (intra-process) and across different processes (inter-process).

Configuring environment

Development Environment

You can run otel-tui as OpenTelemetry Collector, which acts as a terminal OpenTelemetry viewer

docker run -p 4317:4317 -p 4318:4318 --rm -it --name otel-tui ymtdzzz/otel-tui:latest

Install packages

Create the virtual environment and install the dependencies:

uv venv && source .venv/bin/activate

requirements.txt

opentelemetry-api
opentelemetry-sdk
opentelemetry-exporter-otlp-proto-grpc
requests
flask

Quick install with uv

uv pip install -r https://emdneto.github.io/opentelemetry-by-example/python/context-propagation/requirements.txt

Propagate API

Intra-ProcessInter-Process

import time
from typing import Any

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,
)

resource = Resource.create(
    {"service.name": "context-propagation-intraprocess"}
)
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
    )
)

tracer = trace.get_tracer(__name__)


def fetch_data(item_id: str) -> dict[str, Any]:
    with tracer.start_as_current_span("fetch_data") as span:
        span.set_attribute("item.id", item_id)
        time.sleep(0.1)
        with tracer.start_as_current_span("fetch_data_inner") as inner_span:
            inner_span.set_attribute("inner.operation", "fetching")
            time.sleep(0.05)
        return {"id": item_id, "data": "example"}


def validate_data(data: dict[str, Any]) -> bool:
    with tracer.start_as_current_span("validate_data") as span:
        print(data.get("id"))
        span.set_attribute("validation.result", True)
        time.sleep(0.05)
        with tracer.start_as_current_span("validate_data_inner") as inner_span:
            inner_span.set_attribute("inner.operation", "validating")
            time.sleep(0.02)
        return True


def process_request(request_id: str) -> dict[str, Any]:
    """Parent function - creates root span"""
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("request.id", request_id)

        # These calls automatically create child spans
        data = fetch_data(request_id)
        is_valid = validate_data(data)

        if is_valid:
            span.set_attribute("status", "success")
            return {"status": "processed", "data": data}
        else:
            span.set_attribute("status", "failed")
            return {"status": "failed"}


process_request("req-001")
process_request("req-002")

Run this Python snippet

uv run https://emdneto.github.io/opentelemetry-by-example/python/context-propagation/snippet_intraprocess.py

import time
import threading
from typing import Any

import requests
from flask import Flask, jsonify, request
from opentelemetry import trace, propagate
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,
)

resource = Resource.create(
    {"service.name": "context-propagation-interprocess"}
)
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
    )
)

tracer = trace.get_tracer(__name__)

app = Flask(__name__)


@app.route("/api/data", methods=["GET"])
def get_data():
    # Extract context from incoming request headers
    ctx = propagate.extract(request.headers)

    # Start span with extracted context as parent
    with tracer.start_as_current_span("server_operation", context=ctx) as span:
        span.set_attribute("server.endpoint", "/api/data")
        span.add_event("Processing request on server")

        result = {"message": "Hello from server", "data": [1, 2, 3]}

        span.add_event("Request processed successfully")
        return jsonify(result)


def start_server():
    app.run(host="localhost", port=5001, debug=False, use_reloader=False)


def make_http_request() -> dict[str, Any]:
    with tracer.start_as_current_span("client_operation") as span:
        span.set_attribute("client.operation", "fetch_data")
        span.add_event("Making HTTP request")

        headers = {"Content-Type": "application/json"}

        # Inject current span context into headers
        propagate.inject(headers)

        try:
            # Make HTTP request with propagated context
            response = requests.get(
                "http://localhost:5001/api/data", headers=headers
            )

            span.set_attribute("http.status_code", response.status_code)
            span.add_event("HTTP request completed")

            return response.json()
        except Exception as e:
            span.add_event("HTTP request failed", {"error": str(e)})
            return {"error": str(e)}


server_thread = threading.Thread(target=start_server, daemon=True)
server_thread.start()
time.sleep(2)  # Wait for server to start
make_http_request()

Run this Python snippet

uv run https://emdneto.github.io/opentelemetry-by-example/python/context-propagation/snippet_interprocess.py

TraceContextTextMapPropagator

TODO

In Development

W3CBaggagePropagator

TODO

In Development

B3

TODO

In Development

Jaeger

TODO

In Development