分布式链路追踪落地实践：一个Span的一生

分布式链路追踪落地实践：一个Span的一生

一、一个跨越6个微服务的请求

刷题系统的"提交评测"接口涉及6个微服务：API Gateway → Auth Service → Problem Service → Judge Scheduler → Judge Worker → Result Aggregator。当这个请求超时时，没有任何一个服务说"是我慢了"。

在分布式系统中，一个请求会跨越多个服务、多个线程、多个网络调用。如果没有链路追踪，你只能看到孤立的错误日志和监控指标，却无法还原一次完整请求的"行走轨迹"。

链路追踪的核心理念很简单：给每个请求一个全局唯一的 TraceID，在每个服务边界创建一个 Span 来记录本地操作，最终将所有 Span 按父子关系串联成一颗调用树。

flowchart TB A[用户请求] -->|TraceID: abc123| B[API Gateway<br/>Span: handle_request<br/>50ms] B --> C[Auth Service<br/>Span: authenticate<br/>10ms] B --> D[Problem Service<br/>Span: query_problem<br/>45ms] D --> E[MySQL<br/>Span: db_query<br/>8ms] B --> F[Judge Scheduler<br/>Span: schedule_judge<br/>380ms] F --> G[Judge Worker<br/>Span: execute_code<br/>350ms] G --> H[Docker Sandbox<br/>Span: compile<br/>80ms] G --> I[Docker Sandbox<br/>Span: run_tests<br/>260ms] F --> J[Result Aggregator<br/>Span: aggregate_result<br/>30ms] B -.-> K[上报到 Jaeger/Zipkin] C -.-> K D -.-> K E -.-> K F -.-> K G -.-> K H -.-> K I -.-> K J -.-> K style A fill:#ccf style K fill:#cfc

二、Span 的生命周期与传播机制

2.1 Span 的核心字段

一个 Span 记录了以下关键信息：

2.2 上下文传播：TraceID 如何跨服务传递

最关键的问题是：TraceID 和 SpanID 怎么从上一个服务传递到下一个服务？

标准做法是在 HTTP 请求头中注入以下字段（W3C TraceContext 标准）：

traceparent: 00-abc123def456-span001-01 tracestate: vendor=opentelemetry

当 API Gateway 创建根 Span 时，生成 TraceID 和 SpanID，并在调用下游服务前将它们注入到请求头。下游服务从请求头中提取 TraceID，创建自己的 Span（ParentSpanID = 上游 SpanID），形成父子关系。

2.3 采样策略：不是每个请求都追踪

在高 QPS 场景下，记录所有 Span 会产生海量数据。采样策略决定了哪些请求被追踪：

• 全量采样：追踪所有请求（低 QPS 或调试场景）
• 固定比例采样：如 10% 的请求（简单高效）
• 自适应采样：正常请求采样率低（1%），慢请求/错误请求强制采样（100%）

三、最小化链路追踪的实现

以下代码展示了一个零依赖的轻量级链路追踪实现，包含 Span 管理、上下文传播（HTTP 中间件）、采样策略和内存上报器。

