Python 基础体系 · 第 107/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 可观测性：日志、指标、Trace、Context 和故障定位

可观测性（Observability）不是“多打一些日志”，而是让系统外部输出的数据，足以回答下面三类问题：

系统现在是否健康？
哪一类请求、任务或依赖出了问题？
问题是从哪里开始的，经过了哪些组件，最终为什么失败？

在 Python 服务中，日志、指标和 Trace 分别提供不同的观察视角：

**日志（Log）**记录离散事件及其详细上下文；
**指标（Metric）**记录可聚合的数值变化，用于判断趋势、容量和告警；
Trace记录一次请求或任务跨组件传播的因果路径；
Context把请求身份、Trace 身份和租户等上下文，正确地带到异步任务、线程、进程和远程服务中。

四者结合后，典型故障定位路径不是“先搜日志”，而是：

指标发现异常
    ↓
Trace 找到异常请求及调用链
    ↓
Context 将同一个 trace_id / span_id 关联到日志
    ↓
日志提供参数、异常栈和业务状态
    ↓
代码、依赖、容量或发布变更被定位

OpenTelemetry 为 Trace、指标和日志提供统一的数据模型、API 和 SDK。当前 OpenTelemetry Python 文档中，Trace 和 Metrics 标记为 Stable，Logs 仍标记为 Development，因此日志集成尤其需要锁定依赖版本并验证导出结果。(opentelemetry.io)

一、先建立统一模型：事件、时间序列与因果链

1. 日志是离散事件

一条日志通常描述一个发生过的事件：

{
  "timestamp": "2026-09-01T10:15:23.421Z",
  "severity": "ERROR",
  "logger": "payment.service",
  "message": "payment failed",
  "order_id": "o-1001",
  "error_type": "TimeoutError",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7"
}

日志的关键特征是：

有明确时间；
有事件级别；
有文字消息；
可以附加任意字段；
通常包含异常栈；
能够保留某次请求的细节。

日志适合回答：

这个请求使用了什么参数？
哪个异常被抛出？
哪个订单、用户或任务受到影响？

日志不适合直接回答：

最近五分钟错误率是否上升？
当前队列积压是否持续扩大？
所有请求的 P99 延迟是多少？

这些问题需要指标。

2. 指标是可聚合的数值序列

指标可以抽象为：

M = (name, labels, timestamp, value)

其中：

name 是指标名称；
labels 是有限集合的维度；
timestamp 是采样时间；
value 是数值。

例如：

http.server.request.duration
method="GET"
route="/orders/{id}"
status_code="500"
value=0.235

指标的价值在于聚合。假设一分钟内有 $n$ 个请求，错误请求数为 $e$ ，则错误率为：

error\_rate = \frac{e}{n}

如果总请求数为 10,000，错误数为 250：

error\_rate = \frac{250}{10000} = 2.5\%

指标告诉你“整体发生了什么”，但通常无法告诉你“某一个请求为什么发生”。

3. Trace 是一次操作的因果图

Trace 不是一条日志，而是一组具有父子关系的 Span。

一个 Span 表示某段有开始时间和结束时间的操作，例如：

接收 HTTP 请求；
查询数据库；
调用支付服务；
执行 Redis 命令；
消费一条消息；
执行后台任务。

可以把一次请求表示为：

Trace: 下单请求
└── Span: HTTP GET /orders/1001
    ├── Span: 查询订单数据库
    ├── Span: 调用库存服务
    └── Span: 调用支付服务
        └── Span: 支付服务查询风控

每个 Span 至少包含：

trace_id
span_id
parent_span_id
start_time
end_time
name
attributes
events
status

Span 的耗时为：

duration = end\_time - start\_time

父 Span 的耗时通常覆盖子 Span，但不一定等于所有子 Span 耗时之和，因为还包括：

Python 代码自身执行；
排队；
网络发送和接收；
重试；
并发子任务；
未埋点的内部操作。

因此，不能简单地把所有子 Span 的耗时相加来推导请求总耗时。

4. Context 是“当前执行路径”的隐式状态

Context 用于保存当前执行路径相关的数据，例如：

trace_id
span_id
request_id
tenant_id
user_id
deadline
locale

在 Python 中，contextvars 提供了上下文局部变量。它与普通全局变量不同，也不能简单等同于线程本地变量。

ContextVar 的值属于当前 Context；异步任务可以拥有自己的上下文，从而避免不同请求之间互相覆盖。Python 文档建议在模块顶层创建 ContextVar，而不是在闭包中创建；copy_context() 的复制复杂度为 O(1)。(docs.python.org)

二、Python 日志的完整数据流

Python 标准库 logging 的核心对象是：

Logger
  ↓
LogRecord
  ↓
