线上报警像刑侦现场：用 OpenTelemetry + AI 定位故障

💡 线上报警不是让 AI 猜谜语。先把调用链、异常和耗时采成证据，再让模型做归纳，定位结果才有机会落地。

本文要做什么

我们构造一个会稳定复现的订单接口故障：/checkout 调用库存服务时超时。应用会记录：

trace_id：同一次请求的身份证
每个 Span 的名称、耗时和状态
异常类型和错误信息
业务属性，例如订单号和商品编号

最后用 Ollama 读取 JSONL trace，生成一份带证据引用的排障报告。没有 GPU 也能运行，只是本地模型速度会慢一些。

环境准备

需要 Python 3.10+、uv 和 Ollama。Windows 可以先安装 uv：

winget install --id=astral-sh.uv -e

然后创建项目：

mkdir otel-incident-lab
cd otel-incident-lab
uv init --python 3.12
uv add fastapi uvicorn httpx opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-httpx
ollama pull qwen2.5:7b

如果暂时没有 Ollama，前面的 trace 采集和报告提示词仍然可以运行；最后一步先不调用模型。

写一个可复现的故障服务

新建 app.py：

from __future__ import annotations

import asyncio
import json
import time
from pathlib import Path
from typing import Sequence

from fastapi import FastAPI, HTTPException
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
from opentelemetry.sdk.trace.export import (
    BatchSpanProcessor,
    SpanExportResult,
    SpanExporter,
)


TRACE_FILE = Path("traces.jsonl")


class JsonlSpanExporter(SpanExporter):
    """把 Span 写成一行一个 JSON，方便 jq、脚本和模型读取。"""

    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
        with TRACE_FILE.open("a", encoding="utf-8") as file:
            for span in spans:
                context = span.get_span_context()
                record = {
                    "trace_id": format(context.trace_id, "032x"),
                    "span_id": format(context.span_id, "016x"),
                    "name": span.name,
                    "status": span.status.status_code.name,
                    "start_ns": span.start_time,
                    "end_ns": span.end_time,
                    "duration_ms": round((span.end_time - span.start_time) / 1_000_000, 2),
                    "attributes": dict(span.attributes),
                    "events": [
                        {
                            "name": event.name,
                            "attributes": dict(event.attributes or {}),
                        }
                        for event in span.events
                    ],
                }
                file.write(json.dumps(record, ensure_ascii=False) + "\n")
        return SpanExportResult.SUCCESS

    def shutdown(self) -> None:
        return None