""" 最小化分布式链路追踪实现 功能：Span 管理 + HTTP 上下文传播 + 采样 + 上报 """ import time import uuid import random import json import threading from typing import Dict, List, Optional, Any, Callable from dataclasses import dataclass, field from contextvars import ContextVar from collections import defaultdict # ==================== 核心数据结构 ==================== @dataclass class SpanContext: """Span 的上下文信息（用于跨服务传播）""" trace_id: str # 全局链路 ID span_id: str # 当前 Span ID parent_span_id: str = "" # 父 Span ID sampled: bool = True # 是否被采样 baggage: Dict[str, str] = field(default_factory=dict) # 业务透传数据 def to_http_headers(self) -> Dict[str, str]: """序列化为 HTTP 请求头（W3C TraceContext 简化版）""" return { "X-Trace-Id": self.trace_id, "X-Span-Id": self.span_id, "X-Parent-Span-Id": self.parent_span_id, "X-Sampled": "1" if self.sampled else "0", } @classmethod def from_http_headers(cls, headers: Dict[str, str]) -> "SpanContext": """从 HTTP 请求头反序列化""" return cls( trace_id=headers.get("X-Trace-Id", ""), span_id=headers.get("X-Span-Id", ""), parent_span_id=headers.get("X-Parent-Span-Id", ""), sampled=headers.get("X-Sampled", "1") == "1", ) @dataclass class SpanEvent: """Span 内的事件（用于记录关键时间点）""" name: str # 事件名称 timestamp_us: int # 微秒时间戳 attributes: Dict[str, Any] = field(default_factory=dict) @dataclass class Span: """一次操作的完整记录""" context: SpanContext operation_name: str # 操作名称 start_time_us: int # 开始时间（微秒） end_time_us: int = 0 # 结束时间（微秒） tags: Dict[str, str] = field(default_factory=dict) # 标签 events: List[SpanEvent] = field(default_factory=list) # 事件列表 status: str = "OK" # 状态：OK / ERROR @property def duration_ms(self) -> float: """持续时间（毫秒）""" if self.end_time_us == 0: return 0.0 return (self.end_time_us - self.start_time_us) / 1000.0 def add_event(self, name: str, **attributes): """添加事件""" self.events.append(SpanEvent( name=name, timestamp_us=int(time.time() * 1_000_000), attributes=attributes, )) def set_tag(self, key: str, value: str): """设置标签""" self.tags[key] = value def finish(self, error: Optional[Exception] = None): """结束 Span""" self.end_time_us = int(time.time() * 1_000_000) if error: self.status = "ERROR" self.set_tag("error", str(error)) self.set_tag("error.type", type(error).__name__) # ==================== Tracer 核心 ==================== # 使用 ContextVar 实现跨协程的上下文传播 _CURRENT_SPAN: ContextVar[Optional[Span]] = ContextVar( "current_span", default=None ) class Reporter: """上报器接口（内存实现，生产环境替换为 Jaeger/OpenTelemetry）""" def __init__(self): self._spans: List[Span] = [] self._lock = threading.Lock() def report(self, span: Span): """上报一个 Span""" with self._lock: self._spans.append(span) def get_all_spans(self) -> List[Span]: """获取所有 Span（用于调试）""" with self._lock: return list(self._spans) def get_trace_tree(self, trace_id: str) -> Dict[str, List[Span]]: """按 TraceID 获取调用树""" trace_spans = [s for s in self.get_all_spans() if s.context.trace_id == trace_id] # 按父 SpanID 分组 tree = defaultdict(list) for span in trace_spans: tree[span.context.parent_span_id].append(span) return dict(tree) class Sampler: """采样器""" def __init__(self, sample_rate: float = 0.1): """ Args: sample_rate: 采样率（0.0 ~ 1.0） """ self.sample_rate = sample_rate def should_sample(self, trace_id: str) -> bool: """判断是否采样（基于 TraceID 的哈希，确保同一 TraceID 采样一致）""" # 使用 TraceID 的哈希确保一致性 hash_val = hash(trace_id) % 100 return hash_val < (self.sample_rate * 100) class Tracer: """ 链路追踪器 API 设计参考 OpenTelemetry，简化了实现 """ def __init__(self, service_name: str = "unknown", sampler: Optional[Sampler] = None, reporter: Optional[Reporter] = None): self.service_name = service_name self.sampler = sampler or Sampler(sample_rate=0.1) self.reporter = reporter or Reporter() def start_span(self, operation_name: str, parent_ctx: Optional[SpanContext] = None, **tags) -> Span: """ 开始一个新的 Span Args: operation_name: 操作名称 parent_ctx: 父 Span 的上下文（用于跨服务传播） **tags: 初始标签 Returns: Span 对象 """ # 生成 ID trace_id = parent_ctx.trace_id if parent_ctx else uuid.uuid4().hex span_id = uuid.uuid4().hex[:16] parent_span_id = parent_ctx.span_id if parent_ctx else "" # 采样决策 sampled = parent_ctx.sampled if parent_ctx else ( self.sampler.should_sample(trace_id) ) context = SpanContext( trace_id=trace_id, span_id=span_id, parent_span_id=parent_span_id, sampled=sampled, ) span = Span( context=context, operation_name=operation_name, start_time_us=int(time.time() * 1_000_000), tags={ "service.name": self.service_name, **tags, }, ) # 设置当前 Span（用于子 Span 自动关联） _CURRENT_SPAN.set(span) return span def get_current_span(self) -> Optional[Span]: """获取当前活跃的 Span""" return _CURRENT_SPAN.get() def finish_span(self, span: Span, error: Optional[Exception] = None): """结束 Span 并上报""" span.finish(error) if span.context.sampled: self.reporter.report(span) # 清理当前 Span _CURRENT_SPAN.set(None) # ==================== HTTP 中间件 ==================== def trace_middleware(tracer: Tracer): """ 链路追踪 HTTP 中间件（简化版） 用法：在 Flask/FastAPI 的 middleware 中调用 """ def decorator(handler: Callable) -> Callable: def wrapper(request, *args, **kwargs): # 提取上游传递的 TraceID headers = getattr(request, 'headers', {}) parent_ctx = SpanContext.from_http_headers(headers) \ if headers.get("X-Trace-Id") else None # 创建根 Span span = tracer.start_span( operation_name=f"{request.method} {getattr(request, 'path', '/')}", parent_ctx=parent_ctx, http_method=getattr(request, 'method', 'GET'), http_url=str(getattr(request, 'url', '')), ) try: # 执行请求处理 response = handler(request, *args, **kwargs) span.set_tag("http.status_code", str(getattr(response, 'status_code', 200))) span.set_tag("status", "OK") return response except Exception as e: span.set_tag("status", "ERROR") span.set_tag("error", str(e)) span.finish(error=e) raise finally: tracer.finish_span(span) return wrapper return decorator # ==================== 使用示例 ==================== class MockRequest: """模拟 HTTP 请求""" def __init__(self, method="GET", path="/api/submit", headers=None): self.method = method self.path = path self.url = f"http://localhost{path}" self.headers = headers or {} class MockResponse: """模拟 HTTP 响应""" def __init__(self, status_code=200): self.status_code = status_code def demo_tracing(): """演示完整的链路追踪流程""" tracer = Tracer(service_name="api-gateway") # 模拟一个请求处理 @trace_middleware(tracer) def handle_submit(request): # 模拟调用 Auth Service（创建子 Span） auth_span = tracer.start_span( "authenticate", parent_ctx=tracer.get_current_span().context, ) auth_span.set_tag("user_id", "user_123") time.sleep(0.01) # 模拟 10ms 的认证耗时 tracer.finish_span(auth_span) # 模拟调用 Judge Scheduler（创建子 Span） judge_span = tracer.start_span( "schedule_judge", parent_ctx=tracer.get_current_span().context, ) time.sleep(0.15) # 模拟 150ms 的调度耗时 judge_span.add_event("worker_assigned", worker_id="worker-05") tracer.finish_span(judge_span) return MockResponse(status_code=200) # 传入带上游 TraceID 的请求 upstream_trace_id = "abc123def456" request = MockRequest( method="POST", path="/api/submit", headers={"X-Trace-Id": upstream_trace_id}, ) response = handle_submit(request) print(f"Response: {response.status_code}") # 查看链路树 spans = tracer.reporter.get_all_spans() print(f"\n=== 链路追踪报告 ({len(spans)} spans) ===") for span in spans: print(f" [{span.context.parent_span_id or 'ROOT'} -> " f"{span.context.span_id}] " f"{span.operation_name}: " f"{span.duration_ms:.1f}ms ({span.status})") for event in span.events: print(f" Event: {event.name} {event.attributes}") if __name__ == "__main__": demo_tracing()

