learnNote

主题

OpenTelemetry Context 传播机制 深入学习笔记 ​

对应学习计划阶段 5，预计学习时间 2-3 天

目录 ​

  1. 为什么需要 Context 传播
  2. Context 核心概念
  3. Context API 详解
  4. Propagator 传播器
  5. W3C TraceContext 标准
  6. W3C Baggage 机制
  7. 进程内 Context 传播
  8. 跨进程 Context 传播
  9. 异步场景下的 Context 传播
  10. 自定义 Propagator
  11. 实践 Demo
  12. 常见问题 QA
  13. 学习检查清单

1. 为什么需要 Context 传播 ​

1.1 分布式系统的核心挑战 ​

在分布式系统中，一次用户请求会流经多个服务。如果每个服务独立产生 Trace 数据，我们将无法将它们关联起来：

没有 Context 传播：

用户请求
├── [Service A] trace_id=aaa  ← 独立的 Trace
├── [Service B] trace_id=bbb  ← 独立的 Trace
└── [Service C] trace_id=ccc  ← 独立的 Trace

三条完全无关的 Trace，无法拼出完整调用链。

有 Context 传播：

用户请求
├── [Service A] trace_id=xxx, span_id=001 (root)
│   inject → HTTP Header: traceparent=00-xxx-001-01
├── [Service B] trace_id=xxx, span_id=002, parent=001
│   inject → HTTP Header: traceparent=00-xxx-002-01
└── [Service C] trace_id=xxx, span_id=003, parent=002

一条完整的 Trace，清晰展示调用链路。

1.2 Context 传播解决的问题 ​

问题Context 传播如何解决
跨服务追踪断裂通过 HTTP Header 传递 trace_id 和 span_id
不知道谁调用了谁parent_span_id 建立父子关系
跨服务传递业务数据Baggage 机制传递 key-value 对
多种传播格式共存CompositeTextMapPropagator 支持多种格式
进程内异步操作丢失上下文Context attach/detach 机制管理线程/协程上下文

1.3 传播发生在哪里？ ​

┌─────────────────────────────────────────────────────┐
│  进程内传播 (In-Process Propagation)                  │
│                                                      │
│  Thread/Coroutine 之间通过 Context 对象传递            │
│  使用 attach() / detach() 管理当前 Context             │
└────────────┬────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────┐
│  跨进程传播 (Inter-Process Propagation)               │
│                                                      │
│  通过 Propagator 将 Context 序列化到载体中              │
│  载体：HTTP Headers, gRPC Metadata, 消息队列 Header    │
│  使用 inject() / extract() 完成序列化/反序列化          │
└─────────────────────────────────────────────────────┘

2. Context 核心概念 ​

2.1 什么是 Context？ ​

Context 是 OpenTelemetry 中传递请求级别数据的核心机制。它是一个**不可变（immutable）**的键值对容器，存储当前执行环境中的追踪信息和业务数据。

python
from opentelemetry import context

# Context 的本质是一个不可变的 key-value 映射
# 每次修改都会创建一个新的 Context 对象（类似 Python 的 frozenset）

核心特征：

特征说明
不可变修改操作返回新对象，原 Context 不变
隐式传递通过 attach() 绑定到当前执行上下文，无需显式传参
线程/协程安全每个线程/协程有独立的 current Context
键值对存储使用 context.create_key() 创建的 Key 来存取数据

2.2 Context 的层次结构 ​

┌─────────────────────────────────────────────────┐
│                  Context                         │
│                                                  │
│  ┌─────────────────────────────────────────┐    │
│  │  Span Context (trace_id, span_id, ...)   │    │
│  │  → 由 Tracer 自动管理                     │    │
│  └─────────────────────────────────────────┘    │
│                                                  │
│  ┌─────────────────────────────────────────┐    │
│  │  Baggage (user.id, tenant.id, ...)       │    │
│  │  → 由应用代码手动管理                      │    │
│  └─────────────────────────────────────────┘    │
│                                                  │
│  ┌─────────────────────────────────────────┐    │
│  │  其他自定义数据                            │    │
│  │  → 通过 create_key() 自定义               │    │
│  └─────────────────────────────────────────┘    │
└─────────────────────────────────────────────────┘

2.3 Context 与 Span 的关系 ​

python
from opentelemetry import trace, context

tracer = trace.get_tracer("demo")

# start_as_current_span 做了三件事：
# 1. 创建一个新 Span
# 2. 将 Span 放入新的 Context
# 3. 将新 Context attach 到当前线程
with tracer.start_as_current_span("my-operation") as span:
    # 此时 current context 中包含了 span
    current_span = trace.get_current_span()
    assert current_span == span

    # 嵌套 span 自动成为子 span
    with tracer.start_as_current_span("child-operation") as child:
        # child 的 parent_span_id == span 的 span_id
        pass

3. Context API 详解 ​

3.1 核心 API 一览 ​

python
from opentelemetry import context

# 1. 创建 Context Key
my_key = context.create_key("my-data-key")

# 2. 获取当前 Context
ctx = context.get_current()

# 3. 在 Context 中设置值（返回新 Context）
new_ctx = context.set_value(my_key, "my-value")
new_ctx = context.set_value(my_key, "my-value", parent=ctx)  # 指定父 Context

# 4. 从 Context 中获取值
value = context.get_value(my_key)         # 从当前 Context 获取
value = context.get_value(my_key, ctx)    # 从指定 Context 获取

# 5. 将 Context 设为当前（attach）
token = context.attach(new_ctx)

# 6. 恢复之前的 Context（detach）
context.detach(token)

3.2 attach 和 detach 的工作原理 ​

attach() 和 detach() 是 Context 管理的核心。它们实现了一个栈式结构来管理当前 Context：

python
from opentelemetry import context

my_key = context.create_key("demo")

# 初始状态：current context 是空的 ROOT context
print(context.get_value(my_key))  # None

# === 第 1 层 ===
ctx1 = context.set_value(my_key, "value-1")
token1 = context.attach(ctx1)
print(context.get_value(my_key))  # "value-1"

# === 第 2 层（嵌套）===
ctx2 = context.set_value(my_key, "value-2")
token2 = context.attach(ctx2)
print(context.get_value(my_key))  # "value-2"

# === 恢复第 1 层 ===
context.detach(token2)
print(context.get_value(my_key))  # "value-1"

# === 恢复初始状态 ===
context.detach(token1)
print(context.get_value(my_key))  # None

图解 Context 栈：

attach(ctx1)     attach(ctx2)     detach(token2)   detach(token1)
    │                │                │                │
    ▼                ▼                ▼                ▼
┌────────┐     ┌────────┐      ┌────────┐       ┌────────┐
│  ctx2  │ ←── │  ctx2  │ pop  │        │       │        │
├────────┤     ├────────┤  ──→ ├────────┤       ├────────┤
│  ctx1  │     │  ctx1  │      │  ctx1  │ pop   │        │
├────────┤     ├────────┤      ├────────┤  ──→  ├────────┤
│  ROOT  │     │  ROOT  │      │  ROOT  │       │  ROOT  │
└────────┘     └────────┘      └────────┘       └────────┘

