Workers 下的 Tracing 实现

// span.ts
interface Span {
 
  spanId: string            // 当前节点的唯一标识
 
  parentSpanId: string | null // 父节点 ID，用于构建树形结构
 
  name: string              // 节点名称，如 'memory_retrieval'
 
  startTime: number         // 开始执行的时间戳
 
  duration: number          // 执行耗时（毫秒）
 
  status: 'ok' | 'error' | 'degraded'  // 执行状态
 
  input: Record<string, any>   // 节点接收的输入
 
  output: Record<string, any>  // 节点产生的输出
 
  metadata: Record<string, any> // 额外的诊断信息
 
}

// trace.ts
class TraceContext {
 
  readonly traceId: string
 
  private spans: Span[] = []
 
  private startTime: number
 
  constructor(traceId?: string) {
 
    // 如果外部传入了 traceId 就用外部的，否则自动生成一个
 
    // 这在跨服务追踪时很有用：上游服务可以把 traceId 传下来
 
    this.traceId = traceId ?? crypto.randomUUID()
 
    this.startTime = Date.now()
 
  }
 
  // 核心方法：包装一个异步操作，自动记录为 Span
 
  // <T> 是 TypeScript 的泛型——意思是"返回类型取决于你传入的函数返回什么"
 
  // 比如 fn 返回 Promise<string>，那 span() 的返回值也是 Promise<string>
 
  async span<T>(
 
    name: string,
 
    input: Record<string, any>,
 
    fn: () => Promise<T>,
 
    parentSpanId?: string
 
  ): Promise<T> {
 
    const spanId = crypto.randomUUID()
 
    const start = Date.now()
 
    let status: Span['status'] = 'ok'
 
    let output: Record<string, any> = {}
 
    try {
 
      const result = await fn()
 
      output = { result }
 
      return result
 
    } catch (err) {
 
      status = 'error'
 
      output = { error: (err as Error).message }
 
      throw err  // 记录完错误后继续抛出，不吞掉异常
 
    } finally {
 
      // 无论成功还是失败，都会执行 finally 块
 
      // 确保这个 Span 一定会被记录下来
 
      this.spans.push({
 
        spanId,
 
        parentSpanId: parentSpanId ?? null,
 
        name,
 
        startTime: start,
 
        duration: Date.now() - start,
 
        status,
 
        input,
 
        output,
 
        metadata: {}
 
      })
 
    }
 
  }
 
  // 手动将某个已记录的 Span 标记为降级状态
 
  // 适用于"执行成功了，但结果不理想"的场景
 
  markDegraded(spanName: string, reason: string) {
 
    const span = this.spans.find(s => s.name === spanName)
 
    if (span) {
 
      span.status = 'degraded'
 
      span.metadata.degradeReason = reason
 
    }
 
  }
 
  // 判断是否有任何 Span 出错
 
  hasError(): boolean {
 
    return this.spans.some(s => s.status === 'error')
 
  }
 
  // 判断是否有任何 Span 降级
 
  hasDegraded(): boolean {
 
    return this.spans.some(s => s.status === 'degraded')
 
  }
 
  // 获取从创建到现在的总耗时
 
  get totalDuration(): number {
 
    return Date.now() - this.startTime
 
  }
 
  // 将整个 Trace 序列化为 JSON，方便存储和传输
 
  toJSON() {
 
    return {
 
      traceId: this.traceId,
 
      totalDuration: this.totalDuration,
 
      spanCount: this.spans.length,
 
      spans: this.spans
 
    }
 
  }
 
}

// usage.ts
// 1. 在请求入口创建 TraceContext 实例
 
const trace = new TraceContext()
 
// 2. 用 trace.span() 包装每个管线节点
 
//    第一个参数是节点名称，第二个是输入数据，第三个是要执行的异步函数
 
const safetyResult = await trace.span(
 
  'input_safety_check',
 
  { message: userMessage },
 
  () => runSafetyCheck(userMessage)
 
)
 
const memories = await trace.span(
 
  'memory_retrieval',
 
  { query: userMessage, sessionId },
 
  () => retrieveMemories(env, userMessage, sessionId)
 
)
 
// 3. 根据执行结果，手动标记降级状态
 
//    记忆检索成功了（没报错），但结果为空，说明效果不理想
 
if (memories.length === 0) {
 
  trace.markDegraded('memory_retrieval', 'no_memories_found')
 
}
 