Filter
  ↓
Handler
  ↓
Formatter
  ↓
输出目标

1. Logger：产生日志事件

应用代码一般只依赖 Logger：

import logging

logger = logging.getLogger(__name__)

logger.info("order created")
logger.warning("inventory is low")
logger.error("payment failed")

Logger 名称通常使用模块名：

logger = logging.getLogger(__name__)

如果模块是 payment/service.py，Logger 名称可能是：

payment.service

Logger 形成层级结构：

root
├── payment
│   └── payment.service
└── inventory
    └── inventory.client

日志事件会根据 Logger 和 Handler 的级别、过滤器以及传播关系决定是否输出。

2. LogRecord：日志的结构化中间对象

Logger 接收到：

logger.info("order=%s status=%s", order_id, status)

并不会立即只生成一段字符串，而是先生成 LogRecord。其中包括：

Logger 名称；
日志级别；
源文件；
行号；
函数名；
时间；
消息模板；
参数；
异常信息。

消息通常在格式化阶段通过 % 风格组合：

msg % args

因此，推荐这样写：

logger.info("order=%s status=%s", order_id, status)

而不是：

logger.info(f"order={order_id} status={status}")

前者允许日志级别关闭时延迟消息格式化，避免无用字符串构造：

if logger.isEnabledFor(logging.DEBUG):
    logger.debug("large payload=%r", expensive_object)

需要注意，参数表达式本身仍然会先求值：

logger.debug("payload=%r", build_payload())

即使 DEBUG 被关闭，build_payload() 仍会执行。因此昂贵计算必须显式放进级别判断中。

3. Handler：决定日志写到哪里

Handler 负责将日志输出到目标位置，例如：

标准输出；
标准错误；
文件；
轮转文件；
网络；
自定义采集管道。

一个进程可以有多个 Handler：

Logger
├── ConsoleHandler → stderr
└── FileHandler    → application.log

这会产生一个常见问题：重复输出。

例如：

logger = logging.getLogger("app.service")
logger.addHandler(service_handler)

root = logging.getLogger()
root.addHandler(root_handler)

如果 logger.propagate 为 True，同一条记录可能先由 service_handler 输出，再向上传播给 root 的 root_handler，最终出现两份日志。

当模块级 Handler 已经负责输出时，可以关闭传播：

logger.propagate = False

但不要在每次请求中重复添加 Handler，否则请求数越多，输出副本越多。

4. Filter：过滤或增加字段

Filter 不只是“过滤”。它还可以修改 LogRecord，例如把 ContextVar 中的请求信息注入日志。

Python 3.12 起，Filter 可以返回一个新的 LogRecord，替换当前处理流程中的记录，这使得 Handler 级别的字段修改不必影响其他 Handler。(docs.python.org)

下面是一个完整示例：

from __future__ import annotations

import contextvars
import logging
import sys
import uuid


request_id_var: contextvars.ContextVar[str | None] = (
    contextvars.ContextVar("request_id", default=None)
)


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get() or "-"
        return True


logger = logging.getLogger("demo")
logger.setLevel(logging.INFO)
logger.propagate = False

handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.INFO)
handler.addFilter(ContextFilter())
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s "
    "request_id=%(request_id)s "
    "logger=%(name)s "
    "message=%(message)s"
))
logger.addHandler(handler)


def handle_request() -> None:
    token = request_id_var.set(str(uuid.uuid4()))
    try:
        logger.info("request started")
        logger.info("loading user")
    finally:
        request_id_var.reset(token)


if __name__ == "__main__":
    handle_request()

可能输出：

2026-09-01 18:20:01,123 INFO request_id=4c4... logger=demo message=request started
2026-09-01 18:20:01,124 INFO request_id=4c4... logger=demo message=loading user

这里的状态变化是：

旧 Context:
request_id = None

set():
request_id = 4c4...

处理请求:
日志读取到 request_id=4c4...

reset(token):
request_id 恢复为 None

reset(token) 不能省略。否则同一个线程继续处理其他请求时，旧请求的 request_id 可能泄漏到新请求。

Python 3.14 为 Token 增加了上下文管理器支持，因此可以写成：

with request_id_var.set("req-1001"):
    logger.info("inside request")

离开 with 后变量自动恢复。该能力是 Python 3.14 新增内容，代码若要兼容更早版本，仍应使用 set() 加 reset()。(docs.python.org)

三、结构化日志：字段不是装饰，而是查询接口

纯文本日志：

payment failed order=o-1001 user=u-7 amount=99.0

人可以阅读，但机器难以稳定解析。结构化日志把消息和字段分开：

logger.error(
    "payment failed",
    extra={
        "order_id": "o-1001",
        "user_id": "u-7",
        "amount": 99.0,
    },
)