3.3 Token 机制防止 Context 泄漏 ​

attach() 返回的 token 不仅仅是一个标识符，它还携带了"之前的 Context"信息，确保 detach 时能正确恢复：

python
from opentelemetry import context

my_key = context.create_key("demo")

def process_request():
    ctx = context.set_value(my_key, "request-data")
    token = context.attach(ctx)
    try:
        # 业务逻辑
        handle_business_logic()
    finally:
        # ⚠️ 必须在 finally 中 detach，否则 Context 泄漏！
        context.detach(token)

# ❌ 错误示范：忘记 detach
def bad_example():
    ctx = context.set_value(my_key, "data")
    context.attach(ctx)
    # 如果这里抛异常，Context 永远不会被恢复
    do_something_risky()

# ❌ 错误示范：detach 顺序错误
def wrong_order():
    ctx1 = context.set_value(my_key, "v1")
    token1 = context.attach(ctx1)

    ctx2 = context.set_value(my_key, "v2")
    token2 = context.attach(ctx2)

    # 应该先 detach token2，再 detach token1
    context.detach(token1)  # ⚠️ 会打印警告日志
    context.detach(token2)

3.4 为什么 Span 的 with 语句不需要手动 attach/detach？ ​

python
# 使用 with 语句时，Span 自动管理 Context
with tracer.start_as_current_span("my-span") as span:
    # 内部实现等价于：
    # ctx = trace.set_span_in_context(span)
    # token = context.attach(ctx)
    pass
# with 退出时自动 context.detach(token)

# 手动管理时需要自己处理 attach/detach
span = tracer.start_span("my-span")
ctx = trace.set_span_in_context(span)
token = context.attach(ctx)
try:
    # 业务逻辑
    pass
finally:
    context.detach(token)
    span.end()

4. Propagator 传播器 ​

4.1 什么是 Propagator？ ​

Propagator（传播器） 负责在进程间序列化和反序列化 Context。它定义了两个核心操作：

操作方向说明
inject发送端 → 载体将当前 Context 中的信息写入载体（如 HTTP Header）
extract载体 → 接收端从载体中提取信息，重建 Context
┌──────────────┐          HTTP Request           ┌──────────────┐
│   Service A  │ ───────────────────────────────→ │   Service B  │
│              │   Headers:                       │              │
│  inject()    │   traceparent: 00-abc-123-01     │  extract()   │
│  将 Context  │   tracestate: vendor=value       │  从 Headers  │
│  写入 Header │   baggage: user.id=42            │  恢复 Context│
└──────────────┘                                  └──────────────┘

4.2 TextMapPropagator 接口 ​

OpenTelemetry 中最常用的是 TextMapPropagator，用于文本格式的载体（HTTP Headers、gRPC Metadata 等）：

python
from opentelemetry.context.propagation import TextMapPropagator

class TextMapPropagator:
    def inject(
        self,
        carrier,              # 载体（通常是 dict）
        context=None,         # 要注入的 Context（默认当前）
        setter=None           # 自定义的写入方法
    ): ...

    def extract(
        self,
        carrier,              # 载体
        context=None,         # 父 Context（默认当前）
        getter=None           # 自定义的读取方法
    ) -> Context: ...

    @property
    def fields(self) -> set:
        # 返回此 Propagator 会使用的 header 名称
        ...

4.3 内置 Propagator 类型 ​

Propagator功能Header
TraceContextTextMapPropagatorW3C Trace Context 标准traceparent, tracestate
W3CBaggagePropagatorW3C Baggage 标准baggage
B3MultiFormatZipkin B3 格式（多 Header）X-B3-TraceId, X-B3-SpanId, ...
B3SingleFormatZipkin B3 格式（单 Header）b3
JaegerPropagatorJaeger 格式uber-trace-id
CompositeTextMapPropagator组合多个 Propagator取决于组合内容

4.4 全局 Propagator 配置 ​

python
from opentelemetry.propagate import set_global_textmap, get_global_textmap
from opentelemetry.propagators.composite import CompositeTextMapPropagator
from opentelemetry.trace.propagation import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator

# 推荐配置：W3C TraceContext + Baggage
set_global_textmap(
    CompositeTextMapPropagator([
        TraceContextTextMapPropagator(),
        W3CBaggagePropagator(),
    ])
)

# 如果还要兼容 Zipkin B3 格式
from opentelemetry.propagators.b3 import B3MultiFormat

set_global_textmap(
    CompositeTextMapPropagator([
        TraceContextTextMapPropagator(),
        W3CBaggagePropagator(),
        B3MultiFormat(),
    ])
)

# 通过环境变量配置（无需代码修改）
# export OTEL_PROPAGATORS=tracecontext,baggage
# export OTEL_PROPAGATORS=tracecontext,baggage,b3multi

4.5 CompositeTextMapPropagator 的工作原理 ​

inject 时（发送端）：
  Context → [TraceContextPropagator.inject()] → traceparent + tracestate
         → [W3CBaggagePropagator.inject()]   → baggage
         → carrier 同时包含所有 header

extract 时（接收端）：
  carrier → [TraceContextPropagator.extract()] → 提取 trace 信息到 Context
          → [W3CBaggagePropagator.extract()]   → 提取 baggage 到 Context
          → 返回包含所有信息的 Context

⚠️ extract 是链式执行的：
  每个 Propagator 的输出 Context 作为下一个 Propagator 的输入 Context

5. W3C TraceContext 标准 ​

5.1 什么是 W3C TraceContext？ ​

W3C TraceContext 是 W3C 标准化的分布式追踪传播格式（W3C Recommendation），它定义了两个 HTTP Header：

  • traceparent：必需，携带核心追踪标识
  • tracestate：可选，携带厂商特定数据

5.2 traceparent 格式详解 ​

traceparent: {version}-{trace-id}-{parent-id}-{trace-flags}

示例：
traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01
              │  │                                │                  │
              │  │                                │                  └─ trace-flags (8 bit)
              │  │                                │                     01 = sampled
              │  │                                │                     00 = not sampled
              │  │                                └─ parent-id (64 bit, 16 hex chars)
              │  │                                   当前 Span 的 span_id
              │  └─ trace-id (128 bit, 32 hex chars)
              │     整条 Trace 的唯一标识
              └─ version (8 bit)
                 固定为 "00"

各字段说明：

字段长度说明示例
version2 hex (1 byte)版本号，当前固定为 0000
trace-id32 hex (16 bytes)全局唯一的 Trace 标识4bf92f3577b34da6a3ce929d0e0e4736
parent-id16 hex (8 bytes)当前操作的 Span ID00f067aa0ba902b7
trace-flags2 hex (1 byte)控制标志位，bit 0 = sampled01

