链路追踪

文章40 |   阅读 20534 |   点赞0

来源:https://blog.csdn.net/weixin_42073629/category_9940428.html

SkyWalking 源码分析 —— 应用于应用实例的注册

x33g5p2x  于2021-12-21 转载在 其他  
字(7.6k)|赞(0)|评价(0)|浏览(424)

1. 概述

本文主要分享 应用与应用实例的注册。先来简单了解下注册的整体流程:

  • 应用启动,Agent 向 Collector 注册应用
  • 注册应用成功后,Agent 向 Collector 注册应用实例

下面,我们分成两个小节,分别从 API 的实现调用,分享代码的具体实现。
友情提示:推荐阅读 《探针与Collector间通讯协议》 。

2. Collector 注册相关 API

Collector 注册相关 API 相关有四个接口:

  • 2.1 应用的注册 API
  • 2.2 应用实例的正常注册 API
  • 2.3 应用实例的恢复注册 API
  • 2.4 应用实例的心跳 API

API 处理的流程大体如下:

  • 绿框部分,【2.3】【2.4】两个 API ,直接 Service 调用 DAO 方法,无需经过 Graph / Stream 相关方法。

2.1 应用的注册 API

我们先来看看 API 的定义,ApplicationRegisterService.proto ,如下图所示:

2.1.1 ApplicationRegisterServiceHandler#register(…)

ApplicationRegisterServiceHandler#register(Application, StreamObserver<ApplicationMapping>), 代码如下:

  • 第 51 行:获得请求的应用编码( applicationCode )数组

  • 第 54 至 64 行:循环应用编码数组,获取创建应用。

  • 第 57 行:调用 IApplicationIDService#getOrCreate(applicationCode) 方法,获取创建应用,最终获得应用编号( applicationId )。

  • 第 60 至 63 行:获得到应用编号( applicationId != 0 ),则添加到响应。为什么会存在获得不到的情况呢?在下文中,我们会看到,实际异步保存应用,所以会存在获取失败的情况。当获取失败,调用方( 例如 Agent )可以重新发起该请求进行注册应用,从而在异步保存应用,获取到应用编号。

  • 第 67 至 68 行:响应。

2.1.2 IApplicationIDService#getOrCreate(…)

org.skywalking.apm.collector.agent.stream.service.register.IApplicationIDService ,继承 Service 接口,应用编号服务接口。

  • 定义了 #getOrCreate(applicationCode) 接口方法,根据应用编码获取或创建应用,并获得应用编号。

org.skywalking.apm.collector.agent.stream.worker.register.ApplicationIDService ,实现 IApplicationIDService 接口,应用编号服务实现类。

  • 实现了 #getOrCreate(applicationCode) 方法,代码如下:

  • 第 66 行:调用 ApplicationCacheService#get(applicationCode) 方法,从缓存中获取应用编号。在 《SkyWalking 源码分析 —— Collector Cache 缓存组件》 有详细解析。

  • 第 69 至 76 行:当获取不到应用编号时,获取 Application 对应的 Graph<Application> 对象,调用 Graph#start(application) 方法,进行流式处理。在这过程中,会保存应用到存储器。

  • 第 77 行:返回应用编号。

2.1.3 Graph#start(application)

在 #createApplicationRegisterGraph() 方法中,我们可以看到 Application 对应的 Graph<Application> 对象的创建。

  • 第 70 至 74 行:创建当前请求的对应的 Application 记录,并且 applicationId = -1 。
  • ——— 分隔 ———
  • 第 76 行:调用 ApplicationH2RegisterDAO#getMaxApplicationId() 方法,获得 Application 记录的应用编号的最大值
  • 第 77 行:调用 IdAutoIncrement#increment(min, max) 方法,获得应用编号。该方法较为有趣,在下文详细解析。
  • 第 79 至 82 行:创建当前请求的对应的 Application 记录。
  • ——— 分隔 ———
  • 第 85 行:调用 ApplicationEsRegisterDAO#save(Application) 方法,保存应用。

IdAutoIncrement#increment(min, max) 方法,双向均匀自增。可能看起来比较奇怪,以上文 Application 的调用举例子:

minmaxresultapplicationCode
0/1User
0/-1应用 A
-112应用 B
-12-2应用 C
-223应用 D
  • 【User】和【应用 A】是直接获得 result ,不调用 #increment(min, max) 方法。
  • 总的来说,我们可以看到,以 min + max = 0 为中心点( 实际以 0 为中心点), 双向均匀自增。

TODO 【4007】

2.1.4 Application

org.skywalking.apm.collector.storage.table.register.Application ,应用。例如记录在 ES 如下图:

2.2 应用实例的正常注册 API

我们先来看看 API 的定义,InstanceDiscoveryService ,如下图所示:

整体代码和 「2.1 应用的注册 API」 非常相似,所以本小节,更多的是提供代码的链接地址。

2.2.1 InstanceDiscoveryServiceHandler#register(…)

InstanceDiscoveryServiceHandler#register(ApplicationInstance, StreamObserver<ApplicationInstanceMapping>),注册应用实例。

2.2.2 IInstanceIDService#getOrCreate(…)

org.skywalking.apm.collector.agent.stream.service.register.IInstanceIDService ,继承 Service 接口,应用实例编号服务接口。

  • 定义了 #getOrCreate(applicationCode) 接口方法,根据应用编号 + AgentUUID,获取或创建应用实例,并获得应用编号。

org.skywalking.apm.collector.agent.stream.worker.register.InstanceIDService ,实现 IInstanceIDService 接口,应用编号服务实现类。

  • 实现了 #getOrCreate(applicationCode) 方法。