// 4. 后续节点继续用 trace.span() 包装...
 
const response = await trace.span(
 
  'llm_generation',
 
  { prompt: assembledPrompt },
 
  () => callLLM(assembledPrompt)
 
)

// sampling.ts
// 采样配置
 
interface SamplingConfig {
 
  latencyP99Threshold: number  // 延迟阈值（毫秒），超过这个值视为异常慢
 
  normalSampleRate: number     // 正常请求的采样比例，如 0.05 表示 5%
 
}
 
// 采样决策结果
 
interface SampleDecision {
 
  store: 'full' | 'summary'   // 完整存储 or 仅存摘要
 
  destination: 'd1' | 'kv'    // 存到 D1 数据库 or KV 键值存储
 
}
 
function shouldSample(
 
  trace: TraceContext,
 
  config: SamplingConfig
 
): SampleDecision {
 
  // 规则 1：有错误的请求，必须完整保留
 
  // 这是最重要的数据，排查问题全靠它
 
  if (trace.hasError()) {
 
    return { store: 'full', destination: 'd1' }
 
  }
 
  // 规则 2：有降级的请求，也完整保留
 
  // 降级请求往往能解释"为什么回复质量变差了"
 
  if (trace.hasDegraded()) {
 
    return { store: 'full', destination: 'd1' }
 
  }
 
  // 规则 3：耗时异常高的请求，完整保留
 
  // 这些是性能问题的直接证据
 
  if (trace.totalDuration > config.latencyP99Threshold) {
 
    return { store: 'full', destination: 'd1' }
 
  }
 
  // 规则 4：正常请求按比例随机采样
 
  // 比如 normalSampleRate = 0.05，则只有 5% 的正常请求会被完整保存
 
  if (Math.random() < config.normalSampleRate) {
 
    return { store: 'full', destination: 'd1' }
 
  }
 
  // 规则 5：其余正常请求只存摘要
 
  return { store: 'summary', destination: 'kv' }
 
}

// schema.sql
-- traces 表：每条记录代表一次完整请求
 
CREATE TABLE traces (
 
  trace_id TEXT PRIMARY KEY,          -- 请求唯一标识
 
  user_id TEXT NOT NULL,              -- 用户 ID
 
  session_id TEXT NOT NULL,           -- 会话 ID
 
  total_duration INTEGER NOT NULL,    -- 请求总耗时（毫秒）
 
  has_error BOOLEAN DEFAULT FALSE,    -- 是否有节点报错
 
  has_degraded BOOLEAN DEFAULT FALSE, -- 是否有节点降级
 
  created_at TEXT NOT NULL            -- 创建时间
 
);
 
-- spans 表：每条记录代表一个管线节点的执行记录
 
CREATE TABLE spans (
 
  span_id TEXT PRIMARY KEY,           -- 节点唯一标识
 
  trace_id TEXT NOT NULL,             -- 关联的请求 ID
 
  parent_span_id TEXT,                -- 父节点 ID，用于还原树形调用层级
 
  name TEXT NOT NULL,                 -- 节点名称
 
  start_time INTEGER NOT NULL,        -- 开始时间戳，用于还原各节点的执行时序
 
  duration INTEGER NOT NULL,          -- 节点耗时（毫秒）
 
  status TEXT NOT NULL,               -- 状态：ok / error / degraded
 
  input TEXT,                         -- 输入数据（JSON 字符串）
 
  output TEXT,                        -- 输出数据（JSON 字符串）
 
  metadata TEXT,                      -- 诊断附加信息（JSON 字符串）
 
  FOREIGN KEY (trace_id) REFERENCES traces(trace_id)
 
);
 
-- 索引：加速常用的查询场景
 
CREATE INDEX idx_traces_user ON traces(user_id, created_at);
 
CREATE INDEX idx_traces_error ON traces(has_error, created_at);
 
CREATE INDEX idx_spans_trace ON spans(trace_id);