Formatter 可以输出这些额外字段：

formatter = logging.Formatter(
    "%(asctime)s %(levelname)s "
    "message=%(message)s "
    "order_id=%(order_id)s "
    "user_id=%(user_id)s"
)

但是这种写法有一个风险：不是每条日志都有 order_id，Formatter 直接访问不存在的字段会导致格式化错误。

更稳妥的方式是让 Filter 为字段提供默认值：

class DefaultFieldsFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        for field in ("order_id", "user_id", "trace_id", "span_id"):
            if not hasattr(record, field):
                setattr(record, field, "-")
        return True

生产系统中更常见的是直接输出 JSON：

import json
import logging


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "severity": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }

        for key in ("request_id", "trace_id", "span_id", "order_id"):
            if hasattr(record, key):
                payload[key] = getattr(record, key)

        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)

        return json.dumps(payload, ensure_ascii=False)

调用异常日志时使用：

try:
    charge()
except TimeoutError:
    logger.exception(
        "payment timeout",
        extra={"order_id": order_id},
    )

logger.exception() 应在 except 块中使用，它会自动携带当前异常信息。若在没有活动异常的地方调用，通常无法得到有意义的异常栈。

四、Context：为什么不能用全局变量或 `threading.local()`

1. 全局变量会发生请求串线

错误示例：

current_request_id = None


async def handle(request_id: str) -> None:
    global current_request_id
    current_request_id = request_id

    await do_io()
    print(current_request_id)

执行过程可能是：

任务 A 设置 request_id=A
任务 A await

任务 B 设置 request_id=B
任务 B await

任务 A 恢复
任务 A 读取到 request_id=B

原因不是 asyncio 出错，而是两个任务共享了同一个可变全局变量。

2. `threading.local()` 只隔离线程

threading.local() 可以隔离不同线程，但同一个事件循环线程中可能同时运行多个异步任务：

线程 1
├── Task A
└── Task B

因此，线程本地状态无法自然区分 Task A 和 Task B。

ContextVar 的设计目标正是保存并发代码中的上下文局部状态，Python 文档明确建议在异步框架中使用 ContextVar、copy_context() 和 Context 管理上下文。(docs.python.org)

3. ContextVar 的正确生命周期

from contextvars import ContextVar

trace_id_var: ContextVar[str | None] = ContextVar(
    "trace_id",
    default=None,
)


def process(trace_id: str) -> None:
    token = trace_id_var.set(trace_id)
    try:
        do_work()
    finally:
        trace_id_var.reset(token)

这里有三个重要保证：

set() 返回旧状态的 Token；
reset(token) 恢复到设置前的状态；
finally 保证异常路径也能清理。

嵌套上下文也可以正确恢复：

token_a = trace_id_var.set("trace-A")
try:
    token_b = trace_id_var.set("trace-B")
    try:
        assert trace_id_var.get() == "trace-B"
    finally:
        trace_id_var.reset(token_b)

    assert trace_id_var.get() == "trace-A"
finally:
    trace_id_var.reset(token_a)

如果不按 Token 的嵌套顺序恢复，代码可读性和行为都会变差。上下文变量不是用于保存业务状态的万能容器，它更适合保存“当前调用路径”的元数据。

4. 异步任务中的上下文

import asyncio
from contextvars import ContextVar

request_id_var = ContextVar("request_id", default="-")

async def child(name: str) -> None:
    await asyncio.sleep(0)
    print(name, request_id_var.get())

async def main() -> None:
    token = request_id_var.set("req-A")
    try:
        await asyncio.gather(
            child("child-1"),
            child("child-2"),
        )
    finally:
        request_id_var.reset(token)

asyncio.run(main())

预期结果：

child-1 req-A
child-2 req-A

这是因为创建异步任务时，任务会携带当时的执行上下文。实际系统仍然要验证所使用的任务创建方式、线程池和第三方框架是否正确传播上下文，不能把所有边界都假设为自动传播。

5. 线程池是显式边界

将函数提交到线程池时，不能把 ContextVar 传播当作普通参数传递：

from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context

executor = ThreadPoolExecutor(max_workers=4)


def blocking_work() -> str:
    return request_id_var.get()


def submit_with_context():
    ctx = copy_context()
    return executor.submit(ctx.run, blocking_work)

如果直接写：

executor.submit(blocking_work)

线程中的当前 Context 可能没有请求上下文，日志中的 request_id 就会变成默认值。

五、Trace：从“这条日志”变成“这次请求”

1. Trace、Span、Span Context

一个 Trace 表示跨服务的一次完整操作。一个 Trace 包含一个或多个 Span。

Trace ID = T1

