opengauss 源码解析 ​

指标采集、预测与异常检测 ​

代码位置:/gausskernel/dbmind/tools/anomaly_detection

各模组在整体结构上的组合在摩天轮论坛上官方解析文章已经叙述的相对完整详尽。该报告对应为具体模块内部的实现。

Agent 模块 ​

代码位置: ~/agentAgent模块负责数据库指标数据采集与发送，从代码的结构上看，可以分为 3 个子模块，即DBSource(/db_source.py)，MemoryChannel(/channel.py)，HttpSink(/sink.py),负责整合组织各个模块进行协作部分的代码位于~/metric_agent.py 的 agent_main()方法中。

DBSource ​

代码位置:~/agent/db_source.pyDBSource承担数据采集的功能，其承担的三个 task 可以在 agent_main()部分直观的看到,分别为database_exporter,os_exporter,wdr。该三个 task 的代码位于 anomaly_detection/task 中.

在 metric_agent()中的使用

os_exporter ​

负责收集部署 opengauss 的设备上系统的部分性能数据。

可以看到该 task 收集的数据包括 cpu 使用率，io 读写，io 等待，内存使用以及硬盘空间存储部分。

DatabaseExporter ​

负责收集数据库方面的性能数据。

WDR(Workload Diagnosis Report) ​

WDR 基于两次不同时间点系统的性能快照数据，生成两个时间点之间的性能表现报表，用于诊断数据库内核的性能故障。而该部分的 wdr 相关的仅仅是一小部分，wdr中summary级和detail级别的性能数据比起内置的这部分数据要丰富的多。该 task 模块中主要通过 sql 语句进行计数器的查询

MemoryChannel ​

代码位置:~/agent/channel.py 整体为一个存放数据的队列，结构比较简单

HttpSink ​

代码位置:~/agent/sink.py 其从MemoryChannel获取数据，并根据 metric_agent.py 中提供的协议ssl/http进行转发，重试次数为 5 次，间隔 1s

def process(self):
    agent_logger.info('Begin send data to {url}.'.format(url=self._url))
    while self.running:
        contents = self._channel.take()
        if not contents:
            time.sleep(0.5)
            continue

        contents.update(**{'flag': {'host': self.db_host, 'port': self.db_port, 'type': self.db_type}})
        retry_times = 5
        while retry_times:
            try:
                req = request.Request(self._url, headers=_JSON_HEADER,
                    data=json.dumps(contents).encode('utf-8'),
                    method='POST')
                request.urlopen(req, context=self.context)
                break
            except Exception as e:
                agent_logger.error("{error}, retry...".format(error=str(e)))
                retry_times -= 1
                if not retry_times:
                    raise
            time.sleep(1.0)
        time.sleep(self._interval)

Detector ​

代码位置: ~/detector

detector 模块负责数据预测与异常检测，从代码的结构上看，可以分为 3 个部分，即algorithm，service和tools,负责整合组织各个模块进行协作部分的代码位于~/metric_detector.py 中。

receiver ​

代码位置:~/detector/service/resource/receiver.py

receiver为 service 部分中，其功能为接受和存储获取到的性能数据。其中 agent 收集到的三部分 tasks 数据os_exporter,database_exporter以及wdr通过 sqlite 存储在本地。 数据接收:

数据存储：

SQLiteStorage ​

代码位置:~/detector/service/storage/sqlite_storage.py SQLiteStorage实现了一种通过 SQLite 进行本地化存储数据的方式，该类主要是加载 sql 配置，并封装了一些需要的各种 sql 操作以使用,如下:

table.json ​

位于/tools/anomaly_detection 下，为该部分实现时加载的 sql 配置文件

algorithm ​

代码位置:~/detector/algorithmalgorithm为时序预测算法部分，其包括arima和prophet两套时间序列预测算法框架，model中的AlgModel为算法的父类，也提供了使用自定义算法时的模板,使用时若未指定算法会默认使用auto_arima.

fit(): ​

forecast(): ​

trend ​

包括forecast和detect模块，提供时序数据预测与基于阈值的异常检测报警功能

forecast ​

代码位置:~/detector/tools/trend/forecast.py

该模块功能为读取时序数据(数量最少不低于设定值[[minimum_timeseries_length]])，并通过选定的预测算法进行拟合预测，生成数据future_reselt([[status]],[[metric_name]],[[detect_basis]] ([[minimum]],[[maximum]]),[[future_date]],[[future_value]])提供处理

detect ​

代码位置:~/detector/tools/trend/detect.py

该模块根据forecast封装后的数据以及配置中设定的数据阈值进行判断，正常的数据写入[[m_logger]],检测异常的数据，即[[higher]],[[lower]]写入[[a_logger]]警示。