// store-trace.ts
async function storeFullTrace(
 
  env: { DB: D1Database },
 
  trace: TraceContext,
 
  userId: string,
 
  sessionId: string
 
) {
 
  const data = trace.toJSON()
 
  // 第一步：写入 traces 主表
 
  await env.DB.prepare(
 
    `INSERT INTO traces
 
      (trace_id, user_id, session_id, total_duration, has_error, has_degraded, created_at)
 
     VALUES (?, ?, ?, ?, ?, ?, ?)`
 
  ).bind(
 
    data.traceId,
 
    userId,
 
    sessionId,
 
    data.totalDuration,
 
    data.spans.some(s => s.status === 'error'),
 
    data.spans.some(s => s.status === 'degraded'),
 
    new Date().toISOString()
 
  ).run()
 
  // 第二步：批量写入 spans 子表
 
  // 使用 env.DB.batch() 把多条插入合并为一次数据库操作，减少网络开销
 
  const stmt = env.DB.prepare(
 
    `INSERT INTO spans
 
      (span_id, trace_id, parent_span_id, name, start_time, duration, status, input, output, metadata)
 
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
 
  )
 
  await env.DB.batch(
 
    data.spans.map(span =>
 
      stmt.bind(
 
        span.spanId,
 
        data.traceId,
 
        span.parentSpanId,
 
        span.name,
 
        span.startTime,
 
        span.duration,
 
        span.status,
 
        JSON.stringify(span.input),
 
        JSON.stringify(span.output),
 
        JSON.stringify(span.metadata)
 
      )
 
    )
 
  )
 
}

// store-summary.ts
async function storeSummary(
 
  env: { KV: KVNamespace },
 
  trace: TraceContext
 
) {
 
  const data = trace.toJSON()
 
  // 以分钟为粒度生成 key，格式如 "metrics:2026-03-12T10:05"
 
  const minuteKey = `metrics:${new Date().toISOString().slice(0, 16)}`
 
  // 读取当前这一分钟已有的聚合数据
 
  const existing = await env.KV.get(minuteKey, 'json') as {
 
    requestCount: number
 
    errorCount: number
 
    degradedCount: number
 
    totalDuration: number
 
  } | null
 
  const current = existing ?? {
 
    requestCount: 0,
 
    errorCount: 0,
 
    degradedCount: 0,
 
    totalDuration: 0,
 
  }
 
  // 累加本次请求的统计
 
  current.requestCount += 1
 
  current.errorCount += data.spans.filter(s => s.status === 'error').length
 
  current.degradedCount += data.spans.filter(s => s.status === 'degraded').length
 
  current.totalDuration += data.totalDuration
 
  // 写回 KV，设置 24 小时过期（摘要数据不需要长期保留）
 
  await env.KV.put(minuteKey, JSON.stringify(current), {
 
    expirationTtl: 86400  // 24 小时后自动删除
 
  })
 
}

// handler.ts
// Env 类型声明了 Worker 绑定的外部资源
 
// 在 wrangler.toml 中配置 D1 和 KV 绑定后，Workers 运行时会自动注入这些对象
 
interface Env {
 
  DB: D1Database    // Cloudflare D1 关系型数据库
 
  KV: KVNamespace   // Cloudflare KV 键值存储
 
}
 
export default {
 
  // fetch 是 Workers 的入口函数，每次请求都会调用
 
  // request: 用户请求  env: 绑定的外部资源  ctx: 请求上下文（提供 waitUntil 等方法）
 
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
 
    // 1. 在请求入口创建 TraceContext
 
    const trace = new TraceContext()
 
    // 从请求中提取用户信息（实际项目中通常从 JWT token 或 session cookie 里解析）
 
    const { userId, sessionId } = await extractUserInfo(request)
 
    // 2. 执行 AI 管线（这是主路径，用户在等待这部分的结果）
 
    const response = await handleAIPipeline(request, env, trace)
 
    // 3. 采样决策
 
    const decision = shouldSample(trace, {
 
      latencyP99Threshold: 5000,  // 超过 5 秒视为异常
 
      normalSampleRate: 0.05      // 正常请求采样 5%
 
    })
 
    // 4. 把存储操作放进 waitUntil，不阻塞响应返回
 
    ctx.waitUntil(
 
      (async () => {
 
        try {
 
          if (decision.store === 'full') {
 
            // 完整写入 D1
 
            await storeFullTrace(env, trace, userId, sessionId)
 
          }
 
          // 无论是否完整存储，都更新 KV 中的聚合摘要
 
          await storeSummary(env, trace)
 
        } catch (err) {
 
          // Tracing 写入失败不应影响已返回的响应
 
          // 但需要记录下来，避免静默丢失
 
          console.error('Trace storage failed:', err)
 
        }
 
      })()
 
    )
 
    // 5. 立即返回响应，不等后台任务完成
 
    return response
 
  }
 
}