or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

index.md
tile.json

tessl/pypi-opentelemetry-propagator-b3

OpenTelemetry B3 Propagator for distributed tracing context propagation using B3 headers

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/opentelemetry-propagator-b3@1.36.x

To install, run

npx @tessl/cli install tessl/pypi-opentelemetry-propagator-b3@1.36.0

index.mddocs/

OpenTelemetry B3 Propagator

A Python library that provides propagator implementations for the B3 format used in distributed tracing systems, particularly with Zipkin. It enables trace context propagation across service boundaries using B3 headers, supporting both single-header (b3) and multi-header (b3multi) formats.

Package Information

  • Package Name: opentelemetry-propagator-b3
  • Language: Python
  • Installation: pip install opentelemetry-propagator-b3
  • Minimum Python Version: 3.9+

Core Imports

from opentelemetry.propagators.b3 import B3MultiFormat, B3SingleFormat

For backward compatibility (deprecated):

from opentelemetry.propagators.b3 import B3Format  # Deprecated since 1.2.0

Basic Usage

from opentelemetry.propagators.b3 import B3MultiFormat, B3SingleFormat
from opentelemetry import trace

# Configure B3 multi-header propagation
propagator = B3MultiFormat()

# Or configure B3 single-header propagation  
propagator = B3SingleFormat()

# Example: Extract trace context from incoming HTTP headers
propagator = B3MultiFormat()
headers = {
    "x-b3-traceid": "80f198ee56343ba864fe8b2a57d3eff7",
    "x-b3-spanid": "e457b5a2e4d86bd1",
    "x-b3-sampled": "1"
}

# Create context from headers
context = propagator.extract(headers)

# Use context for span creation
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("process-request", context=context):
    # Your application logic here
    pass

# Example: Inject trace context into outgoing HTTP headers
outgoing_headers = {}
propagator.inject(outgoing_headers)
# Headers now contain B3 trace context for downstream services

Capabilities

B3 Multi-Header Format Propagation

Implements the B3 multi-header propagation format using separate HTTP headers for each trace context component (trace ID, span ID, sampling decision, and flags).

class B3MultiFormat(TextMapPropagator):
    """Propagator for the B3 HTTP multi-header format."""
    
    def extract(
        self,
        carrier: CarrierT,
        context: typing.Optional[Context] = None,
        getter: Getter = default_getter,
    ) -> Context:
        """
        Extract trace context from carrier using B3 multi-header format.
        
        Parameters:
        - carrier: Dict-like object containing headers/metadata
        - context: Optional existing context to build upon
        - getter: Function to retrieve values from carrier
        
        Returns:
        Context object with trace information
        """
    
    def inject(
        self,
        carrier: CarrierT,
        context: typing.Optional[Context] = None,
        setter: Setter = default_setter,
    ) -> None:
        """
        Inject trace context into carrier using B3 multi-header format.
        
        Parameters:
        - carrier: Dict-like object to inject headers into
        - context: Optional context containing trace information
        - setter: Function to set values in carrier
        """
    
    @property
    def fields(self) -> typing.Set[str]:
        """
        Returns the set of header fields used by this propagator.
        
        Returns:
        Set containing: "x-b3-traceid", "x-b3-spanid", "x-b3-sampled"
        """

Header Constants:

TRACE_ID_KEY = "x-b3-traceid"      # Trace identifier header
SPAN_ID_KEY = "x-b3-spanid"        # Span identifier header  
SAMPLED_KEY = "x-b3-sampled"       # Sampling decision header
FLAGS_KEY = "x-b3-flags"           # Additional flags header
SINGLE_HEADER_KEY = "b3"           # Single header key (used by single format)

B3 Single-Header Format Propagation

Implements the B3 single-header propagation format using a single combined HTTP header containing all trace context information.

class B3SingleFormat(B3MultiFormat):
    """Propagator for the B3 HTTP single-header format."""
    
    def inject(
        self,
        carrier: CarrierT,
        context: typing.Optional[Context] = None,
        setter: Setter = default_setter,
    ) -> None:
        """
        Inject trace context into carrier using B3 single-header format.
        
        Parameters:
        - carrier: Dict-like object to inject headers into
        - context: Optional context containing trace information
        - setter: Function to set values in carrier
        
        Note: Uses single "b3" header with format: traceid-spanid-sampled
        """
    
    @property
    def fields(self) -> typing.Set[str]:
        """
        Returns the set of header fields used by this propagator.
        
        Returns:
        Set containing: "b3"
        """

Note: B3SingleFormat inherits the extract() method from B3MultiFormat, which handles both single and multi-header extraction automatically.

Deprecated B3 Format (Legacy Support)

Provides backward compatibility for applications using the original B3Format class name.

class B3Format(B3MultiFormat):
    """
    Deprecated alias for B3MultiFormat.
    
    Deprecated since version 1.2.0. Use B3MultiFormat instead.
    """
    
    def __init__(self, *args, **kwargs):
        """Initialize with deprecation warning."""

Types and Dependencies

Required Dependencies

# From opentelemetry-api
from opentelemetry.context import Context
from opentelemetry.propagators.textmap import (
    CarrierT,           # Type alias for carrier objects (dict-like)
    Getter,             # Type for value getter functions
    Setter,             # Type for value setter functions  
    TextMapPropagator,  # Base class for text map propagators
    default_getter,     # Default getter implementation
    default_setter,     # Default setter implementation
)

# From typing/typing-extensions
import typing
from typing_extensions import deprecated

Entry Point Registration

The package automatically registers propagators via entry points:

# Available via opentelemetry_propagator entry point group
"b3" = "opentelemetry.propagators.b3:B3SingleFormat"
"b3multi" = "opentelemetry.propagators.b3:B3MultiFormat"

Version Information

from opentelemetry.propagators.b3.version import __version__
# __version__ = "1.36.0"

Error Handling

The propagators are designed to be robust and handle malformed or missing headers gracefully:

  • Invalid trace/span IDs: Returns the original context unchanged
  • Missing headers: Uses default invalid trace/span ID values
  • Malformed header values: Validates using regex patterns and ignores invalid values
  • Mixed header formats: Single header takes precedence over multi-header when both present

Integration Patterns

Global Configuration

# Direct usage - most common pattern
from opentelemetry.propagators.b3 import B3MultiFormat

propagator = B3MultiFormat()
# Use propagator for extract/inject operations

Custom Carrier Handling

from opentelemetry.propagators.b3 import B3MultiFormat

class CustomGetter:
    def get(self, carrier, key):
        # Custom logic to extract values from carrier
        return carrier.get(key.lower())  # Case-insensitive lookup

class CustomSetter:
    def set(self, carrier, key, value):
        # Custom logic to set values in carrier
        carrier[key.lower()] = value

propagator = B3MultiFormat()
context = propagator.extract(
    carrier=custom_headers,
    getter=CustomGetter()
)

Framework Integration

# Flask example
from flask import request, g
from opentelemetry.propagators.b3 import B3MultiFormat

@app.before_request
def extract_trace_context():
    propagator = B3MultiFormat()
    g.trace_context = propagator.extract(dict(request.headers))

# Requests example  
import requests
from opentelemetry.propagators.b3 import B3SingleFormat

propagator = B3SingleFormat()
headers = {}
propagator.inject(headers)
response = requests.get("http://example.com", headers=headers)