做积极的人,而不是积极废人!
摘要: 原创出处 http://www.iocoder.cn/SkyWalking/collector-storage-module/ 「芋道源码」欢迎转载,保留摘要,谢谢!
本文主要基于 SkyWalking 3.2.6 正式版
本文主要分享 SkyWalking Collector Storage 存储组件。顾名思义,负责将调用链路、应用、应用实例等等信息存储到存储器,例如,ES 、H2 。
友情提示:建议先阅读 《SkyWalking 源码分析 —— Collector 初始化》 ,以了解 Collector 组件体系。 FROM https://github.com/apache/incubating-skywalking
下面我们来看看整体的项目结构,如下图所示 :
apm-collector-core
的 data
和 define
包 :数据的抽象。collector-storage-define
:定义存储组件接口。collector-storage-h2-provider
:基于 H2 的 存储组件实现。该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。collector-storage-es-provider
:基于 Elasticsearch 的集群管理实现。生产环境推荐使用。下面,我们从接口到实现的顺序进行分享。
apm-collector-core
的 data
和 define
包,如下图所示:
我们对类进行梳理分类,如下图:
org.skywalking.apm.collector.core.data.CommonTable
,通用表。
在 collector-storage-define
的 table
包下,我们可以看到所有 Table 类,以 "Table"
结尾。每个 Table 的表名,在每个实现类里,例如 ApplicationTable 。
org.skywalking.apm.collector.core.data.TableDefine
,表定义抽象类。
不同的存储组件实现,有不同的 TableDefine 实现类,如下图:
collector-storage-es-provider
的 define
包下,我们可以看到所有 ES 的 TableDefine 类。collector-storage-h2-provider
的 `define` 包下,我们可以看到所有 H2 的 TableDefine 类。org.skywalking.apm.collector.core.data.ColumnDefine
,字段定义抽象类。
在 collector-storage-xxx-provider
模块中,H2ColumnDefine 、ElasticSearchColumnDefine 实现 ColumnDefine 。
涉及到的类如下图所示:
org.skywalking.apm.collector.core.data.StorageDefineLoader
,调用 org.skywalking.apm.collector.core.define.DefinitionLoader
,从 org.skywalking.apm.collector.core.data.StorageDefinitionFile
中,加载 TableDefine 实现类数组。
另外,在 collector-storage-es-provider
和 collector-storage-h2-provider
里都有 storage.define
文件,如下图:
StorageInstaller#defineFilter(List)
方法,进行过滤。代码比较简单,中文注释已加,胖友自己阅读理解下。
org.skywalking.apm.collector.core.data.Data
,数据抽象类。
dataXXX
]() 前缀的属性,字段值们。xxxColumns
]() 后缀的属性,字段( Column )们。在 collector-storage-define
的 table
包下,我们可以看到所有 Data 类,非 "Table"
结尾,例如 Application 。
org.skywalking.apm.collector.core.data.Column
,字段。
org.skywalking.apm.collector.core.data.Operation
,操作接口。用于两个值之间的操作,例如,相加等等。目前实现类有:
collector-cluster-define
:定义存储组件接口。项目结构如下 :
org.skywalking.apm.collector.storage.StorageModule
,实现 Module 抽象类,集群管理 Module 。
#name()
实现方法,返回模块名为 "storage"
。
#services()
实现方法,返回 Service 类名:在 org.skywalking.apm.collector.storage.dao 包下的所有类 和 IBatchDAO。
在 org.skywalking.apm.collector.storage.table
包下,定义了存储模块所有的 Table 和 Data 实现类。
org.skywalking.apm.collector.storage.StorageInstaller
,存储安装器抽象类,基于 TableDefine ,初始化存储组件的表。
在 collector-storage-define
项目结构图,我们看到一共有两个 bao
包:
org.skywalking.apm.collector.storage.base.dao
,系统的 DAO 接口。org.skywalking.apm.collector.storage.dao
,业务的 DAO 接口。org.skywalking.apm.collector.storage.base.dao.DAO
,继承 Service 接口,DAO 接口。
无任何方法。
org.skywalking.apm.collector.storage.base.dao.AbstractDAO
,实现 DAO 接口,DAO 抽象基类。
在 collector-storage-xxx-provider
模块中,H2DAO 、EsDAO 实现 AbstractDAO 。
org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO
,实现 DAO 接口,持久化 DAO 接口,定义了 Data 的增删改查操作。
org.elasticsearch.action.index.IndexRequestBuilder
对象。注意:#prepareBatchInsert(data)
方法。org.skywalking.apm.collector.storage.base.dao.IBatchDAO
,实现 DAO 接口,批量操作 DAO 接口。
在 collector-storage-xxx-provider
模块中,BatchH2DAO 、BatchEsDAO 实现 IBatchDAO 。
在 StorageModule#services()
方法里,我们可以看到,业务 DAO 按照用途可以拆分成四种:
那么整理如下:
Package | Data | Cache / Register | Persistence | UI | 关联文章 |
---|---|---|---|---|---|
register | Application | √ | |||
register | Instance | √ | √ | √ | |
register | ServiceName | √ | |||
jvm | CpuMetric | √ | √ | ||
jvm | CMetric | √ | √ | ||
jvm | MemoryMetric | √ | √ | ||
jvm | MemoryPoolMetric | √ | √ | ||
global | GlobalTrace | √ | √ | ||
instance | InstPerformance | √ | √ | ||
node | NodeComponent | √ | √ | ||
node | NodeMapping | √ | √ | ||
noderef | NodeReference | √ | √ | ||
segment | SegmentCost | √ | √ | ||
segment | Segment | √ | √ | ||
service | ServiceEntry | √ | √ | ||
serviceref | ServiceReference | √ | √ |
collector-storage-h2-provider
,基于 H2 的存储组件实现。项目结构如下 :
该实现是单机版,建议仅用于 SkyWalking 快速上手,生产环境不建议使用。
由于生产环境主要使用 ES 的存储组件实现,所以本文暂不解析相关实现,感兴趣的胖友自己嗨起来。
collector-storage-es-provider
,基于 ES 的存储组件实现。项目结构如下 :
实际使用时,通过 application.yml
配置如下:
JSON storage: elasticsearch: cluster_name: elasticsearch cluster_transport_sniffer: true cluster_nodes: 127.0.0.1:9300 index_shards_number: 2 index_replicas_number: 0 ttl: 7
cluster_name
、cluster_transport_sniffer
、cluster_nodes
、index_shards_number
、index_replicas_number
参数,Elasticsearch 相关参数。ttl
:保留 N 天内的数据。超过 N 天的数据,将被自动滚动删除。org.skywalking.apm.collector.storage.es.StorageModuleEsProvider
,实现 ModuleProvider抽象类,基于 ES 的存储组件服务提供者。
#name()
实现方法,返回组件服务提供者名为 "elasticsearch"
。
module()
实现方法,返回组件类为 StorageModule 。
#requiredModules()
实现方法,返回依赖组件为 "cluster"
。
#prepare(Properties)
实现方法,执行准备阶段逻辑。
#registerServiceImplementation()
父类方法,注册到 services
。#start()
实现方法,执行启动阶段逻辑。
ElasticSearchClient#initialize()
方法,初始化 ZookeeperClient 。#notifyAfterCompleted()
实现方法,执行启动完成逻辑。
DataTTLKeeperTimer#start()
方法,启动 DataTTLKeeperTimer 。在本文 「5.4 DataTTLKeeperTimer」 详细解析。在 collector-storage-es-provider
项目结构图,我们看到一共有两个 define
包:
org.skywalking.apm.collector.storage.es.base.define
,系统的 TableDefine 抽象类。org.skywalking.apm.collector.storage.es.define
,业务的 TableDefine 实现类。org.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine
,实现 TableDefine 接口,基于 Elasticsearch 的表定义抽象类。
_type
字段,参见 《Elasticsearch学习笔记》「_type」 。org.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine
,实现 ColumnDefine 抽象类,基于 ES 的字段定义。
在 org.apache.skywalking.apm.collector.storage.es.define
包里,我们可以看到,所有基于 ES 的业务 TableDefine 实现类。例如:ApplicationEsTableDefine 。
整体 #refreshInterval()
方法返回的结果如下:
友情提示:ElasticSearchStorageInstaller 主要是对 Elasticsearch Java API 的使用,所以不熟悉的胖友,可以 Google 下。
org.skywalking.apm.collector.storage.es.base.define.ElasticSearchStorageInstaller
,实现 StorageInstaller 抽象类, 基于 ES 存储安装器实现类。
在 collector-storage-es-provider
项目结构图,我们看到一共有两个 dao
包:
org.skywalking.apm.collector.storage.es.base.dao
,系统的 DAO 抽象类。org.skywalking.apm.collector.storage.es.dao
,业务的 DAO 实现类。org.skywalking.apm.collector.storage.es.base.dao.EsDAO
,实现 AbstractDAO 抽象类,基于 ES 的 DAO 抽象类。
org.skywalking.apm.collector.storage.es.base.dao.BatchEsDAO
,实现 IBatchDAO 接口,继承 EsDAO 抽象类,基于 ES 批量操作 DAO 实现类。
org.elasticsearch.action.index.IndexRequestBuilder
和 org.elasticsearch.action.index.UpdateRequestBuilder
数组,创建成 org.elasticsearch.action.bulk.BulkRequestBuilder
对象,批量持久化。在 org.apache.skywalking.apm.collector.storage.es.dao
包里,我们可以看到,所有基于 ES 的业务 DAO 实现类。
实现代码易懂,胖友可以自己阅读。良心如我们,按照 DAO 的业务用途,推荐例子如下:
org.skywalking.apm.collector.storage.es.DataTTLKeeperTimer
,过期数据删除定时器。通过该定时器,只保留 N 天内的数据。
如下是不会删除的数据的表: