背景

随着互联网架构的流行，越来越多的系统开始走向分布式化、微服务化。如何快速发现和定位分布式系统下的各类性能瓶颈成为了摆在开发者面前的难题。借助分布式追踪系统的调用链路还原能力，开发者可以完整地了解一次请求的执行过程和详细信息。但要真正分析出系统的性能瓶颈往往还需要链路拓扑、应用依赖分析等工具的支持。这些工具使用起来虽然简单，但其背后的原理是什么？本文将带您一起探索。

Jaeger 作为从 CNCF 毕业的第七个项目，已经成为了云原生架构下分布式追踪系统的第一选择。本文将以 Jaeger 为例，介绍基于 Tracing 数据的拓扑关系生成原理，文中使用的版本为1.14。

Jaeger 架构

笔者曾在 2018 年初基于 Jaeger 1.2.0 做过一些开发，参见《开放分布式追踪（OpenTracing）入门与 Jaeger 实现》。经过十多个版本的发展，Jaeger 的架构发生了一些变化，目前在大规模生产环境中推荐下面 2 种部署模式。

Direct to storage

Collector 将采集到的 trace 数据直接写入 DB，Spark jobs 定期读取这些 trace 数据并将计算出的拓扑关系再次写入 DB 中。

Kafka as intermediate buffer

Collector 将采集到的 trace 数据写入中间缓冲区 Kafka 中，Ingerster 读取 Kafka 中的数据并持久化到 DB 里。同时，Flink jobs 持续读取 Kafka 中的数据并将计算出的拓扑关系写入 DB 中。

Jaeger 组件

一个完整的 Jaeger 系统由以下几部分组成。

• Jaeger client libraries - 为不同语言实现了符合 OpenTracing 标准的 SDK。应用程序通过 API 写入数据，client library 把 trace 信息按照应用程序指定的采样策略传递给 jaeger-agent。
• Agent - 它是一个监听在 UDP 端口上接收 span 数据的网络守护进程，它会将数据批量发送给 collector。它被设计成一个基础组件，部署到宿主机或容器里。Agent 将 client library 和 collector 解耦，为 client library 屏蔽了路由和发现 collector 的细节。
• Collector - 接收 jaeger-agent 发送过来的 trace 数据，然后在处理管道中对它们进行验证、索引、转换并最终完成持久化存储。Jaeger 的存储组件被设计成可插拔的，目前官方支持 Cassandra、Elasticsearch 和 Kafka。
• Query - 接收查询请求，然后从后端存储组件中检索 trace 并通过 UI 进行展示。
• Ingester - 负责从 Kafka 中读取数据然后写入另一个存储后端。

拓扑关系生成

下图是 Jaeger 官方提供的微服务应用 Hot R.O.D. 的服务间拓扑关系。通过此图，开发者可以清楚地了解过去一段时间里服务间的调用关系和调用次数。

由于生产环境中的 trace 数据量巨大，每次查询时通过扫描数据库中的全量数据来构建拓扑关系不切实际。因此，Jaeger 提供了基于 Spark jobs 和 Flink jobs 两种从 trace 数据中提取拓扑关系的方法。

Jaeger Spark dependencies

Jaeger Spark dependencies 是一个 Spark 任务，它从特定的后端存储中读取 span 数据，计算服务间的拓扑关系，并将结果存储起来供 UI 展示。目前支持的后端存储类型有 Cassandra 和 Elasticsearch。 由于 Cassadra 和 Elasticsearch 计算拓扑关系的逻辑大同小异，下面将以 Elasticsearch 为例进行分析。

Span 数据组织结构

Jaeger 会根据 span 的 StartTime 字段将它们写到 Elasticsearch 以天为单位的 index 里，存放 span 的 index 组织结构如下：

jaeger-span-2019-11-11
jaeger-span-2019-11-12
jaeger-span-2019-11-13
...

拓扑关系计算流程

Spark job 每次运行都会重新计算指定日期的服务间拓扑关系，具体流程如下：

1. 根据传入的日期定位存放 span 数据的 index，例如传入的日期是2019-11-11，则目标 span 为jaeger-span-2019-11-11。如果没有指定日期，便会使用当天的 index，此时 index 里存放着当天自午夜以来的所有 span 数据。
2. 将目标 index 中的 span 数据按 traceID 进行分组，得到Map(traceID: Set(span))。
3. 针对单个 trace，遍历该 trace 的所有 span，计算出对应的拓扑关系List<dependency>。
4. 将Map(traceID: List<dependency>)按 dependency 重新分组，并累加相同 dependency 之间的调用次数。
5. 将计算结果写入 Elasticsearch 用于存放拓扑关系的 index 里。例如jaeger-span-2019-11-11对应的拓扑关系 index 为jaeger-dependencies-2019-11-11。