2.2.3 Graph#start(instance)

在 #createInstanceRegisterGraph() 方法中,我们可以看到 Instance 对应的 Graph<Instance> 对象的创建。

2.2.4 Instance

org.skywalking.apm.collector.storage.table.register.Instance ,应用实例。例如记录在 ES 如下图:

2.3 应用实例的恢复注册 API

我们先来看看 API 的定义,InstanceDiscoveryService.proto ,如下图所示:

一般情况下,Agent 在注册应用时候成功后,如果因为各种原因原因和 Collector 断开了 gRPC Channel 连接( 例如,网络 ),恢复连接后,需要调用该 API ,进行恢复注册。

2.3.1 InstanceDiscoveryServiceHandler#recover(…)

InstanceDiscoveryServiceHandler#recover(ApplicationInstanceRecover, StreamObserver<Downstream>), 代码如下:

  • 第 71 行:调用 TimeBucketUtils#getSecondTimeBucket(time) 方法,将 registerTime 转成 timeBucket 。
  • 第 73 行:调用 IInstanceIDService#recover(instanceId, applicationId, registerTime, osInfo) 方法,恢复注册应用实例。
  • 第 75 至 76 行:响应。

2.3.2 IInstanceIDService#recover(…)

InstanceIDService#recover(instanceId, applicationId, registerTime, osInfo) 实现方法,恢复注册。代码如下:

  • 第 96 至 103 行:创建 Instance 对象,用于下面更新操作。

  • 第 99 行: TODO 【4008】

  • 第 106 行:调用 InstanceEsRegisterDAO#save(Instance) 方法,更新应用实例。

2.4 应用实例的心跳 API

我们先来看看 API 的定义,InstanceDiscoveryService.proto ,如下图所示:

一般情况下,Agent 在注册应用时候成功后,定时向 Collector 发送心跳,记录应用存活。

2.4.1 InstanceDiscoveryServiceHandler#heartbeat(…)

InstanceDiscoveryServiceHandler#heartbeat(ApplicationInstanceHeartbeat, StreamObserver<org.skywalking.apm.network.proto.Downstream>) ,目前该方法暂未实现。实现后,会首先调用一个 Service 方法,而后调用 InstanceEsRegisterDAO#updateHeartbeatTime(instanceId, heartbeatTime) 方法,记录应用实例的心跳时间。

3. Agent 调用注册 API

org.skywalking.apm.agent.core.remote.AppAndServiceRegisterClient ,实现 BootService 、GRPCChannelListener 、Runnable 、TracingContextListener 接口,注册应用与实例的客户端。该客户端会调用上述所有 API 。

  • PROCESS_UUID 静态属性,Agent UUID ,使用 UUID 算法生成,去除多余 "-" 。
  • ———- 分割线 ———-
  • status 属性,gRPC 连接状态。
  • applicationRegisterServiceBlockingStub / instanceDiscoveryServiceBlockingStub / serviceNameDiscoveryServiceBlockingStub 属性,对应 gRPC 提供 API 的阻塞 Stub 。
  • needRegisterRecover 属性,是否需要发起恢复的注册。
  • 如上五个属性,在 #statusChanged(GRPCChannelStatus) 实现方法,根据 gRPC 连接状态的变更,创建或销毁 Stub 。
  • 在 #boot() 实现方法,将自己作为监听器( 因为实现了 GRPCChannelListener 接口 )添加到 GRPCChannelManager 中,从而监听 gRPC Channel 的状态。在 《SkyWalking 源码分析 —— Agent Remote 远程通信服务》 有详细解析。
  • ———- 分割线 ———-
  • applicationRegisterFuture 属性,注册应用与实例的定时任务。
  • #boot() 实现方法,创建 applicationRegisterFuture 。该定时任务无初始化延迟,每 Config.Collector.APP_AND_SERVICE_REGISTER_CHECK_INTERVAL ( 默认:3 s ) 执行一次 #run() 方法。
  • ———- 分割线 ———-
  • lastSegmentTime 属性,最后记录 Segment 的时间。
  • #afterFinished() 实现方法,记录 Segment 最后的时间。
  • #afterBoot() 实现方法,将自己作为监听器( 因为实现了 TracingContextListener 接口 )添加到 GRPCChannelManager 中,从而监听 Segment 的记录。在 《SkyWalking 源码分析 —— Agent 收集 Trace 数据》 有详细解析。

#run() 实现方法,执行应用的注册,应用实例的正常注册、恢复注册、心跳的逻辑。

  • 第 123 行:循环,当 gRPC 处于连接中,并且需要重试( shouldTry )。可能对 shouldTry 会比较疑惑?该变量用于,应用的注册成功后,重新标记 shouldTry = true ,继续执行应用实例的注册。
  • 第 126 至 135 行:当本地应用编号为空时,说明应用暂未注册,调用 「2.1 应用的注册 API」 。
  • 第 138 至 148 行:当本地应用实例编号为空时,说明应用实例暂未注册,调用 「2.2 应用实例的正常注册 API」 。
  • 第 150 至 158 行:当需要发起恢复注册时,即 gRPC Channel 断开后重连成功,调用 「2.3 应用实例的恢复注册 API」 。
  • 第 159 至 167 行:当现在时间超过 lastSegmentTime 一分钟,调用 「2.4 应用实例的心跳 API」 。
  • 第 170 至 173 行:同步应用字典、操作字典。在 《SkyWalking 源码分析 —— Agent DictionaryManager 字典管理》 详细解析。
  • 第 178 至 180 行:当发生异常时,调用 GRPCChannelManager#reportError(t) 方法,处理异常,例如请求超时。

相关文章