AI 驱动的分布式链路追踪异常检测：从海量 Span 到根因定位

Dicky-_-zhang

909人浏览 · 2026-06-11 10:54:51

Dicky-_-zhang · 2026-06-11 10:54:51 发布

AI 驱动的分布式链路追踪异常检测：从海量 Span 到根因定位

cover

一、链路追踪的"数据过载"：看得见链路，看不见问题

分布式链路追踪（如 Jaeger、Zipkin）是微服务可观测性的核心工具，能记录请求在服务间的完整调用路径。但在大规模系统中，链路追踪面临"数据过载"问题：一个日均千万请求的系统，每天产生数十亿个 Span，人工翻看链路数据无异于大海捞针。

更棘手的是异常检测的滞后性：通常是在用户投诉或告警触发后，才去链路追踪系统中查找对应的 Trace，这是"事后排查"模式。AI 驱动的链路异常检测可以实现"事前发现"——自动从海量 Span 中识别异常模式，定位根因服务，甚至预测即将发生的性能退化。

二、链路异常检测的算法架构

链路异常检测的核心思路是：将 Trace 数据建模为有向无环图（DAG），通过图结构分析和时序统计，识别异常 Span 和根因服务。

flowchart TD
    A[Trace 数据流] --> B[Span 图构建]
    B --> C[服务依赖图]
    B --> D[延迟分布统计]

    C --> E[结构异常检测]
    D --> F[延迟异常检测]

    E --> G[异常 Span 定位]
    F --> G

    G --> H[根因推理]
    H --> I[服务级根因]
    H --> J[操作级根因]

    I --> K[AI 生成诊断报告]
    J --> K

延迟异常检测基于每个服务的响应时间分布。正常情况下，服务 A 的 P99 延迟为 200ms，如果突然升高到 2s，就是异常。但更隐蔽的异常是"渐进式退化"——P99 每天增加 10ms，一个月后翻倍，人工很难察觉。统计过程控制（SPC）方法可以检测这种渐进式偏移。

三、工程化实现

3.1 Trace 数据采集与图构建

// TraceGraphBuilder.java
@Component
public class TraceGraphBuilder {

    // 从 Span 列表构建调用图
    public TraceGraph buildGraph(List<Span> spans) {
        Map<String, SpanNode> nodeMap = new HashMap<>();
        List<SpanEdge> edges = new ArrayList<>();

        for (Span span : spans) {
            SpanNode node = SpanNode.builder()
                .spanId(span.getSpanId())
                .service(span.getServiceName())
                .operation(span.getOperationName())
                .durationMs(span.getDurationMs())
                .status(span.getStatus())
                .startTimeMs(span.getStartTimeMs())
                .build();
            nodeMap.put(span.getSpanId(), node);
        }

        // 构建边：parent → child
        for (Span span : spans) {
            if (span.getParentSpanId() != null
                && nodeMap.containsKey(span.getParentSpanId())) {
                edges.add(SpanEdge.builder()
                    .source(span.getParentSpanId())
                    .target(span.getSpanId())
                    .build());
            }
        }

        return new TraceGraph(nodeMap, edges);
    }

    // 从多条 Trace 构建服务依赖图
    public ServiceDependencyGraph buildServiceGraph(
        List<TraceGraph> traces
    ) {
        Map<String, ServiceNode> services = new HashMap<>();
        List<ServiceEdge> serviceEdges = new ArrayList<>();

        for (TraceGraph trace : traces) {
            for (SpanEdge edge : trace.getEdges()) {
                SpanNode source = trace.getNode(edge.getSource());
                SpanNode target = trace.getNode(edge.getTarget());
                if (source == null || target == null) continue;

                String edgeKey = source.getService()
                    + "->" + target.getService();
                ServiceEdge se = serviceEdges.stream()
                    .filter(e -> e.getKey().equals(edgeKey))
                    .findFirst()
                    .orElseGet(() -> {
                        ServiceEdge newEdge = ServiceEdge.builder()
                            .source(source.getService())
                            .target(target.getService())
                            .key(edgeKey)
                            .callCount(0)
                            .errorCount(0)
                            .totalDurationMs(0L)
                            .build();
                        serviceEdges.add(newEdge);
                        return newEdge;
                    });

                se.setCallCount(se.getCallCount() + 1);
                se.setTotalDurationMs(
                    se.getTotalDurationMs() + target.getDurationMs()
                );
                if ("ERROR".equals(target.getStatus())) {
                    se.setErrorCount(se.getErrorCount() + 1);
                }
            }
        }

        return new ServiceDependencyGraph(services, serviceEdges);
    }
}