5.3 tracestate 格式详解 ​

tracestate 携带厂商特定的附加追踪数据，格式为逗号分隔的 key=value 对：

tracestate: vendor1=value1,vendor2=value2

示例：
tracestate: congo=t61rcWkgMzE,rojo=00f067aa0ba902b7

规则：

  • 最左边的 key-value 对优先级最高（代表最近的操作者）
  • 每个厂商应该只有一个条目
  • 最多 32 个条目
  • 总长度不超过 512 字节

5.4 Python 中解析 traceparent ​

python
from opentelemetry.trace.propagation import TraceContextTextMapPropagator
from opentelemetry.propagate import extract
from opentelemetry import trace

propagator = TraceContextTextMapPropagator()

# 模拟接收到的 HTTP 请求头
incoming_headers = {
    "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
    "tracestate": "myvendor=customvalue"
}

# 提取 Context
ctx = propagator.extract(carrier=incoming_headers)

# 从 Context 中获取 SpanContext
span_context = trace.get_current_span(ctx).get_span_context()

print(f"trace_id:    {span_context.trace_id:032x}")
# 4bf92f3577b34da6a3ce929d0e0e4736

print(f"span_id:     {span_context.span_id:016x}")
# 00f067aa0ba902b7

print(f"trace_flags: {span_context.trace_flags:02x}")
# 01 (sampled)

print(f"tracestate:  {span_context.trace_state}")
# TraceState([('myvendor', 'customvalue')])

6. W3C Baggage 机制 ​

6.1 什么是 Baggage？ ​

Baggage 是 OpenTelemetry 提供的跨服务传递业务数据的机制。与 SpanContext（只传递追踪标识）不同，Baggage 允许传递任意 key-value 数据。

Baggage 的典型用途：
├── user.id = "u-12345"          传递用户身份
├── tenant.id = "t-acme"        传递租户信息（多租户系统）
├── request.priority = "high"   传递请求优先级
├── feature.flags = "beta-ui"   传递特性标志
└── region = "us-east-1"        传递区域信息

6.2 Baggage vs Span Attributes ​

特性BaggageSpan Attributes
作用范围跨服务传播仅当前 Span
传播方式通过 HTTP Header（baggage）不传播
大小限制受 HTTP Header 大小限制（通常 8KB）SDK 限制（默认 128 个属性）
可见性所有下游服务都能看到只在当前 Span 的 Exporter 中
安全性⚠️ 明文传播，不要放敏感数据相对安全
性能影响每次请求都携带，有网络开销只在采集时有开销

6.3 Baggage API 使用 ​

python
from opentelemetry import baggage, context

# === 设置 Baggage ===

# 方式 1：设置单个值
ctx = baggage.set_baggage("user.id", "u-12345")
token = context.attach(ctx)

# 方式 2：链式设置多个值
ctx = baggage.set_baggage("user.id", "u-12345")
ctx = baggage.set_baggage("tenant.id", "t-acme", context=ctx)
ctx = baggage.set_baggage("region", "us-east-1", context=ctx)
token = context.attach(ctx)

# === 读取 Baggage ===

# 读取单个值
user_id = baggage.get_baggage("user.id")
print(f"User ID: {user_id}")  # u-12345

# 读取所有 Baggage
all_baggage = baggage.get_all()
for key, value in all_baggage.items():
    print(f"  {key} = {value}")

# === 删除 Baggage ===
ctx = baggage.remove_baggage("user.id")
token2 = context.attach(ctx)

# === 清理 ===
context.detach(token2)
context.detach(token)

6.4 Baggage 的 HTTP 传播格式 ​

baggage: user.id=u-12345,tenant.id=t-acme,region=us-east-1

带属性的 baggage：
baggage: user.id=u-12345;property1=p1,tenant.id=t-acme

格式规则：

  • 多个条目用逗号 , 分隔
  • key 和 value 用等号 = 分隔
  • 条目可以带分号 ; 分隔的属性（metadata）
  • value 需要进行 URL 编码（percent-encoding）

6.5 Baggage 安全注意事项 ​

python
# ⚠️ 安全警告：Baggage 以明文传播，绝不要放敏感数据！

# ❌ 错误：敏感数据
ctx = baggage.set_baggage("user.password", "secret123")      # 绝不要这样做！
ctx = baggage.set_baggage("auth.token", "eyJhbGci...")        # 绝不要这样做！
ctx = baggage.set_baggage("credit.card", "4111-1111-1111")    # 绝不要这样做！

# ✅ 正确：非敏感的业务标识
ctx = baggage.set_baggage("user.id", "u-12345")               # ID 是可以的
ctx = baggage.set_baggage("request.priority", "high")          # 优先级标记
ctx = baggage.set_baggage("deployment.version", "v2.1.0")      # 版本信息

7. 进程内 Context 传播 ​

7.1 隐式传播（推荐） ​

在同一进程内，Context 通过 ContextVar（Python 3.7+）实现隐式传播。start_as_current_span 会自动处理 attach/detach：

python
from opentelemetry import trace

tracer = trace.get_tracer("in-process-demo")

def handle_request():
    with tracer.start_as_current_span("request-handler") as parent:
        # parent span 自动成为当前 Context 的一部分
        validate_input()
        process_data()

def validate_input():
    # 自动继承父 Context，创建的 span 自动成为 child
    with tracer.start_as_current_span("validate-input"):
        pass

def process_data():
    with tracer.start_as_current_span("process-data"):
        save_to_db()

def save_to_db():
    with tracer.start_as_current_span("save-to-db"):
        pass

# 调用后生成的 Trace 结构：
# request-handler
# ├── validate-input
# └── process-data
#     └── save-to-db

7.2 显式传播 ​

某些场景下需要显式传递 Context（如跨线程、回调函数）：

python
from opentelemetry import trace, context

tracer = trace.get_tracer("explicit-demo")

def parent_function():
    with tracer.start_as_current_span("parent"):
        # 捕获当前 Context
        ctx = context.get_current()

        # 将 Context 显式传递给其他函数
        child_function(ctx)

def child_function(parent_ctx):
    # 在指定 Context 下创建 Span
    with tracer.start_as_current_span("child", context=parent_ctx):
        pass

7.3 跨线程 Context 传播 ​

Python 的 ContextVar 默认不会在线程间传播，需要手动处理：

python
import threading
from opentelemetry import trace, context

tracer = trace.get_tracer("thread-demo")

def worker(ctx):
    """在子线程中使用父线程的 Context"""
    token = context.attach(ctx)
    try:
        with tracer.start_as_current_span("worker-task"):
            print(f"Worker span parent: {trace.get_current_span().get_span_context()}")
    finally:
        context.detach(token)

def main():
    with tracer.start_as_current_span("main-task"):
        # 捕获当前 Context
        ctx = context.get_current()

        # 将 Context 传递给子线程
        thread = threading.Thread(target=worker, args=(ctx,))
        thread.start()
        thread.join()

