摘要: 原创出处 http://www.iocoder.cn/SkyWalking/collector-cluster-module/ 「芋道源码」欢迎转载,保留摘要,谢谢!
本文主要基于 SkyWalking 3.2.6 正式版
本文主要分享 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 的集群管理实现。生产环境推荐使用下面,我们从接口到实现的顺序进行分享。
collector-cluster-define
:定义集群管理接口。项目结构如下 :
org.skywalking.apm.collector.cluster.ClusterModule
,实现 Module 抽象类,集群管理 Module 。
#name()
实现方法,返回模块名为 "cluster"
。
#services()
实现方法,返回 Service 类名:ModuleListenerService / ModuleRegisterService 。
org.skywalking.apm.collector.cluster.service.ModuleRegisterService
,继承 Service接口,模块注册服务接口。
#register(moduleName, providerName, registration)
接口方法,注册模块注册信息。一般情况下,实现该接口方法,调用 DataMonitor#register(path, registration)
方法。
org.skywalking.apm.collector.cluster.ModuleRegistration
,模块注册信息抽象类。不同 Module 通过实现 ModuleRegistration ,将它们注册到 ModuleRegisterService。目前子类如下 :
#buildValue()
抽象方法,获得模块注册信息( Value )。
org.skywalking.apm.collector.cluster.service.ModuleListenerService
,继承 Service接口,注册监听器服务接口。
#addListener(listener)
接口方法,添加监听器。一般情况下,实现该接口方法,调用 DataMonitor#addListener(listener)
方法。
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」 详细解析。
org.skywalking.apm.collector.cluster.DataMonitor
,数据监视器接口。通过实现 DataMonitor 接口,基于不同的存储器实现注册发现。目前子类如下 :
#register(path, registration)
接口方法,注册模块注册信息。
#addListener(ClusterModuleListener)
接口方法,添加监听器。
#setClient(Client)
接口方法,设置 Client 。在 client-component
有 ZookeeperClient / H2Client / ElasticSearchClient 等多种实现。
"/skywalking"
。例如说,在 Zookeeper 为根节点的路径。collector-cluster-zookeeper-provider
,基于 Zookeeper 的集群管理实现。项目结构如下 :
实际使用时,通过 application.yml
配置如下:
JSON cluster: zookeeper: hostPort: localhost:2181 sessionTimeout: 100000
org.skywalking.apm.collector.cluster.zookeeper.ClusterModuleZookeeperProvider
,实现 ModuleProvider 抽象类,基于 Zookeeper 的集群管理服务提供者。
#name()
实现方法,返回组件服务提供者名为 "zookeeper"
。
module()
实现方法,返回组件类为 ClusterModule 。
#requiredModules()
实现方法,返回依赖组件为空。
#prepare(Properties)
实现方法,执行准备阶段逻辑。
#registerServiceImplementation()
父类方法,注册到 services
。#start()
实现方法,执行启动阶段逻辑。
ZookeeperClient#initialize()
方法,初始化 ZookeeperClient ,此时会连接 Zookeeper。#notifyAfterCompleted()
实现方法,执行启动完成逻辑。
ClusterZKDataMonitor#start()
方法,启动 ClusterZKDataMonitor 。在本文 「3.4 ClusterZKDataMonitor」 详细解析。org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleRegisterService
,基于 Zookeeper 的模块注册服务实现类。
#register(moduleName, providerName, registration)
实现方法,调用 ClusterZKDataMonitor#register(path, registration)
方法,注册模块注册信息。
org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleListenerService
,基于 Zookeeper 的注册监听器服务实现类。
#addListener(ClusterModuleListener)
实现方法,调用 ClusterZKDataMonitor#addListener(ClusterModuleListener)
方法,注册模块注册信息。
org.skywalking.apm.collector.cluster.zookeeper.ClusterZKDataMonitor
,基于 Zookeeper 的数据监视器实现类。
在看具体代码实现之前,我们先来看看 Zookeeper 是如何存储数据的,如下图所示 :
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
)比较,处理加入或移除逻辑的地址。
org.apache.zookeeper.Watcher
接口,所以实现该方法。synchronized
方法,以保证不会出现并发问题。org.skywalking.apm.collector.client.zookeeper.ZookeeperClient
,实现 Client 接口,Zookeeper 客户端。
代码比较简单,胖友自己阅读理解。
collector-cluster-standalone-provider.ClusterStandaloneDataMonitor
,基于 H2 的 集群管理实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。项目结构如下 :
大体实现和 collector-cluster-zookeeper-provider
差不多,差异在对 DataMonitor 的实现类 ClusterStandaloneDataMonitor 上。
在 ClusterStandaloneDataMonitor 里,实际并未使用 H2Client ,而是基于内存,胖友可以自己查看下。
collector-cluster-redis-provider
:基于 Redis 的集群管理实现。目前暂未完成。
【TODO 4003】等实现后来写写,基于 Redis Pub Sub 保证实时性