分布式追蹤 SkyWalking 源碼分析四 Agent 收集 && 發送 trace資料

分布式鍊路追蹤系統，鍊路的追蹤大體流程如下：

1. Agent 收集 Trace 資料。
2. Agent 發送 Trace 資料給 Collector 。
3. Collector 接收 Trace 資料。
4. Collector 存儲 Trace 資料到存儲器，例如，資料庫。

org.skywalking.apm.agent.core.context.trace.TraceSegment

 ，是一次分布式鍊路追蹤( Distributed Trace ) 的一段。

• 一條 TraceSegment ，用于記錄所線上程( Thread )的鍊路。
• 一次分布式鍊路追蹤，可以包含多條 TraceSegment ，因為存在跨程序( 例如，RPC 、MQ 等等)，或者垮線程( 例如，并發執行、異步回調等等 )。

traceSegmentId

 屬性，TraceSegment 的編号，全局唯一

spans

 屬性，包含的 Span 數組。這是 TraceSegment 的主體，總的來說，TraceSegment 是 Span 數組的封裝。

我們先來看看一個爸爸的情況，常見于 RPC 調用。例如，【服務 A】調用【服務 B】時，【服務 B】建立一個 TraceSegment 對象：

• 将自己的 

  refs

   指向【服務 A】的 TraceSegment 。
• 将自己的 

  relatedGlobalTraces

   設定為 【服務 A】的 DistributedTraceId 對象。

2.1 ID

org.skywalking.apm.agent.core.context.ids.ID

 ，編号。從類的定義上，這是一個通用的編号，由三段整數組成。

目前使用 GlobalIdGenerator 生成，作為全局唯一編号。屬性如下：

• part1

   屬性，應用執行個體編号。

• part2

   屬性，線程編号。

• part3

   屬性，時間戳串，生成方式為 

  ${時間戳} * 10000 + 線程自增序列([0, 9999])

   。例如：15127007074950012 。具體生成方法的代碼，在 GlobalIdGenerator 中詳細解析。

• encoding

   屬性，編碼後的字元串。格式為 

  "${part1}.${part2}.${part3}"

   。例如，

  "12.35.15127007074950000"

   。
  • 使用 

    #encode()

     方法，編碼編号。

• isValid

   屬性，編号是否合法。
  • 使用 

    ID(encodingString)

     構造方法，解析字元串，生成 ID 。

      org.skywalking.apm.agent.core.context.ids.NewDistributedTraceId ，建立的分布式鍊路追蹤編号。當全局鍊路追蹤開始，建立 TraceSegment 對象的過程中，會調用 

DistributedTraceId()

 構造方法，建立 DistributedTraceId 對象。該構造方法内部會調用 

GlobalIdGenerator#generate()

 方法，建立 ID 對象。

  #setOperationId(operationId)

 方法，設定操作編号。考慮到操作名是字元串，Agent 發送給 Collector 占用流量較大。是以，Agent 會将操作注冊到 Collector ，生成操作編号。

2.2.1 Tag

2.2.1.1 AbstractTag

org.skywalking.apm.agent.core.context.tag.AbstractTag<T>

 ，标簽抽象類。注意，這個類的用途是将标簽屬性設定到 Span 上，或者說，它是設定 Span 的标簽的工具類。代碼如下：

• key

   屬性，标簽的鍵。

• #set(AbstractSpan span, T tagValue)

   抽象方法，設定 Span 的标簽鍵 

  key

   的值為 

  tagValue

關于span的類繼承圖

Span 隻有三種實作類：

• EntrySpan ：入口 Span
• LocalSpan ：本地 Span
• ExitSpan ：出口 Span

2.2.2.2.1 EntrySpan

org.skywalking.apm.agent.core.context.trace.EntrySpan

 ，實作 StackBasedTracingSpan 抽象類，入口 Span ，用于服務提供者( Service Provider ) ，例如 Tomcat 。

那麼為什麼 EntrySpan 繼承 StackBasedTracingSpan ？

例如，我們常用的 SprintBoot 場景下，Agent 會在 SkyWalking 插件在 Tomcat 定義的方法切面，建立 EntrySpan 對象，也會在 SkyWalking 插件在 SpringMVC 定義的方法切面，建立 EntrySpan 對象。那豈不是出現兩個 EntrySpan ，一個 TraceSegment 出現了兩個入口 Span ？