# 结果：worker-task 是 main-task 的子 span

便捷工具：

python
from concurrent.futures import ThreadPoolExecutor
from opentelemetry import trace, context

tracer = trace.get_tracer("pool-demo")

def context_aware_task(ctx, task_id):
    """带 Context 的任务"""
    token = context.attach(ctx)
    try:
        with tracer.start_as_current_span(f"task-{task_id}"):
            pass
    finally:
        context.detach(token)

def main():
    with tracer.start_as_current_span("batch-process"):
        ctx = context.get_current()

        with ThreadPoolExecutor(max_workers=3) as pool:
            futures = []
            for i in range(5):
                f = pool.submit(context_aware_task, ctx, i)
                futures.append(f)

            for f in futures:
                f.result()

8. 跨进程 Context 传播 ​

8.1 HTTP 客户端 → 服务端传播 ​

这是最常见的跨进程传播场景：

python
# ========================================
# Service A（发送端 / HTTP 客户端）
# ========================================
import requests
from opentelemetry import trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("service-a")

def call_service_b():
    with tracer.start_as_current_span("call-service-b", kind=trace.SpanKind.CLIENT):
        headers = {}
        # inject() 将当前 Context 中的 trace 信息写入 headers
        inject(headers)

        print(f"Injected headers: {headers}")
        # {'traceparent': '00-abc...123-def...456-01', 'tracestate': ''}

        response = requests.get(
            "http://service-b:8080/api/data",
            headers=headers
        )
        return response.json()
python
# ========================================
# Service B（接收端 / HTTP 服务端）
# ========================================
from flask import Flask, request
from opentelemetry import trace
from opentelemetry.propagate import extract

app = Flask(__name__)
tracer = trace.get_tracer("service-b")

@app.route("/api/data")
def handle_request():
    # extract() 从 HTTP 请求头中恢复 Context
    ctx = extract(request.headers)

    # 使用恢复的 Context 创建 Span
    with tracer.start_as_current_span(
        "handle-request",
        context=ctx,
        kind=trace.SpanKind.SERVER
    ):
        # 这个 span 的 parent 就是 Service A 中的 "call-service-b" span
        data = fetch_from_database()
        return {"data": data}

8.2 gRPC 场景下的传播 ​

python
# gRPC 客户端拦截器
import grpc
from opentelemetry import trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("grpc-client")

class TracingClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        with tracer.start_as_current_span(
            client_call_details.method,
            kind=trace.SpanKind.CLIENT
        ):
            metadata = dict(client_call_details.metadata or [])
            inject(metadata)

            new_details = client_call_details._replace(
                metadata=list(metadata.items())
            )
            return continuation(new_details, request)
python
# gRPC 服务端拦截器
import grpc
from opentelemetry import trace
from opentelemetry.propagate import extract

tracer = trace.get_tracer("grpc-server")

class TracingServerInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        metadata = dict(handler_call_details.invocation_metadata)
        ctx = extract(metadata)

        with tracer.start_as_current_span(
            handler_call_details.method,
            context=ctx,
            kind=trace.SpanKind.SERVER
        ):
            return continuation(handler_call_details)

8.3 消息队列场景下的传播 ​

python
# ========================================
# Producer（生产者）
# ========================================
import json
from opentelemetry import trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("mq-producer")

def send_message(queue_client, message_body):
    with tracer.start_as_current_span(
        "send-message",
        kind=trace.SpanKind.PRODUCER
    ) as span:
        # 将 Context 注入到消息头中
        carrier = {}
        inject(carrier)

        message = {
            "headers": carrier,       # 追踪信息放在消息头中
            "body": message_body
        }

        queue_client.publish(
            queue="order-queue",
            body=json.dumps(message)
        )

        span.set_attribute("messaging.system", "rabbitmq")
        span.set_attribute("messaging.destination", "order-queue")
python
# ========================================
# Consumer（消费者）
# ========================================
from opentelemetry import trace
from opentelemetry.propagate import extract

tracer = trace.get_tracer("mq-consumer")

def handle_message(raw_message):
    message = json.loads(raw_message)

    # 从消息头中提取 Context
    ctx = extract(message.get("headers", {}))

    with tracer.start_as_current_span(
        "process-message",
        context=ctx,
        kind=trace.SpanKind.CONSUMER
    ) as span:
        span.set_attribute("messaging.system", "rabbitmq")
        span.set_attribute("messaging.destination", "order-queue")

        process_order(message["body"])

8.4 Getter 和 Setter 自定义 ​

当载体不是标准 dict 时，需要自定义 Getter/Setter：

python
from opentelemetry.propagators import textmap

# 自定义 Getter（用于非标准载体的 extract）
class MyGetter(textmap.Getter):
    def get(self, carrier, key):
        """从自定义载体中获取值"""
        # 例如：载体是一个自定义的 Message 对象
        value = carrier.get_header(key)
        if value is None:
            return None
        return [value]  # 必须返回列表

    def keys(self, carrier):
        """返回载体中所有可用的 key"""
        return carrier.get_all_header_names()

# 自定义 Setter（用于非标准载体的 inject）
class MySetter(textmap.Setter):
    def set(self, carrier, key, value):
        """向自定义载体中写入值"""
        carrier.set_header(key, value)

# 使用自定义 Getter/Setter
from opentelemetry.propagate import inject, extract

custom_getter = MyGetter()
custom_setter = MySetter()

# inject 时使用自定义 Setter
inject(my_custom_carrier, setter=custom_setter)

# extract 时使用自定义 Getter
ctx = extract(my_custom_carrier, getter=custom_getter)

9. 异步场景下的 Context 传播 ​

9.1 asyncio 中的 Context 传播 ​

Python 的 contextvars.ContextVar 天然支持 asyncio，OpenTelemetry 利用了这一特性：

python
import asyncio
from opentelemetry import trace

tracer = trace.get_tracer("async-demo")

async def handle_request():
    with tracer.start_as_current_span("async-handler"):
        # await 不会丢失 Context
        result = await fetch_data()
        await process_data(result)

async def fetch_data():
    # ✅ 自动继承调用者的 Context
    with tracer.start_as_current_span("fetch-data"):
        await asyncio.sleep(0.1)
        return {"key": "value"}

async def process_data(data):
    # ✅ 自动继承调用者的 Context
    with tracer.start_as_current_span("process-data"):
        await asyncio.sleep(0.05)

9.2 asyncio.gather 中的 Context 传播 ​

python
import asyncio
from opentelemetry import trace

tracer = trace.get_tracer("gather-demo")

async def task_a():
    with tracer.start_as_current_span("task-a"):
        await asyncio.sleep(0.1)

async def task_b():
    with tracer.start_as_current_span("task-b"):
        await asyncio.sleep(0.2)