Span A: API 请求
  Span ID = S1
  Parent = None

Span B: 数据库查询
  Span ID = S2
  Parent = S1

Span C: 调用库存服务
  Span ID = S3
  Parent = S1

Trace ID 标识整条调用链，Span ID 标识其中一个操作，Parent Span ID 表示直接调用关系。

一个常见误解是：

request_id 就是 trace_id。

二者作用相似但不等价：

request_id 通常由应用定义；
trace_id 遵循 Trace 系统的传播和采样规则；
一个业务请求可能产生多个内部任务；
一个消息处理过程可能没有传统 HTTP request；
重试可能产生多个 Span；
异步消息、批处理和并发 fan-out 可能需要 Span Links，而不是简单父子关系。

2. Span 的生命周期

手动创建 Span 时，生命周期应覆盖真正需要观测的代码：

from opentelemetry import trace

tracer = trace.get_tracer(__name__)


def load_order(order_id: str):
    with tracer.start_as_current_span("load_order") as span:
        span.set_attribute("app.order_id", order_id)

        order = query_database(order_id)

        if order is None:
            span.add_event("order_not_found")
            return None

        return order

这个 with 结构保证：

进入 with:
    创建 Span
    设置为当前 Span

执行 query_database:
    子操作可以找到当前 Span

离开 with:
    结束 Span
    恢复父 Span

如果发生异常，Span 应该记录异常并标记失败：

from opentelemetry.trace import Status, StatusCode

with tracer.start_as_current_span("charge_payment") as span:
    try:
        charge()
    except TimeoutError as exc:
        span.record_exception(exc)
        span.set_status(Status(StatusCode.ERROR, "payment timeout"))
        raise

OpenTelemetry Python 的手动埋点文档建议在记录异常时同时设置错误状态；Span 还可以添加 attributes、events 和 links。(opentelemetry.io)

不要把所有函数都创建成 Span。Span 应当对应一个能够解释耗时、错误或依赖关系的操作。为每个简单 getter 创建 Span，通常只会增加遥测量和查询噪声。

3. 属性、事件和日志的区别

可以用下面的规则区分：

数据	适合表达	例子
Span attribute	操作的稳定维度	`http.request.method=GET`
Span event	Span 生命周期中的重要瞬间	`cache_miss`
Log	详细离散事件和诊断信息	订单参数、异常栈
Metric	可聚合的数值	请求数、耗时、队列长度

Span attribute 适合筛选：

service.name = "order-api"
http.route = "/orders/{id}"
http.response.status_code = 500

但不应把高基数字段无节制放入指标标签。例如：

user_id
order_id
raw_url
exception_message

这些字段可能几乎每次都不同。它们更适合放入日志或 Trace 属性，而不是 Metrics 的标签维度。

六、Metrics：计数、仪表、直方图和高基数

1. Counter：只增不减的累计量

Counter 适合表示：

请求总数
处理成功总数
异常总数
重试总数

概念上：

C(t_2) \geq C(t_1), \quad t_2 \geq t_1

进程重启后计数器可能重新从零开始，后端通常通过时间序列的重置语义处理这一点。

错误示例：

# 不适合作为 Counter
counter.add(-1)

如果数值需要增加和减少，应该使用 UpDownCounter；如果表达当前值，应该使用 Gauge。

2. Gauge：某一时刻的值

Gauge 适合：

当前队列长度
当前活跃连接数
当前内存使用量
当前线程数

Gauge 的值可以上下变化：

G(t_2) \gtrless G(t_1)

它表达的是状态，不是事件累计。

3. Histogram：分布而不是平均数

请求耗时不能只记录平均值。假设 100 个请求中：

99 个请求耗时 10 ms
1 个请求耗时 10 s

平均值约为：

\frac{99 \times 10 + 10000}{100} = 109.9ms

平均值看起来并不夸张，但 1% 的请求已经严重超时。

Histogram 记录值的分布，可以计算：

P50；
P90；
P95；
P99；
指定桶中的请求数量。

因此，HTTP 请求耗时、数据库查询耗时、消息处理耗时通常适合 Histogram。

4. 指标命名和维度设计

指标名称应表达测量对象：

app.http.server.request.count
app.http.server.request.duration
app.payment.failure.count
app.queue.depth

维度应该有限且稳定：

method
route
status_code
service
dependency

推荐使用模板路由：

route="/orders/{order_id}"

而不是原始路径：

route="/orders/100001"
route="/orders/100002"
route="/orders/100003"

如果有 $k$ 个标签，每个标签分别有 $c_1, c_2, \ldots, c_k$ 种取值，理论时间序列数量上界约为：

N = \prod_{i=1}^{k} c_i

例如：

method: 4
route: 30
status: 6
region: 5