3.2 延迟异常检测

// LatencyAnomalyDetector.java
@Component
public class LatencyAnomalyDetector {

    // 基于统计过程控制的异常检测
    public List<Anomaly> detectLatencyAnomalies(
        ServiceDependencyGraph graph,
        Map<String, ServiceBaseline> baselines
    ) {
        List<Anomaly> anomalies = new ArrayList<>();

        for (ServiceEdge edge : graph.getEdges()) {
            ServiceBaseline baseline = baselines.get(edge.getKey());
            if (baseline == null) continue;

            double avgDuration = (double) edge.getTotalDurationMs()
                / edge.getCallCount();

            // 使用 3-sigma 规则：超过均值 + 3 倍标准差视为异常
            double threshold = baseline.getMeanDurationMs()
                + 3 * baseline.getStdDevDurationMs();

            if (avgDuration > threshold) {
                double severity = (avgDuration - baseline.getMeanDurationMs())
                    / baseline.getStdDevDurationMs();

                anomalies.add(Anomaly.builder()
                    .type("LATENCY_ANOMALY")
                    .serviceEdge(edge.getKey())
                    .currentValue(avgDuration)
                    .baselineValue(baseline.getMeanDurationMs())
                    .severity(severity)
                    .description(String.format(
                        "%s 平均延迟 %.0fms，基线 %.0fms，"
                        + "偏离 %.1f 个标准差",
                        edge.getKey(), avgDuration,
                        baseline.getMeanDurationMs(), severity
                    ))
                    .build());
            }
        }

        return anomalies;
    }

    // 渐进式退化检测：线性回归斜率检验
    public List<Anomaly> detectGradualDegradation(
        Map<String, List<DataPoint>> timeSeries,
        double slopeThreshold
    ) {
        List<Anomaly> anomalies = new ArrayList<>();

        for (var entry : timeSeries.entrySet()) {
            String service = entry.getKey();
            List<DataPoint> points = entry.getValue();

            if (points.size() < 30) continue; // 至少 30 个数据点

            // 简单线性回归
            double[] regression = linearRegression(points);
            double slope = regression[0]; // 斜率
            double intercept = regression[1];

            // 斜率显著大于 0 表示渐进式退化
            if (slope > slopeThreshold) {
                anomalies.add(Anomaly.builder()
                    .type("GRADUAL_DEGRADATION")
                    .serviceEdge(service)
                    .severity(slope / slopeThreshold)
                    .description(String.format(
                        "%s 延迟呈上升趋势，"
                        + "每分钟增加 %.2fms",
                        service, slope
                    ))
                    .build());
            }
        }

        return anomalies;
    }

    private double[] linearRegression(List<DataPoint> points) {
        double sumX = 0, sumY = 0, sumXY = 0, sumX2 = 0;
        int n = points.size();

        for (int i = 0; i < n; i++) {
            sumX += i;
            sumY += points.get(i).getValue();
            sumXY += i * points.get(i).getValue();
            sumX2 += (double) i * i;
        }

        double slope = (n * sumXY - sumX * sumY)
            / (n * sumX2 - sumX * sumX);
        double intercept = (sumY - slope * sumX) / n;

        return new double[]{slope, intercept};
    }
}

3.3 AI 根因推理

// RootCauseAnalyzer.java
@Component
@RequiredArgsConstructor
public class RootCauseAnalyzer {

    public RootCauseReport analyze(
        List<Anomaly> anomalies,
        ServiceDependencyGraph graph
    ) {
        // 按严重程度排序
        anomalies.sort(Comparator.comparingDouble(
            Anomaly::getSeverity).reversed()
        );

        // 识别根因服务：异常的"源头"
        // 如果服务 A → B → C 都异常，A 是根因
        List<String> rootCauseServices = findRootCauseServices(
            anomalies, graph
        );

        return RootCauseReport.builder()
            .anomalies(anomalies)
            .rootCauseServices(rootCauseServices)
            .recommendation(generateRecommendation(
                anomalies, rootCauseServices
            ))
            .build();
    }

    private List<String> findRootCauseServices(
        List<Anomaly> anomalies,
        ServiceDependencyGraph graph
    ) {
        Set<String> anomalousServices = anomalies.stream()
            .map(a -> a.getServiceEdge().split("->")[0])
            .collect(Collectors.toSet());

        // 找到没有上游异常的服务，即为根因
        return anomalousServices.stream()
            .filter(service -> {
                List<String> upstreams = graph.getUpstreamServices(service);
                return upstreams.stream()
                    .noneMatch(anomalousServices::contains);
            })
            .collect(Collectors.toList());
    }
}