背景

隨著互聯網架構的流行，越來越多的系統開始走向分布式化、微服務化。如何快速發現和定位分布式系統下的各類性能瓶頸成為了擺在開發者面前的難題。借助分布式追蹤系統的調用鏈路還原能力，開發者可以完整地了解一次請求的執行過程和詳細信息。但要真正分析出系統的性能瓶頸往往還需要鏈路拓撲、應用依賴分析等工具的支持。這些工具使用起來雖然簡單，但其背后的原理是什么？本文將帶您一起探索。

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 每次運行都會重新計算指定日期的服務間拓撲關系，具體流程如下：

根據傳入的日期定位存放 span 數據的 index，例如傳入的日期是2019-11-11，則目標 span 為jaeger-span-2019-11-11。如果沒有指定日期，便會使用當天的 index，此時 index 里存放著當天自午夜以來的所有 span 數據。

將目標 index 中的 span 數據按 traceID 進行分組，得到Map(traceID: Set(span))。

針對單個 trace，遍歷該 trace 的所有 span，計算出對應的拓撲關系List<dependency>。

將Map(traceID: List<dependency>)按 dependency 重新分組，并累加相同 dependency 之間的調用次數。

將計算結果寫入 Elasticsearch 用于存放拓撲關系的 index 里。例如jaeger-span-2019-11-11對應的拓撲關系 index 為jaeger-dependencies-2019-11-11。

拓撲關系查詢

查詢時會根據傳入的 lookback 查詢對應時間段的 dependency 數據，默認為過去 24 小時。查詢過程很簡單：

找出和指定時間范圍有交集的所有 dependency index。

從這些 index 中過濾出符合要求的所有 dependency。

將 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);

filter((FilterFunction<Span>) span -> span.isClient() || span.isServer())只保留有 client 或 server 標簽的 span。

keyBy((KeySelector<Span, String>) span -> String.format("%d:%d", span.getTraceIdHigh(), span.getTraceIdLow()))?- 基于 traceIdHigh 和 traceIdLow 的組合將 span 數據分組。

window(EventTimeSessionWindows.withGap(Time.minutes(3)))?- 為每個 trace 創建一個 gap 為 3 分鐘的會話窗口。表示對于某個 trace，如果 3 分鐘內沒有新的 span 數據到達，就認為窗口結束，進而觸發后續的聚合操作。

SpanToTraceWindowFunction?- 負責將會話窗口里的 span 數據收集到一起。

AdjusterFunction負責去除一個 trace 里的重復 span。

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);

flatMap(new TraceToDependencies())?- 遍歷一個 trace 的所有 span 數據，收集該 trace 表征的拓撲關系。

keyBy(key -> key.getParent() + key.getChild())?- 將具有相同父子關系的 dependency 分到一組。

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 方案。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

總結

以上是生活随笔為你收集整理的基于 Tracing 数据的拓扑关系生成原理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。