则最多：

4 \times 30 \times 6 \times 5 = 3600

如果再加入 100 万个用户 ID：

3600 \times 1{,}000{,}000 = 3.6 \times 10^9

这不仅增加存储成本，还可能使指标后端查询、内存和采集器处理能力失控。

七、OpenTelemetry：统一采集，不等于统一存储

OpenTelemetry 主要解决的是：

应用如何产生遥测数据
应用如何传播上下文
应用如何把数据导出到采集器或后端

它不要求日志、指标和 Trace 必须存入同一个数据库。统一的是：

资源信息；
服务身份；
上下文关联；
属性命名；
导出协议；
采集流程。

典型数据流如下：

flowchart LR
    A[Python 应用] --> B[Logger]
    A --> C[Meter]
    A --> D[Tracer]

    B --> E[日志处理器]
    C --> F[MetricReader]
    D --> G[SpanProcessor]

    E --> H[OTel Collector]
    F --> H
    G --> H

    H --> I[日志后端]
    H --> J[指标后端]
    H --> K[Trace 后端]

1. SDK 初始化顺序

Trace 的最小手动初始化示例：

from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    BatchSpanProcessor,
    ConsoleSpanExporter,
)

resource = Resource.create({
    "service.name": "order-api",
    "deployment.environment": "production",
})

provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(ConsoleSpanExporter())
)

trace.set_tracer_provider(provider)

tracer = trace.get_tracer("order-api")

然后创建 Span：

def handle_order(order_id: str) -> None:
    with tracer.start_as_current_span("handle_order") as span:
        span.set_attribute("app.order_id", order_id)
        logger.info("handling order")

这个示例使用控制台导出器，适合本地验证，不适合直接作为生产传输方式。生产环境通常使用 OTLP 导出到 OpenTelemetry Collector，再由 Collector 转发到后端。OpenTelemetry Python SDK 负责具体的采样、批处理和交付。(opentelemetry.io)

2. Metrics 的手动埋点

以下示例展示指标的基本生命周期：

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    ConsoleMetricExporter,
    PeriodicExportingMetricReader,
)

reader = PeriodicExportingMetricReader(
    ConsoleMetricExporter(),
    export_interval_millis=5000,
)

meter_provider = MeterProvider(
    resource=resource,
    metric_readers=[reader],
)

metrics.set_meter_provider(meter_provider)

meter = metrics.get_meter("order-api")

request_counter = meter.create_counter(
    "app.http.server.request.count",
    description="Number of HTTP requests",
)

request_duration = meter.create_histogram(
    "app.http.server.request.duration",
    unit="s",
    description="HTTP request duration",
)

记录数据：

import time


def handle_request(route: str, method: str) -> None:
    start = time.perf_counter()

    try:
        request_counter.add(
            1,
            {
                "http.route": route,
                "http.request.method": method,
            },
        )

        # 业务处理
    finally:
        duration = time.perf_counter() - start
        request_duration.record(
            duration,
            {
                "http.route": route,
                "http.request.method": method,
            },
        )

这里使用 time.perf_counter() 测量持续时间，而不是 time.time()。前者用于测量间隔，通常更适合耗时统计；后者可能受到系统时钟校正影响。

指标记录最好在请求完成时执行，这样异常路径也能进入 Histogram。若只在成功路径记录，失败请求会从耗时分布中消失，造成观测偏差。

3. OpenTelemetry 日志状态需要单独验证

OpenTelemetry Python 文档中的日志集成使用 Python 标准库 logging 作为日志入口，再由 OpenTelemetry LoggingHandler 处理。当前文档明确说明 Logs API 和 SDK 仍在开发中，并给出了 LoggerProvider、LoggingHandler 和日志导出器的示例。(opentelemetry.io)

因此，日志接入不应只凭一段示例代码判断“已经完成”。至少需要验证：

1. logging 是否仍然输出到原有目标
2. 是否出现重复日志
3. trace_id / span_id 是否正确关联
4. 异常栈是否保留
5. 进程退出时批处理数据是否 flush
6. 当前安装版本的导出器类名是否一致

OpenTelemetry 文档还提示，较早版本的控制台日志导出器名称存在差异，因此示例代码必须与实际锁定版本对应，而不能不加验证地复制到生产环境。(opentelemetry.io)

八、日志、Trace 和 Context 如何关联

最实用的关联字段通常是：

trace_id
span_id
request_id
service.name
deployment.environment

在存在当前 Span 时，可以从 OpenTelemetry Context 中读取当前 Span：

from opentelemetry import trace

span = trace.get_current_span()
span_context = span.get_span_context()

if span_context.is_valid:
    trace_id = format(span_context.trace_id, "032x")
    span_id = format(span_context.span_id, "016x")
