OpenTelemetry Trace Context Propagation [Python]

This guide covers Python-specific implementation of context propagation. For a comprehensive overview of context propagation concepts, W3C TraceContext, propagators, and troubleshooting, see the OpenTelemetry Context Propagation guide.

Automatic propagation

OpenTelemetry Python handles traceparent headers automatically in most scenarios. When using auto-instrumentation libraries, HTTP client libraries automatically inject traceparent headers into outgoing requests, and server libraries automatically extract them from incoming requests.

Auto-instrumentation

bash

# Enable auto-instrumentation
export OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED=true
export OTEL_SERVICE_NAME=my-service
opentelemetry-instrument python app.py

Auto-instrumentation packages for HTTP clients (requests, urllib3, httpx) and web frameworks (Flask, Django, FastAPI) automatically inject W3C tracecontext headers to outgoing HTTP requests and extract them from incoming requests.

Manual propagation

When automatic instrumentation is not available, you can manually handle traceparent headers using OpenTelemetry's propagation API.

Extracting context

python

from opentelemetry import trace
from opentelemetry.propagate import extract
from opentelemetry.trace import SpanKind

# Extract context from incoming request headers
# Headers should be in dict format: {"traceparent": "00-..."}
headers = {"traceparent": request.headers.get("traceparent")}
context = extract(headers)

# Create a new span with the extracted parent context
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span(
    f"HTTP {request.method}",
    context=context,
    kind=SpanKind.SERVER
) as span:
    # Your business logic here
    process_request(request)

Injecting context

python

from opentelemetry import trace
from opentelemetry.propagate import inject
import requests

# Create carrier dict for headers
headers = {}

# Inject current trace context into headers
inject(headers)

# Make HTTP request with injected headers
response = requests.get("http://example.com/api", headers=headers)

Debugging propagation

Logging context

Log incoming traceparent headers and current span context for debugging:

python

import logging
from opentelemetry import trace
from opentelemetry.trace import get_current_span

logger = logging.getLogger(__name__)

def log_trace_context(request_headers):
    # Log incoming traceparent header
    traceparent = request_headers.get("traceparent")
    if traceparent:
        logger.info(f"Incoming traceparent: {traceparent}")
    else:
        logger.info("No traceparent header found")

    # Log current span context
    current_span = get_current_span()
    if current_span and current_span.get_span_context().is_valid():
        context = current_span.get_span_context()
        logger.info(f"Current trace context - TraceId: {context.trace_id:032x}, "
                   f"SpanId: {context.span_id:016x}, "
                   f"Sampled: {context.trace_flags.sampled}")
    else:
        logger.info("No valid span context found")

Validating format

Validate traceparent headers to ensure they follow the W3C specification:

python

import re
from typing import Dict, Optional

class TraceparentValidator:
    TRACEPARENT_PATTERN = re.compile(r'^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$')

    @classmethod
    def is_valid_traceparent(cls, traceparent: str) -> bool:
        return bool(traceparent and cls.TRACEPARENT_PATTERN.match(traceparent))

    @classmethod
    def parse_traceparent(cls, traceparent: str) -> Optional[Dict[str, str]]:
        if not cls.is_valid_traceparent(traceparent):
            raise ValueError(f"Invalid traceparent format: {traceparent}")

        parts = traceparent.split('-')
        return {
            'version': parts[0],
            'trace_id': parts[1],
            'span_id': parts[2],
            'flags': parts[3],
            'is_sampled': parts[3] == '01'
        }

# Usage
validator = TraceparentValidator()
traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"

if validator.is_valid_traceparent(traceparent):
    parsed = validator.parse_traceparent(traceparent)
    print(f"Valid traceparent: {parsed}")
else:
    print("Invalid traceparent format")

Getting trace info

Access current trace context information:

python

from opentelemetry import trace
from opentelemetry.trace import get_current_span
from typing import Dict

def get_trace_info() -> Dict[str, str]:
    current_span = get_current_span()

    if not current_span or not current_span.get_span_context().is_valid():
        return {"error": "No valid span context available"}

    context = current_span.get_span_context()

    return {
        "trace_id": f"{context.trace_id:032x}",
        "span_id": f"{context.span_id:016x}",
        "is_sampled": str(context.trace_flags.sampled),
        "is_remote": str(context.is_remote),
        "trace_state": str(context.trace_state) if context.trace_state else "None"
    }