答案是當然不會！Agent 隻會在第一個方法切面，生成 EntrySpan 對象，第二個方法切面，棧深度 + 1。這也是上面我們看到的 

#finish(TraceSegment)

 方法，隻在棧深度為零時，出棧成功。通過這樣的方式，保持一個 TraceSegment 有且僅有一個 EntrySpan 對象。

對新進入的方法切面，就把棧深度+1

而對于StackBasedTracingSpan的finish方法，把棧深度減少

2.2.2.2.2 ExitSpan

org.skywalking.apm.agent.core.context.trace.ExitSpan

 ，繼承 StackBasedTracingSpan 抽象類，出口 Span ，用于服務消費者( Service Consumer )，例如 HttpClient 、MongoDBClient 。

一個 TraceSegment 可以有多個 ExitSpan，例如，Dubbo A 服務在處理一個請求時，會調用 Dubbo B 服務得到相應之後，緊接着調用了 Dubbo C 服務，這樣，該 TraceSegment 就有了兩個完全獨立的 ExitSpan。

那麼為什麼 ExitSpan 繼承 StackBasedTracingSpan ？

例如，我們可能在使用的 Dubbox 場景下，【Dubbox 服務 A】使用 HTTP 調用【Dubbox 服務 B】時，實際過程是，【Dubbox 服務 A】=》【HttpClient】=》【Dubbox 服務 B】。Agent 會在【Dubbox 服務 A】建立 ExitSpan 對象，也會在 【HttpClient】建立 ExitSpan 對象。那豈不是一次出口，出現兩個 ExitSpan ？

答案是當然不會！Agent 隻會在【Dubbox 服務 A】，生成 EntrySpan 對象，第二個方法切面，棧深度 + 1。這也是上面我們看到的 

#finish(TraceSegment)

 方法，隻在棧深度為零時，出棧成功。通過這樣的方式，保持一次出口有且僅有一個 ExitSpan 對象。

當然，一個 TraceSegment 會有多個 ExitSpan 對象 ，例如【服務 A】遠端調用【服務 B】，然後【服務 A】再次遠端調用【服務 B】，或者然後【服務 A】遠端調用【服務 C】。

2.3 TraceSegmentRef

org.skywalking.apm.agent.core.context.trace.TraceSegmentRef

 ，TraceSegment 指向，通過 

traceSegmentId

 和 

spanId

 屬性，指向父級 TraceSegment 的指定 Span 。

3. Context

在 「2. Trace」 中，我們看了 Trace 的資料結構，本小節，我們一起來看看 Context 是怎麼收集 Trace 資料的。

3.1 ContextManager

org.skywalking.apm.agent.core.context.ContextManager

 ，實作了 BootService 、TracingContextListener 、IgnoreTracerContextListener 接口，鍊路追蹤上下文管理器。

CONTEXT

 靜态屬性，線程變量，存儲 AbstractTracerContext 對象。為什麼是線程變量呢？

一個 TraceSegment 對象，關聯到一個線程，負責收集該線程的鍊路追蹤資料，是以使用線程變量。

而一個 AbstractTracerContext 會關聯一個 TraceSegment 對象，ContextManager 負責擷取、建立、銷毀 AbstractTracerContext 對象。

#getOrCreate(operationName, forceSampling)

 靜态方法，擷取 AbstractTracerContext 對象。若不存在，進行建立。

• 要需要收集 Trace 資料的情況下，建立 TracingContext 對象。
• 不需要收集 Trace 資料的情況下，建立 IgnoredTracerContext 對象。

在下面的 

#createEntrySpan(...)

 、

#createLocalSpan(...)

 、

#createExitSpan(...)

 等等方法中，都會調用 AbstractTracerContext 提供的方法。這些方法的代碼，我們放在 「3.2 AbstractTracerContext」 一起解析，保證流程的整體性。

另外，ContextManager 封裝了所有 AbstractTracerContext 提供的方法，進而實作，外部調用者，例如 SkyWalking 的插件，隻調用 ContextManager 的方法，而不調用 AbstractTracerContext 的方法。

