SkyWalking 源码解析 —— Collector Remote 远程通信服务

摘要: 原创出处 http://www.iocoder.cn/SkyWalking/collector-remote-module/ 「芋道源码」欢迎转载,保留摘要,谢谢!

本文主要基于 SkyWalking 3.2.6 正式版


1. 概述

本文主要分享 SkyWalking Collector Remote 远程通信服务。该服务用于 Collector 集群内部通信。

目前集群内部通信的目的,跨节点的流式处理。Remote Module 应用在 SkyWalking 架构图如下位置( 红框 ) :

FROM https://github.com/apache/incubating-skywalking

下面我们来看看整体的项目结构,如下图所示 :

  • collector-remote-define :定义远程通信接口。
  • collector-remote-kafka-provider :基于 Kafka 的远程通信实现。目前暂未完成
  • collector-remote-grpc-provider :基于 Google gRPC 的远程通信实现。生产环境目前使用

下面,我们从接口到实现的顺序进行分享。

2. collector-remote-define

collector-remote-define :定义远程通信接口。项目结构如下 :

整体流程如下图:

我们按照整个流程的处理顺序,逐个解析涉及到的类与接口。

2.1 RemoteModule

org.skywalking.apm.collector.remote.RemoteModule ,实现 Module 抽象类,远程通信 Module 。

#name() 实现方法,返回模块名为 "remote"

#services() 实现方法,返回 Service 类名:RemoteSenderService 、RemoteDataRegisterService 。

2.2 RemoteSenderService

org.skywalking.apm.collector.remote.service.RemoteSenderService ,继承 Service 接口,远程发送服务接口,定义了 #send(graphId, nodeId, data, selector) 接口方法,调用 RemoteClient ,发送数据。

2.3 RemoteClientService

org.skywalking.apm.collector.remote.service.RemoteClientService ,继承 Service 接口,远程客户端服务接口,定义了 #create(host, port, channelSize, bufferSize) 接口方法,创建 RemoteClient 对象。

2.4 RemoteClient

org.skywalking.apm.collector.remote.service.RemoteClient ,继承 java.lang.Comparable 接口,远程客户端接口。定义了如下接口方法:

2.5 CommonRemoteDataRegisterService

在说 CommonRemoteDataRegisterService 之前,首先来说下 CommonRemoteDataRegisterService 的意图。

在上文中,我们可以看到发送给 Collector 是 Data 对象,而 Data 是数据的抽象类,在具体反序列化 Data 对象之前,程序是无法得知它是 Data 的哪个实现对象。这个时候,我们可以给 Data 对象的每个实现类,生成一个对应的数据协议编号

  • 在发送数据之前,序列化 Data 对象时,增加该 Data 对应的协议编号,一起发送。
  • 在接收数据之后,反序列化数据时,根据协议编号,创建 Data 对应的实现类对象。

org.skywalking.apm.collector.remote.service.CommonRemoteDataRegisterService ,通用远程数据注册服务。

  • id 属性,数据协议自增编号。
  • dataClassMapping 属性,数据类型( Class<? extends Data> )与数据协议编号的映射。
  • dataInstanceCreatorMapping 属性,数据协议编号与数据对象创建器( RemoteDataInstanceCreator )的映射。

2.5.1 RemoteDataRegisterService

org.skywalking.apm.collector.remote.service.RemoteDataRegisterService ,继承 Service 接口,远程客户端服务接口,定义了 #register(Class<? extends Data>, RemoteDataInstanceCreator) 接口方法,注册数据类型对应的远程数据创建器( RemoteDataRegisterService.RemoteDataInstanceCreator )对象。

CommonRemoteDataRegisterService 实现了 RemoteDataRegisterService 接口,#register(Class<? extends Data>, RemoteDataInstanceCreator) 实现方法。

另外,AgentStreamRemoteDataRegister 会调用 RemoteDataRegisterService#register(Class<? extends Data>, RemoteDataInstanceCreator) 方法,注册每个数据类型的 RemoteDataInstanceCreator 对象。注意,例如 Application::new 是 RemoteDataInstanceCreator 的匿名实现类