else:
    trace_id = "-"
    span_id = "-"

然后通过 Filter 注入日志：

class TraceContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        span = trace.get_current_span()
        context = span.get_span_context()

        if context.is_valid:
            record.trace_id = format(context.trace_id, "032x")
            record.span_id = format(context.span_id, "016x")
        else:
            record.trace_id = "-"
            record.span_id = "-"

        return True

完整调用关系：

def create_order(order_id: str):
    with tracer.start_as_current_span("create_order"):
        logger.info(
            "creating order",
            extra={"order_id": order_id},
        )
        return save_order(order_id)

日志最终可以查询：

trace_id = "4bf92f3577b34da6a3ce929d0e0e4736"

然后从同一个 Trace 看到：

HTTP 请求耗时 820 ms
数据库查询耗时 40 ms
库存服务耗时 70 ms
支付服务耗时 700 ms
日志：payment timeout

这比单独搜索“payment timeout”更可靠，因为日志消息本身可能被重试、批处理或多个服务重复产生。

九、跨进程和跨服务：Propagation 才能连接 Trace

进程内 Context 不能自动穿过网络。服务 A 必须把 Trace Context 注入请求，服务 B 再提取它。

OpenTelemetry 默认使用 W3C Trace Context 和 W3C Baggage；常见 HTTP 头包括：

traceparent
baggage

Propagation 的作用是把因果上下文从一个服务或进程移动到另一个服务。对于 Flask、Django、Celery 等常见 Python 框架，OpenTelemetry instrumentation libraries 可以自动完成部分传播。(opentelemetry.io)

手动注入示例：

from opentelemetry.propagate import inject


headers: dict[str, str] = {}
inject(headers)

# 将 headers 发送给下游服务

手动提取示例：

from opentelemetry.propagate import extract
from opentelemetry import trace

headers = {
    "traceparent": incoming_traceparent,
}

parent_context = extract(headers)

with tracer.start_as_current_span(
    "process_downstream_request",
    context=parent_context,
):
    handle_request()

传播路径是：

服务 A 当前 Span
    ↓ inject(headers)
HTTP 请求
    ↓
服务 B extract(headers)
    ↓
服务 B 创建子 Span

如果服务 B 没有提取上下文，它仍然可以创建 Span，但这个 Span 会成为新的 Trace 根节点，调用链就会断裂。

1. 不要把所有业务字段都放进 Baggage

Baggage 是随请求传播的键值上下文。它适合放：

tenant_id
routing_group
experiment_group

不适合放：

密码
访问令牌
完整用户对象
大型 JSON
高敏感业务数据

传播意味着数据可能经过多个服务和网络边界。任何进入 Baggage 的数据都应该经过安全审查。

2. 消息队列不是普通父子调用

HTTP 调用通常是：

A Span
└── B Span

但消息队列场景可能是：

生产者 Span ──发送消息──> 消息
                              ↓
                         消费者 Span

如果一条消息被多个消费者处理，或者一个批次包含多个来源消息，简单设置单一父 Span 可能无法准确表达因果关系。此时可以使用 Span Links 表达“这个操作与多个来源有关”，而不是强行构造一棵单父树。OpenTelemetry Python 文档将 Span Link 定义为因果关联但非父子关系。(opentelemetry.io)

十、采样：为什么不是每条 Trace 都保留

生产系统中，Trace 可能数量巨大。采样决定哪些 Trace 被记录和导出。

设总请求数为 $R$ ，采样率为 $p$ ，理论导出量约为：

E = R \times p

例如：

每秒请求数 R = 1000
采样率 p = 0.1

则平均每秒导出约：

1000 \times 0.1 = 100

但采样会改变可见数据。若错误率为 $q$ ，采样后观察到的错误 Trace 期望数量约为：

R \times q \times p

当 $q$ 很小时，固定概率采样可能长期看不到错误请求。

因此常见策略是：

普通成功请求进行概率采样；
错误请求提高采样率；
慢请求提高采样率；
特定租户、接口或发布版本暂时提高采样率。

采样必须尽早决定。若在应用已经创建大量 Span、生成大量日志之后才丢弃，后端成本可能下降，但应用 CPU、内存和网络开销未必下降。

十一、故障定位：从症状到根因的推导流程

场景：接口成功率下降且延迟上升

假设指标显示：

过去 10 分钟：
request_count        = 120000
error_count          = 3600
error_rate            = 3.0%
P95 latency           = 1.8s
P99 latency           = 6.4s

第一步不是立即重启，而是按维度拆分：

service
route
status_code
dependency
deployment_version
region

如果发现：

route="/orders/{id}"
version="2026.09.01"
error_rate=12%

