分布式链路追踪 SkyWalking 源码分析 —— Collector Cluster 集群管理
1. 概述
本文主要分享 SkyWalking Collector Cluster Module,负责集群的管理,即 Collector 节点的注册于发现。
友情提示:建议先阅读 《SkyWalking 源码分析 —— Collector 初始化》 ,以了解 Collector 组件体系。
Cluster Module 在 SkyWalking 架构图处于如下位置( 红框 ) :
FROM https://github.com/apache/incubating-skywalking
下面我们来看看整体的项目结构,如下图所示 :
collector-cluster-define
:定义集群管理接口。collector-cluster-standalone-provider
:基于 H2 的 集群管理实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。collector-cluster-redis-provider
:基于 Redis 的集群管理实现。目前暂未完成。collector-cluster-zookeeper-provider
:基于 Zookeeper 的集群管理实现。生产环境推荐使用
下面,我们从接口到实现的顺序进行分享。
2. collector-cluster-define
collector-cluster-define
:定义集群管理接口。项目结构如下 :
交互如下图 :
ModuleListenerService 暴露给其他 Module 注册监听器 ( ClusterModuleListener ) 到 DataMonitor 。
ModuleRegisterService 暴露给其他 Module 注册组件登记( ModuleRegistration ) 到 DataMonitor 。
通过实现 DataMonitor 接口,基于不同的存储器实现注册发现。
2.1 ClusterModule
org.skywalking.apm.collector.cluster.ClusterModule
,实现 Module 抽象类,集群管理 Module 。
#name()
实现方法,返回模块名为 "cluster"
。
#services()
实现方法,返回 Service 类名:ModuleListenerService / ModuleRegisterService 。
2.2 ModuleRegisterService
org.skywalking.apm.collector.cluster.service.ModuleRegisterService
,继承 Service接口,模块注册服务接口。
#register(moduleName, providerName, registration)
接口方法,注册模块注册信息。一般情况下,实现该接口方法,调用 DataMonitor#register(path, registration)
方法。
2.2.1 ModuleRegistration
org.skywalking.apm.collector.cluster.ModuleRegistration
,模块注册信息抽象类。不同 Module 通过实现 ModuleRegistration ,将它们注册到 ModuleRegisterService。目前子类如下 :
#buildValue()
抽象方法,获得模块注册信息( Value )。
2.3 ModuleListenerService
org.skywalking.apm.collector.cluster.service.ModuleListenerService
,继承 Service接口,注册监听器服务接口。
#addListener(listener)
接口方法,添加监听器。一般情况下,实现该接口方法,调用 DataMonitor#addListener(listener)
方法。
2.3.1 ClusterModuleListener
org.skywalking.apm.collector.cluster.ClusterModuleListener
,集群组件监听器抽象类。目前子类如下 :
构造方法,创建地址数组( addresses
)。该数组的读写方法如下:
#addAddress(address)
#removeAddress(address)
#getAddresses()
#path()
抽象方法,返回路径。该路径即为 ClusterModuleListener 监听的"事件"。多个 Collector 节点的相同 Module ,通过路径分组形成集群。
#serverJoinNotify(serverAddress)
/ #serverQuitNotify(serverAddress)
抽象方法,通知服务的加入 / 下线。目前只有 GRPCRemoteSenderService 真正( 其它都是空方法 )实现该方法,在 《SkyWalking 源码分析 —— Collector Remote 远程通信服务》「3.2 GRPCRemoteSenderService」 详细解析。
2.4 DataMonitor
org.skywalking.apm.collector.cluster.DataMonitor
,数据监视器接口。通过实现 DataMonitor 接口,基于不同的存储器实现注册发现。目前子类如下 :
#register(path, registration)
接口方法,注册模块注册信息。
#addListener(ClusterModuleListener)
接口方法,添加监听器。
`#getListener(path)` 接口方法,获得监听指定路径的监听器。
#setClient(Client)
接口方法,设置 Client 。在 client-component
有 ZookeeperClient / H2Client / ElasticSearchClient 等多种实现。
`BASE_CATALOG` 属性,基础目录为
"/skywalking"
。例如说,在 Zookeeper 为根节点的路径。`#createPath(path)` 接口方法,使用 Client 创建路径。
`#setData(path)` 接口方法,使用 Client 设置路径的值。
3. collector-cluster-zookeeper-provider
collector-cluster-zookeeper-provider
,基于 Zookeeper 的集群管理实现。项目结构如下 :
实际使用时,通过 application.yml
配置如下:
JSON cluster: zookeeper: hostPort: localhost:2181 sessionTimeout: 100000
生产环境下,推荐 Zookeeper 配置成集群。
3.1 ClusterModuleZookeeperProvider
org.skywalking.apm.collector.cluster.zookeeper.ClusterModuleZookeeperProvider
,实现 ModuleProvider 抽象类,基于 Zookeeper 的集群管理服务提供者。
#name()
实现方法,返回组件服务提供者名为 "zookeeper"
。
module()
实现方法,返回组件类为 ClusterModule 。
#requiredModules()
实现方法,返回依赖组件为空。
#prepare(Properties)
实现方法,执行准备阶段逻辑。
第 63 行 :创建 ClusterZKDataMonitor 对象。
第 69 行 :创建 ZookeeperClient 对象。注意,此时并未连接 Zookeeper 。
第 71 至 73 行 :创建 ZookeeperModuleListenerService / ZookeeperModuleRegisterService 对象,并调用
#registerServiceImplementation()
父类方法,注册到services
。
#start()
实现方法,执行启动阶段逻辑。
第 79 行 :调用
ZookeeperClient#initialize()
方法,初始化 ZookeeperClient ,此时会连接 Zookeeper。
#notifyAfterCompleted()
实现方法,执行启动完成逻辑。
第 88 行 :调用
ClusterZKDataMonitor#start()
方法,启动 ClusterZKDataMonitor 。在本文 「3.4 ClusterZKDataMonitor」 详细解析。
3.2 ZookeeperModuleRegisterService
org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleRegisterService
,基于 Zookeeper 的模块注册服务实现类。
#register(moduleName, providerName, registration)
实现方法,调用 ClusterZKDataMonitor#register(path, registration)
方法,注册模块注册信息。
3.3 ZookeeperModuleListenerService
org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleListenerService
,基于 Zookeeper 的注册监听器服务实现类。
#addListener(ClusterModuleListener)
实现方法,调用 ClusterZKDataMonitor#addListener(ClusterModuleListener)
方法,注册模块注册信息。
3.4 ClusterZKDataMonitor
org.skywalking.apm.collector.cluster.zookeeper.ClusterZKDataMonitor
,基于 Zookeeper 的数据监视器实现类。
在看具体代码实现之前,我们先来看看 Zookeeper 是如何存储数据的,如下图所示 :
紫色部分,通过调用 `#createPath(path)` 方法,顺着路径,逐层创建持久节点。
黄色部分,通过调用 `#setData(path)` 方法,创建临时节点,设置 Collector 模块地址。若 Collector 集群有 N 个节点,则此处会有 N 个临时节点。
打开
zkClient.sh
,我们来看一个例子 :[zk: localhost:2181(CONNECTED) 1] ls /skywalking
[remote, ui, agent_jetty, agent_gRPC]
[zk: localhost:2181(CONNECTED) 2] ls /skywalking/ui
[jetty]
[zk: localhost:2181(CONNECTED) 3] ls /skywalking/ui/jetty
[localhost:12800]
[zk: localhost:2181(CONNECTED) 4] get /skywalking/ui/jetty/localhost:12800
/
cZxid = 0x24
ctime = Thu Dec 14 16:05:25 CST 2017
mZxid = 0x24
mtime = Thu Dec 14 16:05:25 CST 2017
pZxid = 0x24
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x16052d8b9f40006
dataLength = 1
numChildren = 0
#register(path, registration)
实现方法,添加到组件注册信息集合( registrations
)。
#start()
方法,启动 ClusterZKDataMonitor ,将组件注册信息( registrations
) 写到 Zookeeper 中。
#addListener(listener)
实现方法,添加到监听器集合( listeners
)。
#process(WatchedEvent)
实现方法,处理有 Collector 节点的组件加入或下线。总体逻辑是,从 Zookeeper 获取变更的路径下的地址数组,和本地的地址( ClusterModuleListener.addresses
)比较,处理加入或移除逻辑的地址。
ClusterZKDataMonitor 实现
org.apache.zookeeper.Watcher
接口,所以实现该方法。该方法是
synchronized
方法,以保证不会出现并发问题。
3.5 ZookeeperClient
org.skywalking.apm.collector.client.zookeeper.ZookeeperClient
,实现 Client 接口,Zookeeper 客户端。
代码比较简单,胖友自己阅读理解。
4. collector-cluster-standalone-provider
collector-cluster-standalone-provider.ClusterStandaloneDataMonitor
,基于 H2 的 集群管理实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。项目结构如下 :
大体实现和 collector-cluster-zookeeper-provider
差不多,差异在对 DataMonitor 的实现类 ClusterStandaloneDataMonitor 上。
在 ClusterStandaloneDataMonitor 里,实际并未使用 H2Client ,而是基于内存,胖友可以自己查看下。
5. collector-cluster-redis-provider
collector-cluster-redis-provider
:基于 Redis 的集群管理实现。目前暂未完成。
更多相关文章
- DoDAF2.0方法论探究
- http协议请求方法都有哪些?网络安全学习提升
- 面试官:高并发下重启服务,接口调用老是超时,你有什么解决办法?
- 【前端词典】8 个提高 JS 性能的方法
- AngularJS 日期时间选择组件(附详细使用方法)
- 如何实时主动监控你的网站接口是否挂掉并及时报警
- 5 种方法教你用Python玩转histogram直方图
- 使用Flask在服务器实现一个API接口。
- MySQL+Flask,在本地实现一个API接口。