建立出traceContext

+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

核心類實作TracingContext

建立EntrySpan

父span存在，就直接start；父span不存在，就建立一個EntrySpan

建立exitSpan，原理類似

++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

結束span

調用pop彈棧，然後調用finish，結束本線程的traceSegment

3.2.3.3 傳輸

org.skywalking.apm.agent.core.context.CarrierItem

 ，傳輸載體項。代碼如：

• headKey

   屬性，Header 鍵。

• headValue

   屬性，Header 值。

• next

   屬性，下一個項。

CarrierItem 有兩個子類：

• CarrierItemHead ：Carrier 項的頭( Head )，即首個元素。
• SW3CarrierItem ：

  header = sw3

   ，用于傳輸 ContextCarrier 。

NetworkAddressDictionary 與服務端定期同步的方法是syncRemoteDictionary()方法，具體實作如下：

在前面介紹 skywalking-agent 初始化的時候，Agent 會定期向服務端發送心跳，在心跳發送完之後，就會調用 EndpointNameDictionary、NetworkAddressDictionary 的 syncRemoteDictionary() 方法進行同步，看看代碼 ServiceAndEndpointRegisterClient 的153行和154行就知道了

Tomcat 插件

先看下tomcat工作流程

接着看 TomcatInstrumentation.getInstanceMethodsInterceptPoints()方法，它傳回了兩個 InstanceMethodsInterceptPoint 對象，一個攔截 invoke()方法，一個攔截 throwable()方法。

先來看攔截 invoke()方法的 TomcatInvokeInterceptor，它的 beforeMethod() 方法核心就是建立 EntrySpan

接下來看，ContextManager.createEntry() 方法的實作，前面說過其核心是調用 getOrCreate() 方法擷取/建立目前 TracingContext 對象，然後調用 TracingContext.createEntry() 方法建立（或是重新 start ）目前 EntrySpan 對象，這裡更詳細的說一下一些實作細節吧。

ContextManager.createEntry() 方法首先會檢測目前 ContextCarrier 是否合法，其實就是檢查 ContextCarrier 的8個核心字段是否填充好了，如果合法，就證明是上遊有 Trace 資訊傳遞下來了

    TracingContext.stopSpan() 方法的具體在前面已經詳細分析過了，其中會調用 StackBasedTracingSpan.finish() 方法嘗試關閉目前 Span，這裡會檢測該 Span 的 operationId 字段，如果為空，則嘗試再次通過 DictionaryManager元件用 operationName 換取 operationId

    這裡的核心有兩步，上面明顯能看出來的是将 Filter 執行個體串成 Chain，另一個核心步驟就是通過 ExtensionLoader 加載 Filter 對象，原理是SPI，但是 Dubbo 的 SPI 實作有點優化，但是原理和思想基本一樣

    apm-dubbo-2.7.x-plugin 插件的 skywalking-plugin.def 中定義的插件是 DubboInstrumentation，它攔截的是 MonitorFilter.invoke()方法，具體的 Interceptor 實作是 DubboInterceptor，在其 beforeMethod() 方法中會根據目前 MonitorFilter 所在的服務角色（Consumer/Provider）建立對應的 Span（ExitSpan/EntrySpan）

在 ContextManager.createExitSpan()方法中除了建立 ExitSpan 之外，還會調用 inject() 方法将 Trace 資訊記錄到 CarrierContext 中，這樣後面通過CarrierItem 持久化的時候才有值。

DubboInterceptor.afterMethod() 方法實作比較簡單，有異常就是通過 log 方式記錄到目前 Span，最後嘗試關閉目前 Span。

+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

Agent發送trace資料

Agent 收集到 Trace 資料後，不是寫入外部消息隊列( 例如，Kafka )或者日志檔案，而是 Agent 寫入記憶體消息隊列，背景線程【異步】發送給 Collector 。

核心類為TraceSegmentServiceClient，負責将 TraceSegment 異步發送到 Collector

核心方法consume

1.判斷狀态是Connected

2.開啟一個觀察器 upstreamSegmentStreamObserver

3.循環data，然後轉換并且把這個upstreamSegment，發送到collector