async def main():
    with tracer.start_as_current_span("main"):
        # ✅ asyncio.gather 中的每个 task 都会继承当前 Context
        await asyncio.gather(task_a(), task_b())
        # task-a 和 task-b 都是 main 的子 span

# Trace 结构：
# main
# ├── task-a
# └── task-b

9.3 asyncio.create_task 的 Context 传播 ​

python
import asyncio
from opentelemetry import trace

tracer = trace.get_tracer("create-task-demo")

async def background_work():
    # ✅ Python 3.7+ 中，create_task 会自动复制当前 ContextVar
    with tracer.start_as_current_span("background-work"):
        await asyncio.sleep(1)

async def main():
    with tracer.start_as_current_span("main"):
        # create_task 在 Python 3.7+ 会复制当前 Context
        task = asyncio.create_task(background_work())
        await task

9.4 需要注意的陷阱 ​

python
import asyncio
from opentelemetry import trace, context

tracer = trace.get_tracer("pitfall-demo")

# ❌ 陷阱 1：在 span 结束后还有异步操作使用旧 span
async def bad_fire_and_forget():
    with tracer.start_as_current_span("parent"):
        # background_work 创建时捕获了 parent 的 Context
        asyncio.create_task(slow_background_work())
    # ⚠️ parent span 可能在 background_work 完成前就结束了
    # 导致 background_work 的 span 关联到一个已结束的 parent

# ✅ 正确做法：等待所有子任务完成
async def good_fire_and_forget():
    with tracer.start_as_current_span("parent"):
        task = asyncio.create_task(slow_background_work())
        await task  # 确保子任务在 parent span 结束前完成

# ❌ 陷阱 2：回调函数中丢失 Context
async def callback_pitfall():
    with tracer.start_as_current_span("setup"):
        ctx = context.get_current()

        def callback():
            # ⚠️ 回调执行时，Context 可能已经变了
            # 需要手动 attach
            token = context.attach(ctx)
            try:
                with tracer.start_as_current_span("callback"):
                    pass
            finally:
                context.detach(token)

        # 注册回调
        loop = asyncio.get_event_loop()
        loop.call_later(1.0, callback)

9.5 aiohttp 中的 Context 传播 ​

python
import aiohttp
from opentelemetry import trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("aiohttp-client")

async def call_downstream():
    with tracer.start_as_current_span("http-call", kind=trace.SpanKind.CLIENT):
        headers = {}
        inject(headers)

        async with aiohttp.ClientSession() as session:
            async with session.get(
                "http://downstream-service/api",
                headers=headers
            ) as response:
                return await response.json()

10. 自定义 Propagator ​

10.1 为什么需要自定义 Propagator？ ​

  • 需要兼容公司内部的追踪格式
  • 需要在特殊载体（非 HTTP）中传播 Context
  • 需要传播额外的自定义数据

10.2 实现自定义 Propagator ​

python
from opentelemetry.context.propagation import TextMapPropagator
from opentelemetry.propagators import textmap
from opentelemetry import trace, context
from opentelemetry.trace import TraceFlags, SpanContext, NonRecordingSpan
from typing import Optional, List, Set

class CustomPropagator(TextMapPropagator):
    """自定义传播器示例：使用 X-Custom-Trace 头"""

    TRACE_HEADER = "x-custom-trace"
    # 格式: {trace_id}:{span_id}:{sampled}

    def inject(
        self,
        carrier,
        context: Optional[context.Context] = None,
        setter: textmap.Setter = textmap.default_setter,
    ):
        span = trace.get_current_span(context)
        span_context = span.get_span_context()

        if not span_context.is_valid:
            return

        sampled = "1" if span_context.trace_flags & TraceFlags.SAMPLED else "0"
        value = f"{span_context.trace_id:032x}:{span_context.span_id:016x}:{sampled}"

        setter.set(carrier, self.TRACE_HEADER, value)

    def extract(
        self,
        carrier,
        context: Optional[context.Context] = None,
        getter: textmap.Getter = textmap.default_getter,
    ) -> context.Context:
        if context is None:
            context_to_use = context.get_current()
        else:
            context_to_use = context

        header_value = getter.get(carrier, self.TRACE_HEADER)
        if not header_value:
            return context_to_use

        value = header_value[0] if isinstance(header_value, list) else header_value

        try:
            trace_id_str, span_id_str, sampled = value.split(":")
            trace_id = int(trace_id_str, 16)
            span_id = int(span_id_str, 16)
            trace_flags = TraceFlags.SAMPLED if sampled == "1" else TraceFlags.DEFAULT
        except (ValueError, IndexError):
            return context_to_use

        span_context = SpanContext(
            trace_id=trace_id,
            span_id=span_id,
            is_remote=True,
            trace_flags=trace_flags,
        )

        span = NonRecordingSpan(span_context)
        return trace.set_span_in_context(span, context_to_use)

    @property
    def fields(self) -> Set[str]:
        return {self.TRACE_HEADER}

10.3 注册自定义 Propagator ​

python
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositeTextMapPropagator
from opentelemetry.trace.propagation import TraceContextTextMapPropagator

# 与标准 propagator 组合使用
set_global_textmap(
    CompositeTextMapPropagator([
        TraceContextTextMapPropagator(),     # W3C 标准
        CustomPropagator(),                  # 自定义格式
    ])
)

# 使用方式和标准 propagator 完全一致
from opentelemetry.propagate import inject, extract

headers = {}
inject(headers)
# headers 同时包含 traceparent 和 x-custom-trace

11. 实践 Demo ​

Demo 1：进程内 Context 传播完整示例 ​

python
"""
demo_in_process_context.py
演示进程内 Context 传播机制：attach/detach、隐式传播、显式传播
"""
from opentelemetry import trace, context
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

def setup():
    resource = Resource.create({"service.name": "context-demo"})
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
    trace.set_tracer_provider(provider)

def demo_implicit_propagation():
    """演示隐式传播：with 语句自动管理 Context"""
    tracer = trace.get_tracer("implicit-demo")

    print("\n=== 隐式传播 Demo ===")
    with tracer.start_as_current_span("parent") as parent:
        print(f"Parent span_id: {parent.get_span_context().span_id:016x}")

        with tracer.start_as_current_span("child") as child:
            print(f"Child span_id:  {child.get_span_context().span_id:016x}")
            print(f"Child parent:   {child.parent.span_id:016x}")
            assert child.parent.span_id == parent.get_span_context().span_id