而旧版本为：

version="2026.08.31"
error_rate=0.3%

此时发布版本是强相关线索，但还不是根因。

第二步，从错误 Trace 中查看 Span 树：

Trace T1
└── HTTP GET /orders/{id}       2.4s ERROR
    ├── database.query          30ms OK
    ├── inventory.check         80ms OK
    └── payment.authorize       2.2s ERROR
        └── http.client         2.2s timeout

第三步，按 trace_id 查询日志：

{
  "severity": "ERROR",
  "message": "payment timeout",
  "order_id": "o-1001",
  "retry_count": 3,
  "timeout_ms": 500,
  "trace_id": "T1"
}

第四步，比较时间关系：

单次调用超时：500 ms
重试次数：3
总耗时：约 2.2 s

如果重试之间存在退避，那么总耗时可以近似写为：

T_{total} = \sum_{i=1}^{n} (T_{request_i} + T_{backoff_i}) + T_{local}

其中：

$n$ 是尝试次数；
$T_{request_i}$ 是第 $i$ 次请求耗时；
$T_{backoff_i}$ 是第 $i$ 次重试前等待时间；
$T_{local}$ 是本地处理时间。

由此可以解释为什么 P99 从几百毫秒上升到数秒：不是数据库变慢，而是下游支付服务超时触发了重试。

第五步，确认故障路径：

新版本发布
    ↓
支付客户端超时配置或重试策略变化
    ↓
支付下游少量超时
    ↓
每个请求进行多次重试
    ↓
请求耗时增加
    ↓
连接池占用上升
    ↓
更多请求排队
    ↓
错误率和 P99 继续上升

这里已经从“接口变慢”推导到了“下游超时 + 重试放大 + 连接池压力”的故障模型。

十二、常见反例：为什么看似有观测，实际上无法定位

反例一：日志只有字符串，没有字段

logger.error(f"request failed: {exc}")

问题：

异常类型不稳定；
没有请求 ID；
没有 Trace ID；
没有路由、版本和依赖信息；
可能丢失异常栈。

改为：

logger.exception(
    "request failed",
    extra={
        "route": route,
        "request_id": request_id,
        "dependency": "payment",
    },
)

反例二：所有东西都使用 `ERROR`

logger.error("cache miss")
logger.error("user input invalid")
logger.error("database unavailable")

这会让告警失去区分度。

更合理的区分是：

DEBUG   诊断细节
INFO    正常业务事件
WARNING 可恢复异常或退化
ERROR   当前操作失败
CRITICAL 进程或系统级不可用

级别不是严重程度的唯一指标。一个可预期的用户输入错误，可能应该计入业务指标，但不应该触发系统级告警。

反例三：指标标签使用原始 URL

duration.record(
    elapsed,
    {"path": request.path},
)

如果 request.path 包含订单号、用户 ID 或随机 Token，时间序列会持续膨胀。

应使用路由模板：

{"http.route": "/orders/{order_id}"}

原始 URL 可以放到受控的日志字段中，但也要注意敏感信息和长度限制。

反例四：每个异常都创建新的 Counter

错误示意：

def handle(exc):
    counter = meter.create_counter(
        f"error.{type(exc).__name__}"
    )
    counter.add(1)

问题是指标工具对象可能不断创建，且异常类型或消息可能形成不受控的指标名称。

应在初始化阶段创建固定名称的指标：

error_counter = meter.create_counter(
    "app.request.error.count"
)

error_counter.add(
    1,
    {"error.type": type(exc).__name__},
)

即使 error.type 也应限制为有限集合，避免把动态异常消息当作标签。

反例五：只记录入口 Span，不记录关键依赖

HTTP /orders/{id}  5s

这只能说明接口慢，不能说明慢在哪里。

至少应为有明显延迟或失败可能的依赖建立 Span：

数据库
缓存
外部 HTTP
消息队列
文件或对象存储
锁等待
线程池任务

但这不意味着每一层函数都要埋点。埋点粒度应该能够解释性能和错误，而不是机械覆盖所有函数。

反例六：Context 只在成功路径清理

token = request_id_var.set(request_id)
handle()
request_id_var.reset(token)

如果 handle() 抛异常，reset 不会执行。

必须使用：

token = request_id_var.set(request_id)
try:
    handle()
finally:
    request_id_var.reset(token)

或者在 Python 3.14 中使用：

with request_id_var.set(request_id):
    handle()

十三、并发、进程模型和日志完整性

1. 多线程

多个线程共享进程内 Logger 配置。标准 logging 的 Handler 通常通过锁保护写入，但这不等于所有自定义 Handler 都线程安全。

自定义 Handler 至少要考虑：