2.5.2 RemoteDataIDGetter

org.skywalking.apm.collector.remote.service.RemoteDataIDGetter ,继承 Service 接口,远程数据协议编号获取器接口,定义了 #getRemoteDataId(Class<? extends Data>) 接口方法,根据数据类型获取数据协议编号。

CommonRemoteDataRegisterService 实现了 RemoteDataIDGetter 接口,#getRemoteDataId(Class<? extends Data>) 实现方法。

2.5.3 RemoteDataInstanceCreatorGetter

org.skywalking.apm.collector.remote.service.RemoteDataInstanceCreatorGetter ,继承 Service 接口,远程数据创建器的获取器接口,定义了 #getInstanceCreator(remoteDataId 接口方法,根据数据协议编号获得远程数据创建器( RemoteDataInstanceCreator )。

CommonRemoteDataRegisterService 实现了 RemoteDataInstanceCreatorGetter 接口,#getInstanceCreator(remoteDataId) 实现方法。

2.6 RemoteSerializeService

org.skywalking.apm.collector.remote.service.RemoteSerializeService ,远程通信序列化服务接口,定义了 #serialize(Data) 接口方法,序列化数据,生成 Builder 对象。

2.7 RemoteSerializeService

org.skywalking.apm.collector.remote.service.RemoteDeserializeService ,远程通信序反列化服务接口,定义了 #deserialize(RemoteData, Data) 接口方法,反序列化传输数据。

3. collector-remote-grpc-provider

collector-remote-grpc-provider ,基于 Google gRPC 的远程通信实现。

项目结构如下 :

默认配置,在 application-default.yml 已经配置如下:

remote:
  gRPC:
    host: localhost
    port: 11800

3.1 RemoteModuleGRPCProvider

org.skywalking.apm.collector.remote.grpc.RemoteModuleGRPCProvider ,实现 ModuleProvider 抽象类,基于 gRPC 的组件服务提供者实现类。

#name() 实现方法,返回组件服务提供者名为 "gRPC"

module() 实现方法,返回组件类为 RemoteModule 。

#requiredModules() 实现方法,返回依赖组件为 clustergRPC_manager


#prepare(Properties) 实现方法,执行准备阶段逻辑。

  • 第 53 至 56 行 :创建 CommonRemoteDataRegisterService 、GRPCRemoteSenderService 对象,并调用 #registerServiceImplementation() 父类方法,注册到 services

#start() 实现方法,执行启动阶段逻辑。

#notifyAfterCompleted() 实现方法,方法为空。

3.2 GRPCRemoteSenderService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteSenderService ,继承 ClusterModuleListener 抽象类,实现 RemoteSenderService 接口,基于 gPRC 的远程发送服务实现类。

3.2.1 注册发现

通过继承 ClusterModuleListener 抽象类,实现了监听 Collector 集群节点的加入或离开。

  • remoteClients 属性,连接 Collector 集群节点的客户端数组。每个 Collector 集群节点,对应一个客户端。
  • #path() 实现方法,返回监听的目录 "/" + RemoteModule.NAME + "/" + RemoteModuleGRPCProvider.NAME 。Collector 集群中,每个节点的 Remote Server 都会注册到该目录下。
  • #serverJoinNotify(serverAddress) 实现方法,当新的节点加入,创建新的客户端连接。
  • #serverQuitNotify(serverAddress) 实现方法,当老的节点离开,移除对应的客户端连接。

3.2.2 负载均衡

RemoteModuleGRPCProvider 基于不同的选择器 ( Selector ) ,提供不同的客户端选择( org.skywalking.apm.collector.remote.grpc.service.selector.RemoteClientSelector )实现 :

3.3 GRPCRemoteClientService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteClientService ,实现 RemoteClientService 接口,基于 gRPC 的远程客户端服务实现类。

#create(host, port, channelSize, bufferSize) 实现方法,创建 GRPCRemoteClient 对象。

3.4 GRPCRemoteClient