def demo_explicit_context():
    """演示显式 Context 操作：create_key、set_value、get_value"""
    print("\n=== 显式 Context 操作 Demo ===")

    request_id_key = context.create_key("request-id")

    # 设置值
    ctx = context.set_value(request_id_key, "req-abc-123")
    token = context.attach(ctx)

    try:
        # 读取值
        value = context.get_value(request_id_key)
        print(f"Current request_id: {value}")  # req-abc-123

        # 嵌套修改
        ctx2 = context.set_value(request_id_key, "req-def-456")
        token2 = context.attach(ctx2)
        try:
            value2 = context.get_value(request_id_key)
            print(f"Nested request_id: {value2}")  # req-def-456
        finally:
            context.detach(token2)

        # 恢复外层值
        value_restored = context.get_value(request_id_key)
        print(f"Restored request_id: {value_restored}")  # req-abc-123
    finally:
        context.detach(token)

    # 全部 detach 后
    value_none = context.get_value(request_id_key)
    print(f"After detach: {value_none}")  # None

def demo_cross_thread():
    """演示跨线程 Context 传播"""
    import threading

    tracer = trace.get_tracer("thread-demo")
    results = []

    def worker(ctx, task_name):
        token = context.attach(ctx)
        try:
            with tracer.start_as_current_span(task_name) as span:
                results.append(f"{task_name}: span_id={span.get_span_context().span_id:016x}")
        finally:
            context.detach(token)

    print("\n=== 跨线程传播 Demo ===")
    with tracer.start_as_current_span("main-thread") as parent:
        ctx = context.get_current()
        threads = []
        for i in range(3):
            t = threading.Thread(target=worker, args=(ctx, f"worker-{i}"))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

    for r in results:
        print(f"  {r}")

if __name__ == "__main__":
    setup()
    demo_implicit_propagation()
    demo_explicit_context()
    demo_cross_thread()

Demo 2：两个微服务间的 Context 传播 ​

python
"""
demo_cross_service_propagation.py
使用 Flask 演示两个微服务之间的 Context 传播（可在同一进程模拟）
需要安装：pip install flask requests opentelemetry-sdk opentelemetry-exporter-otlp
"""
import json
import threading
import time
import requests
from flask import Flask, request as flask_request
from opentelemetry import trace, baggage, context
from opentelemetry.propagate import inject, extract, set_global_textmap
from opentelemetry.propagators.composite import CompositeTextMapPropagator
from opentelemetry.trace.propagation import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

# ============================================================
# 设置全局传播器
# ============================================================
set_global_textmap(
    CompositeTextMapPropagator([
        TraceContextTextMapPropagator(),
        W3CBaggagePropagator(),
    ])
)

# ============================================================
# Service B（下游服务）
# ============================================================
service_b_app = Flask("service-b")
service_b_provider = TracerProvider(
    resource=Resource.create({"service.name": "service-b"})
)
service_b_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
service_b_tracer = trace.get_tracer("service-b", tracer_provider=service_b_provider)

@service_b_app.route("/api/process")
def service_b_process():
    # 1. 从 HTTP 请求头提取 Context
    ctx = extract(flask_request.headers)

    # 2. 使用提取的 Context 创建 span
    with service_b_tracer.start_as_current_span(
        "service-b.process",
        context=ctx,
        kind=trace.SpanKind.SERVER,
    ) as span:
        # 3. 读取传播过来的 Baggage
        user_id = baggage.get_baggage("user.id", ctx)
        tenant_id = baggage.get_baggage("tenant.id", ctx)

        span.set_attribute("user.id", user_id or "unknown")
        span.set_attribute("tenant.id", tenant_id or "unknown")

        result = {
            "status": "processed",
            "user_id": user_id,
            "tenant_id": tenant_id,
        }
        return json.dumps(result)

# ============================================================
# Service A（上游服务 / 客户端）
# ============================================================
service_a_provider = TracerProvider(
    resource=Resource.create({"service.name": "service-a"})
)
service_a_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
service_a_tracer = trace.get_tracer("service-a", tracer_provider=service_a_provider)

def service_a_call():
    with service_a_tracer.start_as_current_span(
        "service-a.handle-user-request",
        kind=trace.SpanKind.SERVER,
    ):
        # 设置 Baggage（业务数据）
        ctx = baggage.set_baggage("user.id", "u-42")
        ctx = baggage.set_baggage("tenant.id", "t-acme", context=ctx)
        token = context.attach(ctx)

        try:
            with service_a_tracer.start_as_current_span(
                "service-a.call-service-b",
                kind=trace.SpanKind.CLIENT,
            ):
                # 注入 Context 到 HTTP 请求头
                headers = {}
                inject(headers)

                print("\n--- Injected headers ---")
                for k, v in headers.items():
                    print(f"  {k}: {v}")

                # 调用 Service B
                response = requests.get(
                    "http://127.0.0.1:5001/api/process",
                    headers=headers,
                )
                print(f"\n--- Service B response ---")
                print(f"  {response.json()}")
        finally:
            context.detach(token)

if __name__ == "__main__":
    # 启动 Service B
    server_thread = threading.Thread(
        target=lambda: service_b_app.run(port=5001, debug=False, use_reloader=False),
        daemon=True,
    )
    server_thread.start()
    time.sleep(1)

    # Service A 发起调用
    service_a_call()

    # 等待 span 导出
    time.sleep(1)
    print("\n✅ Demo 完成！观察控制台输出的 span 信息，注意 trace_id 在两个服务间是一致的。")

Demo 3：异步 Context 传播 ​

python
"""
demo_async_context.py
演示 asyncio 场景下的 Context 传播
"""
import asyncio
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

def setup():
    resource = Resource.create({"service.name": "async-context-demo"})
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
    trace.set_tracer_provider(provider)

tracer = trace.get_tracer("async-demo")

async def fetch_user(user_id: str):
    with tracer.start_as_current_span("fetch-user") as span:
        span.set_attribute("user.id", user_id)
        await asyncio.sleep(0.05)
        return {"id": user_id, "name": "Alice"}

async def fetch_orders(user_id: str):
    with tracer.start_as_current_span("fetch-orders") as span:
        span.set_attribute("user.id", user_id)
        await asyncio.sleep(0.08)
        return [{"order_id": "ord-1"}, {"order_id": "ord-2"}]

async def enrich_order(order):
    with tracer.start_as_current_span("enrich-order") as span:
        span.set_attribute("order.id", order["order_id"])
        await asyncio.sleep(0.03)
        order["enriched"] = True
        return order

async def handle_request(user_id: str):
    with tracer.start_as_current_span("handle-request") as span:
        span.set_attribute("user.id", user_id)

        # 并发获取用户信息和订单
        user, orders = await asyncio.gather(
            fetch_user(user_id),
            fetch_orders(user_id),
        )

        # 并发 enrich 每个订单
        enriched = await asyncio.gather(
            *[enrich_order(o) for o in orders]
        )

        return {"user": user, "orders": enriched}

async def main():
    setup()
    result = await handle_request("u-42")
    print(f"\n--- Result ---")
    print(f"User: {result['user']['name']}")
    print(f"Orders: {len(result['orders'])}")

    # 等待 span 导出
    await asyncio.sleep(0.5)

    # Trace 结构：
    # handle-request
    # ├── fetch-user         (并发)
    # ├── fetch-orders       (并发)
    # ├── enrich-order[0]    (并发)
    # └── enrich-order[1]    (并发)
    print("\n✅ 所有 span 应该共享同一个 trace_id，且都是 handle-request 的子 span")

