构建分布式追踪收集器

项目概述

本项目实现一个面向 trace collection 的完整系统，核心要解决的问题是 accepting spans from many services and assembling them into traces with correct causal order。整体由 Span ingest、Trace assembler、Span store、Sampling policy、Tail-based sampler 五个模块组成，每个模块单一职责、可独立测试，最后通过集成测试验证端到端正确性。设计上把 a dropped span that leaves a trace with a broken parent link and an incomplete call tree 作为第一类需求先行处理，而非事后再补。

架构与关键决策

数据模型为 a store of spans keyed by trace id, assembled into a tree by parent span references and retained for a configurable window。Span ingest 定义其余阶段共同依赖的数据形状并在边界处校验输入；Trace assembler 承载 trace collection 的正确性逻辑；Span store 专门处理 a dropped span that leaves a trace with a broken parent link and an incomplete call tree，在失败路径上给出明确的拒绝、重试或降级策略；Sampling policy 提供可观测性，每次操作输出带关联 id 的结构化事件；Tail-based sampler 在过载时按既定方式降级，避免过载演变为数据损坏。配置与代码分离，使风险变更可经 diff 审阅。

实施与测试

按五个阶段依次实现 Span ingest、Trace assembler、Span store、Sampling policy、Tail-based sampler，每阶段配套聚焦单测。集成测试组合各模块并驱动真实 trace collection 场景：先验证正常路径，再刻意触发 a dropped span that leaves a trace with a broken parent link and an incomplete call tree 并断言系统降级而非损坏。所有依赖均在进程内或被桩替换，用例在 CI 中确定可复现。

限制与后续

• 单区域设计，多区域复制超出范围。
• Span ingest 默认进程内，未水平共享。
• 无鉴权层，假定部署在强制身份的边缘之后。

代码示例

# setup for distributed-tracing-collector
pnpm init
pnpm add fastify opentelemetry level
pnpm add -D typescript tsx vitest

distributed_tracing_collector/
├── src/index.ts
├── src/span-ingest.ts
├── src/trace-assembler.ts
├── tests/distributed_tracing_collector.test.ts
└── package.json

// Span ingest: trace collection core
export function spaningest(input: SpaningestInput): SpaningestResult {
  // Build the span ingest that anchors trace collection and defines the core data shape.
  return processSpaningest(input);
}

// Trace assembler — handles trace collection Add the trace assembler so
export class Traceassembler {
  constructor(private opts: TraceassemblerOptions) {}
  async run(ctx: Context): Promise<TraceassemblerOutput> {
    return this.execute(ctx);
  }
}

// Span store: trace collection stage
export const spanstore = (next: Next) => async (req: Request) => {
  const out = await transform(req);
  return next(out);
};

// Sampling policy for trace collection
export async function samplingpolicy(stream: AsyncIterable<Chunk>): Promise<Aggregate> {
  let acc = init();
  for await (const c of stream) acc = merge(acc, c);
  return acc;
}

// Tail-based sampler: trace collection finalization
export function tailbasedsampler(state: State): Result {
  validate(state);
  return finalize(state);
}

# run the trace collection test suite
pnpm vitest run