# Usage in your application
def my_handler():
    trace_info = get_trace_info()
    print(f"Current trace info: {trace_info}")

    # Include trace info in logs
    logger.info("Processing request", extra=trace_info)

Custom propagators

You can create custom propagators for non-standard trace context formats:

python

from typing import Dict, Optional
from opentelemetry.propagators import textmap
from opentelemetry.trace import SpanContext, TraceFlags, set_span_in_context
from opentelemetry.context import Context

class CustomPropagator(textmap.TextMapPropagator):
    def extract(self, carrier: Dict[str, str], context: Optional[Context] = None) -> Context:
        # Extract custom headers
        custom_trace_id = carrier.get("x-custom-trace-id")
        custom_span_id = carrier.get("x-custom-span-id")

        if custom_trace_id and custom_span_id:
            # Convert custom format to OpenTelemetry format
            try:
                trace_id = int(custom_trace_id, 16)
                span_id = int(custom_span_id, 16)

                span_context = SpanContext(
                    trace_id=trace_id,
                    span_id=span_id,
                    is_remote=True,
                    trace_flags=TraceFlags(0x01)  # Sampled
                )

                from opentelemetry.trace import NonRecordingSpan
                return set_span_in_context(
                    NonRecordingSpan(span_context),
                    context
                )
            except ValueError:
                # Invalid format, return original context
                pass

        return context or Context()

    def inject(self, carrier: Dict[str, str], context: Optional[Context] = None) -> None:
        from opentelemetry.trace import get_current_span

        span = get_current_span(context)
        if span and span.get_span_context().is_valid():
            span_context = span.get_span_context()
            carrier["x-custom-trace-id"] = f"{span_context.trace_id:032x}"
            carrier["x-custom-span-id"] = f"{span_context.span_id:016x}"

    @property
    def fields(self) -> set:
        return {"x-custom-trace-id", "x-custom-span-id"}

# Register custom propagator
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

set_global_textmap(CompositePropagator([
    TraceContextTextMapPropagator(),
    CustomPropagator(),
]))

Configuration

Environment variables

Configure propagation format using environment variables:

bash

# Default: W3C TraceContext and Baggage
export OTEL_PROPAGATORS="tracecontext,baggage"

# B3 format (requires opentelemetry-propagator-b3)
export OTEL_PROPAGATORS="b3"

# Multiple formats
export OTEL_PROPAGATORS="tracecontext,b3,baggage"

Programmatic configuration

python

from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator

# Set custom propagator combination
set_global_textmap(CompositePropagator([
    TraceContextTextMapPropagator(),
    W3CBaggagePropagator(),
]))

Production considerations

Performance

Propagation has minimal performance impact:

Header extraction/injection is fast
Context objects are lightweight
No network calls during propagation

Security

Be aware of potential security implications:

Trace IDs can be used to correlate requests
Don't log sensitive information in trace context
Consider network security when propagating across service boundaries

Monitoring propagation

Monitor propagation health in production:

python

import time
from collections import defaultdict
from opentelemetry import trace
from opentelemetry.trace import get_current_span

class PropagationMonitor:
    def __init__(self):
        self.stats = defaultdict(int)
        self.start_time = time.time()

    def record_request(self, has_parent: bool, is_sampled: bool):
        self.stats['total_requests'] += 1
        if has_parent:
            self.stats['requests_with_parent'] += 1
        if is_sampled:
            self.stats['sampled_requests'] += 1

    def get_stats(self):
        elapsed = time.time() - self.start_time
        total = self.stats['total_requests']

        return {
            'total_requests': total,
            'requests_with_parent': self.stats['requests_with_parent'],
            'propagation_rate': self.stats['requests_with_parent'] / total if total > 0 else 0,
            'sampling_rate': self.stats['sampled_requests'] / total if total > 0 else 0,
            'elapsed_seconds': elapsed
        }

# Usage in request handler
monitor = PropagationMonitor()

def handle_request(request):
    current_span = get_current_span()
    has_parent = current_span.get_span_context().is_remote if current_span else False
    is_sampled = current_span.get_span_context().trace_flags.sampled if current_span else False

    monitor.record_request(has_parent, is_sampled)

    # Process request...