多个线程同时 emit
Formatter 是否持有可变状态
缓冲区是否需要锁
flush 是否安全
关闭时是否丢数据

2. 多进程

多进程部署时，每个进程拥有：

独立的 Python 堆；
独立的 Logger 对象；
独立的 Context；
独立的指标 SDK 状态；
独立的 BatchSpanProcessor 缓冲区。

因此：

进程 A 的 Context 不会自动出现在进程 B
进程 A 的 Counter 不等于全局 Counter
进程 A 的日志文件可能与进程 B 交错或冲突

多进程日志不要让多个进程直接竞争写同一个普通文件。更稳妥的方案是：

每个进程输出 stdout/stderr
    ↓
运行时或采集器统一收集

如果必须文件输出，应明确使用多进程安全的架构，例如集中式日志进程或进程间队列，而不是仅仅给每个进程配置同一个 RotatingFileHandler。

3. 异常退出和批处理导出

BatchSpanProcessor、BatchLogRecordProcessor 等批处理组件会先把数据放到内存中，再异步导出。优点是减少每次业务操作的同步开销，代价是：

进程突然崩溃时，缓冲区数据可能丢失；
关闭时需要 flush；
导出线程异常时，业务线程可能感知不到；
队列满时可能丢弃或阻塞，具体行为取决于实现和配置。

发布、滚动重启和进程退出时，应验证 SDK 的 shutdown 流程。不能假设“进程收到停止信号后所有遥测数据自然会写完”。

十四、配置和交付：可观测性也是生产配置

可观测性配置至少包括：

service.name
deployment.environment
service.version
采样率
导出端点
超时
批处理大小
队列容量
日志级别
敏感字段处理规则

服务身份必须稳定：

service.name = order-api
service.version = 2026.09.01
deployment.environment = production

不要把 Pod 名称、随机进程 ID 或完整容器 ID 当作 service.name，否则同一个服务会被拆成大量服务实体。

配置迁移时，需要同时验证业务和遥测：

1. 新版本是否仍能启动
2. Logger 是否重复输出
3. 指标名称和标签是否兼容
4. Trace 是否还能跨服务连接
5. 日志是否仍含 trace_id
6. 导出失败是否影响业务请求
7. 回滚后旧版本是否还能理解新字段

灰度发布可以利用可观测性比较两个版本：

version="old"
version="new"

比较维度包括：

错误率
P50/P95/P99
外部依赖耗时
超时率
重试率
进程 CPU
进程 RSS
线程池队列长度
遥测丢弃数

回滚条件不应只看平均延迟。例如：

新版本 P99 延迟连续 5 分钟超过旧版本 2 倍
或支付错误率超过 1%
或遥测导出队列持续满载

阈值必须结合业务 SLO、容量和依赖特性设定，不能直接套用固定数字。

十五、如何验证一套可观测性实现确实有效

可以用一个最小端到端测试验证：

客户端发送请求
    ↓
服务创建根 Span
    ↓
设置 request_id / tenant_id
    ↓
调用内部函数
    ↓
记录指标
    ↓
记录带异常栈的日志
    ↓
导出 Trace、Metric、Log
    ↓
用 trace_id 查询三类数据

测试代码不应只断言“程序没有异常”，而应检查以下结果：

日志

是否有 service.name
是否有 trace_id
是否有 span_id
是否有结构化业务字段
异常日志是否包含堆栈

指标

请求总数是否增加
异常请求是否进入错误计数
耗时是否记录到 Histogram
标签是否来自有限集合

Trace

根 Span 是否存在
子 Span 是否有正确 parent
异常 Span 是否被标记为 ERROR
Span 结束时间是否晚于开始时间
跨服务 trace_id 是否保持一致

Context

并发请求是否互不串线
异常路径是否恢复旧值
线程池任务是否显式传播
后台任务结束后是否清理

一个实用的并发隔离测试：

import asyncio
from contextvars import ContextVar

request_id = ContextVar("request_id", default="-")


async def worker(value: str) -> str:
    token = request_id.set(value)
    try:
        await asyncio.sleep(0)
        return request_id.get()
    finally:
        request_id.reset(token)


async def main():
    result = await asyncio.gather(
        worker("A"),
        worker("B"),
        worker("C"),
    )
    print(result)


asyncio.run(main())

预期输出应保持：

['A', 'B', 'C']

如果输出出现：

['C', 'C', 'C']

通常意味着实现错误地使用了共享全局变量，而不是任务局部上下文。

十六、四类数据如何协同定位不同故障

1. 错误率上升

优先使用：

Metric → Trace → Log

指标按路由、版本、状态码拆分；Trace 找到代表性失败请求；日志查看参数、异常类型和重试细节。

2. 延迟上升但错误率不变