友情提示:本小节会涉及较多 gRPC 相关的知识,建议不熟悉的胖友自己 Google ,补充下姿势。

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteClient ,实现 RemoteClient 接口,基于 gRPC 的远程客户端实现类。

  • client 属性,GRPCClient 对象。相比来说,GRPCRemoteClient 偏业务的封装,内部调用 GRPCClient 对象。
  • carrier 属性,DataCarrier 对象,本地消息队列。GRPCRemoteClient 在被调用发送数据时,先提交到本地队列,异步消费进行发送到远程 Collector 节点。DataCarrier 在 《SkyWalking 源码分析 —— DataCarrier 异步处理库》 详细解析。
    • 第 63 行:调用 DataCarrier#consume(IConsumer, num) 方法,设置消费者为 RemoteMessageConsumer 对象。

#push(graphId, nodeId, data) 实现方法,异步发送消息到远程 Collector 。

  • 第 73 行:调用 RemoteDataIDGetter#getRemoteDataId(Class<? extends Data>) 方法,获得数据协议编号
  • 第 76 至 80 行:创建传输数据( RemoteMessage.Builder ) 对象。RemoteMessage 通过 Protobuf 创建定义,如下图所示:
  • 第 83 行:调用 DataCarrier#produce(data) 方法,发送数据到本地队列。

RemoteMessageConsumer批量消费本地队列的数据,逐条发送数据到远程 Collector 节点。

  • #consume(List<RemoteMessage>) 实现方法,代码如下:
    • 第 100 行:创建 StreamObserver 对象。StreamObserver 主要是 gPRC 相关的 API 的调用。
    • 第 101 至 103 行:调用 io.grpc.stub.StreamObserver#onNext(RemoteMessage) 方法,逐条发送数据。
    • 第 106 行:调用 io.grpc.stub.StreamObserver#onCompleted() 方法,全部请求数据发送完成

3.5 RemoteCommonServiceHandler

org.skywalking.apm.collector.remote.grpc.handler.RemoteCommonServiceHandler ,实现 org.skywalking.apm.collector.server.grpc.GRPCHandler 接口,继承 RemoteCommonServiceGrpc.RemoteCommonServiceImplBase 抽象类,远程通信通用逻辑处理器。

其中,RemoteCommonServiceGrpc.RemoteCommonServiceImplBase 在 RemoteCommonService.proto 文件的定义如下图:

#call(StreamObserver<Empty>) 实现方法,代码如下:

  • #onNext(RemoteMessage) 方法,处理每一条消息,代码如下:
    • 第 65 行:调用 RemoteDataInstanceCreatorGetter#getInstanceCreator(remoteDataId) 方法,获得数据协议编号对应的 RemoteDataInstanceCreator 对象。然后,调用 RemoteDataInstanceCreator#createInstance(id) 方法,创建数据协议编号对应的 Data 实现类对应的对象。
    • 第 70 行:调用 GraphManager#findGraph(graphId) 方法,获得 graphId 对应的 Graph 对象。然后,调动 GraphNodeFinder#findNext(nodeId) 方法,获得 Next 对象。
    • 第 71 行:调用 Next#execute(Data) 方法,继续流式处理。

3.6 GRPCRemoteSerializeService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteSerializeService ,实现 RemoteSerializeService 接口,基于 gRPC 的远程通信序列化服务实现类。

3.7 GRPCRemoteDeserializeService

org.skywalking.apm.collector.remote.grpc.service.GRPCRemoteDeserializeService ,实现 GRPCRemoteDeserializeService 接口,基于 gRPC 的远程通信反序列化服务实现类。

4. collector-remote-grpc-provider

collector-remote-kafka-provider :基于 Kafka 的远程通信实现。

目前暂未完成

TODO 【4005】collector-remote-grpc-provider

赞(0) 打赏

如未加特殊说明,此网站文章均为原创,转载必须注明出处。Java 技术驿站 » SkyWalking 源码解析 —— Collector Remote 远程通信服务
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

关注【Java 技术驿站】公众号,每天早上 8:10 为你推送一篇技术文章

扫描二维码关注我!


关注【Java 技术驿站】公众号 回复 “VIP”,获取 VIP 地址永久关闭弹出窗口

免费获取资源

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