拓扑关系查询

查询时会根据传入的 lookback 查询对应时间段的 dependency 数据，默认为过去 24 小时。查询过程很简单：

1. 找出和指定时间范围有交集的所有 dependency index。
2. 从这些 index 中过滤出符合要求的所有 dependency。
3. 将 dependency 在 UI 层进行聚合展示。

Jaeger Analytics

Jaeger Analytics 是一个 Flink 任务，它从 Kafka 中消费 span 数据，实时计算服务间的拓扑关系，最后将计算结果写入 Cassadra 中。

拓扑关系计算流程

将 Kafka 设为 source

这里将 Kafka 设置为 Flink 任务的 source，此时 span 数据将不断地从 Kafka 流向 Flink 任务。

将离散的 span 聚合成 trace

DataStream<Iterable<Span>> traces = spans
    .filter((FilterFunction<Span>) span -> span.isClient() || span.isServer())
    .name(FILTER_LOCAL_SPANS)
    .keyBy((KeySelector<Span, String>) span -> String
        .format("%d:%d", span.getTraceIdHigh(), span.getTraceIdLow()))
    .window(EventTimeSessionWindows.withGap(Time.minutes(3)))
    .apply(new SpanToTraceWindowFunction()).name(SPANS_TO_TRACES)
    .map(new AdjusterFunction<>()).name(DEDUPE_SPAN_IDS)
    .map(new CountSpansAndLogLargeTraceIdFunction()).name(COUNT_SPANS);

1. filter((FilterFunction<Span>) span -> span.isClient() || span.isServer())只保留有 client 或 server 标签的 span。
2. keyBy((KeySelector<Span, String>) span -> String.format("%d:%d", span.getTraceIdHigh(), span.getTraceIdLow())) - 基于 traceIdHigh 和 traceIdLow 的组合将 span 数据分组。
3. window(EventTimeSessionWindows.withGap(Time.minutes(3))) - 为每个 trace 创建一个 gap 为 3 分钟的会话窗口。表示对于某个 trace，如果 3 分钟内没有新的 span 数据到达，就认为窗口结束，进而触发后续的聚合操作。
4. SpanToTraceWindowFunction - 负责将会话窗口里的 span 数据收集到一起。
5. AdjusterFunction负责去除一个 trace 里的重复 span。
6. CountSpansAndLogLargeTraceIdFunction以 10 秒为单位，统计 span 数量随时间的分布。

计算 dependencies

DataStream<Dependency> dependencies = traces
    .flatMap(new TraceToDependencies()).name(TRACE_TO_DEPENDENCIES)
    .keyBy(key -> key.getParent() + key.getChild())
    .timeWindow(Time.minutes(30))
    .sum("callCount").name(PREAGGREGATE_DEPENDENCIES);

1. flatMap(new TraceToDependencies()) - 遍历一个 trace 的所有 span 数据，收集该 trace 表征的拓扑关系。
2. keyBy(key -> key.getParent() + key.getChild()) - 将具有相同父子关系的 dependency 分到一组。
3. timeWindow(Time.minutes(30)).sum("callCount") - 为每对依赖关系创建一个 30 分钟的时间窗口，窗口结束触发计数操作。

将 Cassandra 设为 sink

这里将 Cassandra 设置为 Flink 任务的 sink，当依赖关系因满足时间窗口的触发条件被计算完毕后，将以dependencies(ts, ts_index, dependencies)的形式持久化到 Cassandra 中。

拓扑关系查询

根据指定的时间范围过滤出所有符合要求的 dependency，然后在 UI 层进行聚合展示。从 Cassandra 查询 dependency 使用的 CQL 如下。

SELECT ts, dependencies FROM dependencies WHERE ts_index >= startTs AND ts_index < endTs

总结

Spark jobs、Flink jobs 两种计算拓扑关系的方案虽然在细节上有所不同，但整体流程非常相似，可总结成下图。

对于 Jaeger Spark dependencies， 拓扑关系的精确程度和 Spark job 的执行频率密切相关。执行频率越高，查询结果越精确，但消耗的计算资源也会越多。举个例子，如果 Spark job 每小时运行一次，拓扑关系可能无法反映最近一小时服务间的调用情况。

对于 Jaeger Analytics，它以 Kafka 作为缓存，增量地处理到达的 span 数据，具有更好的实时性。如果对最近时间拓扑关系的精确程度有比较高的要求，建议选用 Jaeger Analytics 方案。

(en)