resource = Resource.create({"service.name": "checkout-demo"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(JsonlSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("checkout-demo")

app = FastAPI(title="Checkout demo")
FastAPIInstrumentor.instrument_app(app, tracer_provider=provider)
HTTPXClientInstrumentor().instrument(tracer_provider=provider)


@app.get("/inventory/{sku}")
async def inventory(sku: str):
    if sku == "demo-fail":
        await asyncio.sleep(1.2)
        raise HTTPException(status_code=503, detail="inventory dependency timeout")
    return {"sku": sku, "available": 8}


@app.post("/checkout/{sku}")
async def checkout(sku: str):
    with tracer.start_as_current_span("checkout.reserve") as span:
        span.set_attribute("checkout.sku", sku)
        started = time.perf_counter()
        try:
            async with httpx.AsyncClient(timeout=0.5) as client:
                response = await client.get(f"http://127.0.0.1:8000/inventory/{sku}")
                response.raise_for_status()
        except Exception as exc:
            span.record_exception(exc)
            span.set_attribute("error.type", type(exc).__name__)
            span.set_attribute("error.message", str(exc))
            span.set_attribute("checkout.elapsed_ms", round((time.perf_counter() - started) * 1000, 2))
            raise HTTPException(status_code=502, detail="inventory service unavailable") from exc
        return {"ok": True, "inventory": response.json()}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

这里有一个容易漏掉的导入：checkout 使用了 httpx.AsyncClient，所以顶部必须保留 import httpx。启动服务：

uv run python app.py

另开一个终端，先请求正常路径，再制造故障：

curl.exe -X POST http://127.0.0.1:8000/checkout/demo-ok
curl.exe -X POST http://127.0.0.1:8000/checkout/demo-fail

traces.jsonl 会出现多行 Span。由于 BatchSpanProcessor 是异步批量导出，停止服务前等几秒，或再次请求一次，确保最后一批数据写入文件。

先不用 AI，人工检查证据

PowerShell 可以直接查看失败 Span：

Get-Content traces.jsonl | ConvertFrom-Json |
  Where-Object { $_.status -ne "UNSET" -and $_.status -ne "OK" } |
  Select-Object trace_id,name,status,duration_ms,attributes

也可以用 Python 按 trace 聚合：

import json
from collections import defaultdict

traces = defaultdict(list)
with open("traces.jsonl", encoding="utf-8") as file:
    for line in file:
        item = json.loads(line)
        traces[item["trace_id"]].append(item)

for trace_id, spans in traces.items():
    print(f"\nTRACE {trace_id}")
    for span in sorted(spans, key=lambda value: value["start_ns"]):
        print(span["name"], span["status"], span["duration_ms"], span["attributes"])

在真实系统里，先人工确认“哪条证据支持哪个结论”，比直接把全部日志扔给模型更重要。

让本地模型生成排障报告

新建 triage.py。它只把最近一个 trace 的结构化数据交给 Ollama，不上传整台机器的日志：

from __future__ import annotations

import json
import os
import urllib.request
from collections import defaultdict


def load_latest_trace(path: str = "traces.jsonl") -> list[dict]:
    groups: dict[str, list[dict]] = defaultdict(list)
    with open(path, encoding="utf-8") as file:
        for line in file:
            item = json.loads(line)
            groups[item["trace_id"]].append(item)
    if not groups:
        raise RuntimeError("没有找到 trace，请先调用接口")
    return max(groups.values(), key=lambda items: max(item["start_ns"] for item in items))


def ask_ollama(spans: list[dict]) -> str:
    evidence = json.dumps(spans, ensure_ascii=False, indent=2)
    prompt = f"""
你是值班 SRE。只根据下面的 OpenTelemetry trace 证据排查问题，不要编造不存在的日志。

请输出：
1. 结论（一个句子）
2. 证据（引用 span 名称、状态和耗时）
3. 最可能的根因（说明置信度：高/中/低）
4. 低风险验证步骤（不要直接修改生产数据）
5. 如果证据不足，明确列出还需要什么数据

TRACE 证据：
{evidence}
"""
    body = json.dumps(
        {
            "model": os.getenv("OLLAMA_MODEL", "qwen2.5:7b"),
            "messages": [{"role": "user", "content": prompt}],
            "stream": False,
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        "http://127.0.0.1:11434/api/chat",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=120) as response:
        result = json.loads(response.read().decode("utf-8"))
    return result["message"]["content"]


if __name__ == "__main__":
    print(ask_ollama(load_latest_trace()))

运行：

uv run python triage.py

预期报告应该指出：inventory/demo-fail 返回 503，checkout.reserve 因 0.5 秒客户端超时转成 502。模型如果只说“网络可能有问题”，就说明证据或 Prompt 还不够具体，不能把模糊答案当成修复结论。

生产落地时怎么接 Collector

本文用 JSONL 文件是为了让每一步都能在单机复现。生产环境一般改成：

应用 SDK → OTLP → OpenTelemetry Collector → Jaeger/Tempo + 日志平台
                                      ↘ 诊断服务 → LLM

Collector 负责批量、重试、采样和导出；模型诊断服务只读取被授权的 trace。不要让模型直接访问 Collector 管理端口，也不要把完整请求体、Cookie、Authorization 写进 Span 属性。

让结果可信的四条规则

先证据后结论：Prompt 要求引用 Span 名称、状态和耗时。
先只读后修复：模型先生成验证步骤，修复由人或受控流水线执行。
控制数据范围：只传当前 trace 和必要属性，先做脱敏。
保留原始链路：AI 报告是辅助视图，不能替代原始 trace 和告警记录。

常见坑

只采日志，不采 trace_id

没有关联 ID，模型只能猜哪条日志属于哪次请求。入口、下游调用和异常日志必须共享 trace 上下文。

把高基数字段塞进 attributes

用户输入、完整 URL、订单内容会造成指标和存储爆炸。属性只保留排障必需的枚举、短 ID 和状态。

让模型直接执行修复

“重启服务”“删除缓存”“修改参数”都属于有副作用的动作。报告里只生成命令草稿，执行必须经过确认、审批和审计。

总结

AI 做故障定位的正确姿势不是“把日志丢给聊天框”，而是：

规范化采集 → trace 聚合 → 脱敏裁剪 → AI 分析 → 人工验证 → 受控修复

先把证据链做扎实，模型才有机会从“像是网络问题”进步到“哪个 Span、哪一个状态、哪一个超时导致了失败”。

参考资料

作者：牛马便利店一号店员

文档信息

本文作者：牛马
本文链接：https://geekhappy.com/2026/08/25/otel-ai-incident-triage/
版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）

牛马假日