if __name__ == "__main__":
    asyncio.run(main())

Demo 4：Baggage 全链路传递 ​

python
"""
demo_baggage_propagation.py
演示 Baggage 在服务链路中的全链路传递
"""
from opentelemetry import trace, baggage, context
from opentelemetry.propagate import inject, extract, set_global_textmap
from opentelemetry.propagators.composite import CompositeTextMapPropagator
from opentelemetry.trace.propagation import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    SimpleSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import Resource

# 配置
set_global_textmap(
    CompositeTextMapPropagator([
        TraceContextTextMapPropagator(),
        W3CBaggagePropagator(),
    ])
)

provider = TracerProvider(resource=Resource.create({"service.name": "baggage-demo"}))
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("baggage-demo")

def simulate_service(name, incoming_headers):
    """模拟一个服务：接收请求 → 处理 → 调用下游"""

    # 1. 提取上游传来的 Context
    ctx = extract(incoming_headers)

    with tracer.start_as_current_span(
        f"{name}.handle",
        context=ctx,
        kind=trace.SpanKind.SERVER,
    ) as span:
        # 2. 读取 Baggage
        user_id = baggage.get_baggage("user.id", ctx)
        tenant_id = baggage.get_baggage("tenant.id", ctx)
        priority = baggage.get_baggage("request.priority", ctx)

        print(f"\n[{name}] Received Baggage:")
        print(f"  user.id = {user_id}")
        print(f"  tenant.id = {tenant_id}")
        print(f"  request.priority = {priority}")

        # 3. 将 Baggage 记录到 Span 属性中
        if user_id:
            span.set_attribute("user.id", user_id)
        if tenant_id:
            span.set_attribute("tenant.id", tenant_id)

        # 4. 也可以添加新的 Baggage
        new_ctx = baggage.set_baggage(
            f"{name}.processed", "true", context=ctx
        )
        token = context.attach(new_ctx)

        try:
            # 5. 准备调用下游
            outgoing_headers = {}
            inject(outgoing_headers)
            return outgoing_headers
        finally:
            context.detach(token)

def demo_full_chain():
    print("=== Baggage 全链路传递 Demo ===")

    # Gateway 设置初始 Baggage
    with tracer.start_as_current_span("gateway", kind=trace.SpanKind.SERVER):
        ctx = baggage.set_baggage("user.id", "u-42")
        ctx = baggage.set_baggage("tenant.id", "t-acme", context=ctx)
        ctx = baggage.set_baggage("request.priority", "high", context=ctx)
        token = context.attach(ctx)

        try:
            headers = {}
            inject(headers)
        finally:
            context.detach(token)

    print(f"\n[Gateway] Injected headers:")
    for k, v in headers.items():
        print(f"  {k}: {v}")

    # 模拟服务链路: Gateway → Service A → Service B → Service C
    headers_a = simulate_service("service-a", headers)
    headers_b = simulate_service("service-b", headers_a)
    headers_c = simulate_service("service-c", headers_b)

    print(f"\n[Final] Headers reaching end of chain:")
    for k, v in headers_c.items():
        print(f"  {k}: {v}")

if __name__ == "__main__":
    demo_full_chain()
    print("\n✅ Demo 完成！Baggage 成功在 3 个服务间传递。")

12. 常见问题 QA ​

Q1：Context 和 SpanContext 有什么区别？ ​

概念ContextSpanContext
包含内容Span、Baggage、自定义数据trace_id、span_id、trace_flags、tracestate
可变性不可变，修改返回新对象不可变
作用范围进程内管理执行上下文跨进程传播追踪标识
关系Context 包含 SpanContextSpanContext 是 Context 的一部分
python
from opentelemetry import trace, context

# Context 是容器
ctx = context.get_current()

# SpanContext 是容器中 Span 的身份标识
span = trace.get_current_span(ctx)
span_context = span.get_span_context()
print(span_context.trace_id, span_context.span_id)

Q2：inject/extract 的载体（carrier）可以是什么？ ​

默认的 DefaultGetter 和 DefaultSetter 支持任何类似 dict 的对象。对于自定义载体需要实现 Getter/Setter：

python
# ✅ dict（最常用）
headers = {}
inject(headers)

# ✅ Flask 的 request.headers（只读，用于 extract）
ctx = extract(flask_request.headers)

# ✅ requests 的 PreparedRequest headers
inject(prepared_request.headers)

# ❌ 不可变对象（如 tuple）不能直接作为 inject 的载体
# 需要先转成 dict

Q3：为什么我的跨服务 Trace 断裂了？ ​

排查清单：

python
# 1. 检查是否配置了 Propagator
from opentelemetry.propagate import get_global_textmap
propagator = get_global_textmap()
print(f"当前 Propagator: {propagator}")
# 如果是 NoOpTextMapPropagator，说明没有配置

# 2. 检查 inject 是否正确
headers = {}
inject(headers)
print(f"Injected headers: {headers}")
# 应该包含 traceparent

# 3. 检查 HTTP 请求是否携带了 headers
# 确保调用 requests.get/post 时传入了 headers 参数

# 4. 检查服务端是否正确 extract
# 确保使用了正确的载体（如 flask_request.headers）

# 5. 检查服务端创建 span 时是否使用了 extract 出的 context
# with tracer.start_as_current_span("...", context=extracted_ctx):

常见原因：

原因解决方案
没有配置 Propagator调用 set_global_textmap()
inject 时没有活跃的 Span确保 inject 在 Span 的 with 块内调用
没有传递 headers检查 HTTP 客户端调用
extract 后没有使用 Context将 ctx 传递给 start_as_current_span(context=ctx)
反向代理/网关过滤了 header配置代理透传 traceparent 等 header

Q4：Baggage 和 Span Attributes 应该如何选择？ ​

决策流程：

这个数据需要传递给下游服务吗？
├── 是 → 这个数据是敏感的吗？
│        ├── 是 → ❌ 不要用 Baggage，考虑其他安全传递方式
│        └── 否 → ✅ 使用 Baggage
└── 否 → ✅ 使用 Span Attributes

Q5：attach/detach 必须配对使用吗？会导致什么问题？ ​

必须配对！ 不配对会导致 Context 泄漏：

python
# ❌ Context 泄漏
def leaky():
    ctx = baggage.set_baggage("key", "value")
    context.attach(ctx)  # 没有 detach！

leaky()
# 此后所有操作都会看到 key=value 的 baggage
# 其他请求也会受影响（如果是在 Web 服务器中）

# ✅ 正确做法
def safe():
    ctx = baggage.set_baggage("key", "value")
    token = context.attach(ctx)
    try:
        # 业务逻辑
        pass
    finally:
        context.detach(token)