四、生产级链路追踪的关键工程问题

数据量控制：一个高 QPS 系统每天产生数十亿个 Span。即使 1% 采样，也是海量数据。解决方案：采样 + 聚合 + TTL 过期。只保留最近 24-48 小时的详细 Trace。

跨语言传播的一致性：W3C TraceContext 标准已在主流语言中落地（Java/Go/Python/Node.js），但不同 SDK 的实现细节（如 baggage 的序列化格式）仍有差异。统一使用 OpenTelemetry 是当前最佳实践。

Span 数量爆炸：如果不加控制，一个"SELECT *"可能为每一行生成一个 Span。需要在业务层面限制 Span 的粒度——通常每个 RPC 调用一个 Span、每次数据库查询一个 Span 即可。

存储成本：全量 Span 的存储成本极高。常用策略是将 Span 采样上报到 Jaeger/Zipkin（热数据），同时将聚合指标（如延迟百分位、错误率）写入 Prometheus（长期存储）。

五、总结

链路追踪本质上是为分布式请求建立一条时间线和一张关系图：

1. TraceID 是全局唯一标识：它像行李标签一样贴在每个请求上，跨服务不丢失。
2. Span 是追踪的最小单元：记录一个操作的开始/结束时间、标签、事件。
3. 上下文传播是链路完整性的生命线：W3C TraceContext 标准只需两个请求头就完成了 TraceID 和采样决策的传递。
4. 采样是'不太贵'的前提：10% 的采样 + 慢请求强制采样能在成本和价值间取得最佳平衡。

实施链路追踪最划算的起步方式是用 OpenTelemetry + Jaeger，只需添加一个 SDK 和少量配置，就能获得跨服务的调用链可视化。这是我在实习中学到的"ROI 最高的基础设施"。

本文从零实现了一个轻量级链路追踪系统，涵盖 Span 生命周期、HTTP 上下文传播、采样策略和内存上报器。代码中的 Tracer、SpanContext 和中间件可直接用于理解和原型验证分布式追踪的核心原理。