Q6：如何在日志中自动包含 trace_id 和 span_id？ ​

python
import logging
from opentelemetry import trace

class TraceLogFormatter(logging.Formatter):
    def format(self, record):
        span = trace.get_current_span()
        ctx = span.get_span_context()
        if ctx.is_valid:
            record.trace_id = f"{ctx.trace_id:032x}"
            record.span_id = f"{ctx.span_id:016x}"
        else:
            record.trace_id = "0" * 32
            record.span_id = "0" * 16
        return super().format(record)

formatter = TraceLogFormatter(
    "%(asctime)s [trace_id=%(trace_id)s span_id=%(span_id)s] %(message)s"
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger = logging.getLogger("my-app")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# 使用
tracer = trace.get_tracer("demo")
with tracer.start_as_current_span("request"):
    logger.info("Processing request")
    # 输出: 2025-01-01 12:00:00 [trace_id=abc...123 span_id=def...456] Processing request

Q7：traceparent 中的 trace-flags 有什么作用？ ​

trace-flags 目前只定义了一个 bit（bit 0 = sampled）：

trace-flags = 01  → sampled (该 trace 被采样，应该记录)
trace-flags = 00  → not sampled (该 trace 未被采样)

下游服务应该尊重上游的采样决策：
- 如果 trace-flags = 01，下游也应该采样
- 如果 trace-flags = 00，下游可以选择不记录

这通过 ParentBasedSampler 实现：
from opentelemetry.sdk.trace.sampling import ParentBasedSampler, ALWAYS_ON
sampler = ParentBasedSampler(root=ALWAYS_ON)
# 如果 parent 是 sampled，则子 span 也 sampled
# 如果 parent 是 not sampled，则子 span 也 not sampled

Q8：如何在不使用自动检测的情况下确保 Context 不丢失？ ​

关键原则：Context 传播的完整性依赖于每一跳都正确执行 inject/extract。

python
# 中间件模式：统一处理 extract（服务端）
class TracingMiddleware:
    def __init__(self, app, tracer):
        self.app = app
        self.tracer = tracer

    def __call__(self, environ, start_response):
        from opentelemetry.propagate import extract
        headers = {
            key[5:].replace("_", "-").lower(): value
            for key, value in environ.items()
            if key.startswith("HTTP_")
        }
        ctx = extract(headers)

        with self.tracer.start_as_current_span(
            f"{environ['REQUEST_METHOD']} {environ['PATH_INFO']}",
            context=ctx,
            kind=trace.SpanKind.SERVER,
        ):
            return self.app(environ, start_response)

# HTTP 客户端封装：统一处理 inject（客户端）
import requests
from opentelemetry.propagate import inject

def traced_request(method, url, **kwargs):
    tracer = trace.get_tracer("http-client")
    with tracer.start_as_current_span(
        f"HTTP {method} {url}",
        kind=trace.SpanKind.CLIENT,
    ) as span:
        headers = kwargs.pop("headers", {})
        inject(headers)
        kwargs["headers"] = headers

        response = requests.request(method, url, **kwargs)
        span.set_attribute("http.status_code", response.status_code)
        return response

Q9：多种 Propagator 格式之间如何选择？ ​

格式推荐场景优势
W3C TraceContext新系统、标准化环境W3C 标准，最广泛支持
B3 Multi已有 Zipkin 基础设施Zipkin 生态兼容
B3 Single需要减少 Header 数量单个 Header
Jaeger已有 Jaeger 基础设施Jaeger 原生支持

最佳实践：

  • 新项目使用 W3C TraceContext + Baggage
  • 迁移期使用 CompositeTextMapPropagator 同时支持多种格式
  • 环境变量 OTEL_PROPAGATORS 可以无代码切换

Q10：Context 传播在 asyncio + 多线程混合场景下如何处理？ ​

python
import asyncio
import concurrent.futures
from opentelemetry import trace, context

tracer = trace.get_tracer("mixed-demo")

def cpu_intensive_work(ctx, data):
    """在线程池中执行的 CPU 密集型任务"""
    token = context.attach(ctx)
    try:
        with tracer.start_as_current_span("cpu-work"):
            return sum(range(data))
    finally:
        context.detach(token)

async def handle_request():
    with tracer.start_as_current_span("async-handler"):
        ctx = context.get_current()

        loop = asyncio.get_event_loop()
        # 将 Context 传递给线程池中的函数
        result = await loop.run_in_executor(
            concurrent.futures.ThreadPoolExecutor(),
            cpu_intensive_work,
            ctx,
            1000000,
        )
        return result

Q11：Baggage 的性能影响有多大？ ​

因素影响
序列化/反序列化极低（简单的字符串操作）
网络传输每个请求增加 baggage Header（通常 < 1KB）
内存每个 Context 增加一个小 dict
数据增长链路越长，baggage Header 越大（每个服务可能添加新条目）

建议：

  • 控制 Baggage 条目在 10 个以内
  • 每个 value 保持简短（< 256 字符）
  • 不要在 Baggage 中放大量数据
  • 定期清理不再需要的 Baggage 条目

Q12：为什么 extract 后要把 Context 传给 start_as_current_span？ ​

python
# ❌ 错误：extract 后没使用 Context
ctx = extract(request.headers)
with tracer.start_as_current_span("handler"):  # 使用的是默认空 Context
    pass  # 这个 span 不会和上游关联！

# ✅ 正确做法 1：传递 Context 参数
ctx = extract(request.headers)
with tracer.start_as_current_span("handler", context=ctx):
    pass  # 自动成为上游 span 的子 span

# ✅ 正确做法 2：先 attach 再创建 span
ctx = extract(request.headers)
token = context.attach(ctx)
try:
    with tracer.start_as_current_span("handler"):
        pass  # 也会自动关联
finally:
    context.detach(token)

13. 学习检查清单 ​

  • [ ] 能解释 Context、Propagator、Baggage 的概念和关系
  • [ ] 理解 attach()、detach()、token 的工作原理和必须配对的原因
  • [ ] 能区分 Context 和 SpanContext 的区别
  • [ ] 能配置全局 Propagator（set_global_textmap）
  • [ ] 理解 CompositeTextMapPropagator 的工作原理（链式 extract）
  • [ ] 能解析 W3C traceparent Header 的各个字段
  • [ ] 能使用 inject() 和 extract() 实现跨服务 Context 传播
  • [ ] 能使用 Baggage API 设置、读取、删除 Baggage
  • [ ] 知道 Baggage 的安全注意事项（不放敏感数据）
  • [ ] 能处理跨线程场景的 Context 传播
  • [ ] 理解 asyncio 场景下 Context 的自动传播机制
  • [ ] 能实现自定义 Propagator
  • [ ] 能排查跨服务 Trace 断裂的问题
  • [ ] 知道多种 Propagator 格式的适用场景

下一阶段：6. 导出与后端集成

前一阶段：4. Logs 深入