写在开篇

玩过 Zabbix 的朋友都了解，Zabbix 提供了与 Elasticsearch（ES）的集成，允许用户将历史数据存储在 Elasticsearch 中，而无需使用传统数据库。在高监控负载的生产环境中，这种架构带来了多重好处。首先，它有效减轻了数据库的负担，从而提高了整体性能。其次，通过将历史数据存储在 Elasticsearch 中，节省了数据库的存储空间。此外，ES 集群可在使用常规 PC 服务器搭建的情况下进行快速扩展，具有低成本的优势。这样的架构特别适合需要延长监控历史数据保留周期的场景，并能充分利用 Elasticsearch 强大的搜索功能。

Zabbix对接Elasticsearch官方参考文档：
https://www.zabbix.com/documentation/5.0/en/manual/appendix/install/elastic_search_setup

监控项类型对应关系如下：

值得注意的是，官方文档也明确指出，当所有历史数据存储在 Elasticsearch 上时，Zabbix 将不再计算趋势，并且这些数据也不会存储在传统数据库中。因此，在选择存储方案时，需要根据实际需求平衡数据分析、趋势计算以及存储成本等因素。

虽然对接Elasticsearch后有诸多好处，但是它不往数据库写历史数据了，同时还不再计算趋势数据了。有这么一个场景，3点需求如下：

• 需求一： 关系型数据库的存储空间不足，且有通过数据库自身的机制定期删除旧的历史监控数据（如表分区，按天、按周、按月，到一定时间后删除旧的表分区）。
• 需求二： 对监控历史数据的保留周期有要求，少则半年，多则1年以上。
• 需求三： 还需要利用到趋势数据用作其他用途，必须要有趋势数据。

面对上述的3点需求，怎么办？个人认为有4种可能的解决方案：

• 方案一： 直接对接数据库，不要对接ES，然后自己写程序，从数据库查询历史数据相关的表，查询到的数据往ES集群里丢，但这会增加数据库IO压力，而且查询的范围和周期不好控制，容易把数据库搞崩溃或者丢数据，并且不敢保证自己的sql是最优的。
• 方案二： 在方案一的基础上，使用第三方组件，我研究过Logstash（是一个开源数据收集引擎），它有JDBC插件，输入端为数据库，输出端为ES，但我发现最终还要自己写sql，个人认为缺点和方案一基本一致。无非就是连接数据库和连接ES的代码不用自己写了，但是查询的sql还是自己写的。
• 方案三： 对接ES，然后写程序或第三方组件从ES获取数据，回写到数据库的历史表，但感觉缺点也是超级明显，不可靠、不可控、不稳定、丢数据、给数据库带来压力。
• 方案四： 修改zabbix源码实现对数据库和Elasticsearch的同时写入，这个方案也是终极方案，能够有效的减轻数据库压力，保证数据一致性，保证程序的稳定性，个人最为是最优的解决方案。

源码修改

我的开发和测试环境说明：

下载源码包后解压之后主要关注src/libs/zbxhistory目录下的源代码文件：

[root@workhost zbxhistory]# ls -l
-rwxrwxrwx 1 root root  24108 Jan  4 16:02 history.c
-rwxrwxrwx 1 root root  36629 Jan  4 14:48 history_elastic.c
-rw-r--r-- 1 root root 126576 Jan  4 14:49 history_elastic.o
-rwxrwxrwx 1 root root   1883 Jan  3 15:09 history.h
-rw-r--r-- 1 root root  48192 Jan  4 13:02 history.o
-rwxrwxrwx 1 root root  32105 Dec 14  2020 history_sql.c
-rw-r--r-- 1 root root  51088 Jan  3 16:40 history_sql.o
-rw-r--r-- 1 root root 227368 Jan  4 14:49 libzbxhistory.a

最重要的3个源代码文件说明：

• history.c: 包含历史数据处理逻辑的主要源代码文件。
• history_elastic.c: 包含与将历史数据写入Elasticsearch相关的源代码文件。
• history_sql.c: 包含与将历史数据写入关系型数据库相关的源代码文件。

主要思路说明：

• 1. 在zabbix_server.conf中，如果配置了HistoryStorageURL、HistoryStorageTypes监控数据就往ES写入，如果没有配置就往数据库写入。所以，这是逻辑问题，而不是功能问题。也就是说监控数据写入关系数据库的功能和写入ES的功能代码不用修改，只需要找到控制往哪里写入的逻辑并按自己的需求调整它即可。
• 2. 所以，只需要关注history.c源代码文件，history.c是历史数据处理逻辑的主要代码文件。

重要说明：从搭建开发和测试环境到调试代码实现需求，满打满算用了1天半的时间，很想把在整个调测过程遇到的问题、心得都分享出来，但发现真的不知道怎么去表达，而且还会长篇大论。所以，我直接把修改后的源码放出来，需要调整的地方我都做好了中文注释，没有中文注释的说明保持原样。

history.c源代码文件：

/*
** Zabbix
** Copyright (C) 2001-2020 Zabbix SIA
**
** This program is free software; you can redistribute it and/or modify
** it under the terms of the GNU General Public License as published by
** the Free Software Foundation; either version 2 of the License, or
** (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU General Public License for more details.
**
** You should have received a copy of the GNU General Public License
** along with this program; if not, write to the Free Software
** Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
**/

#include "common.h"
#include "log.h"
#include "zbxalgo.h"
#include "zbxhistory.h"
#include "history.h"

#include "../zbxalgo/vectorimpl.h"

ZBX_VECTOR_IMPL(history_record, zbx_history_record_t)

extern char	*CONFIG_HISTORY_STORAGE_URL;
extern char	*CONFIG_HISTORY_STORAGE_OPTS;

zbx_history_iface_t	history_ifaces[ITEM_VALUE_TYPE_MAX];

// 增加用于es集群后端存储的全局数组
zbx_history_iface_t	history_ifaces_es[ITEM_VALUE_TYPE_MAX];

// 原来是在zbx_history_init函数里面的，把它放到这里让它变成全局变量
const char	*opts[] = {"dbl", "str", "log", "uint", "text"};

/************************************************************************************
 *                                                                                  *
 * Function: zbx_history_init                                                       *
 *                                                                                  *
 * Purpose: initializes history storage                                             *
 *                                                                                  *
 * Comments: History interfaces are created for all values types based on           *
 *           configuration. Every value type can have different history storage     *
 *           backend.                                                               *
 *                                                                                  *
 ************************************************************************************/
// 这个函数的作用是初始化历史存储，基于为所有值类型创建历史接口配置每种值类型都可以有不同的历史存储后端。
int	zbx_history_init(char **error)
{
	int		i, ret;

	for (i = 0; i < ITEM_VALUE_TYPE_MAX; i++)
	{
		if (NULL == CONFIG_HISTORY_STORAGE_URL || NULL == strstr(CONFIG_HISTORY_STORAGE_OPTS, opts[i]))
			ret = zbx_history_sql_init(&history_ifaces[i], i, error); // 保持不变
		else
		    ret = zbx_history_elastic_init(&history_ifaces_es[i], i, error); // 修改为使用全局数组history_ifaces_es
			
			// 增加初始化sql类型数据库存储后端接口
			ret = zbx_history_sql_init(&history_ifaces[i], i, error);

		if (FAIL == ret)
			return FAIL;
	}
	return SUCCEED;
}

/************************************************************************************
 *                                                                                  *
 * Function: zbx_history_destroy                                                    *
 *                                                                                  *
 * Purpose: destroys history storage                                                *
 *                                                                                  *
 * Comments: All interfaces created by zbx_history_init() function are destroyed    *
 *           here.                                                                  *
 *                                                                                  *
 ************************************************************************************/
// 这个函数的作用是将在zbx_history_init()函数创建的所有接口都在此处销毁。
void zbx_history_destroy(void)
{
    int i;

    for (i = 0; i < ITEM_VALUE_TYPE_MAX; i++)
    {
        // 增加判断条件
        if (NULL == CONFIG_HISTORY_STORAGE_URL || NULL == strstr(CONFIG_HISTORY_STORAGE_OPTS, opts[i]))
        {
            zbx_history_iface_t *writer = &history_ifaces[i];
            writer->destroy(writer);
        }
        else
        {
            // 增加销毁history_ifaces_es
            zbx_history_iface_t *writer_es = &history_ifaces_es[i]; // 修改为使用全局数组history_ifaces_es
            writer_es->destroy(writer_es);
			
			// 同时增加销毁history_ifaces
            zbx_history_iface_t *writer = &history_ifaces[i];
            writer->destroy(writer);			
        }
    }
}


/************************************************************************************
 *                                                                                  *
 * Function: zbx_history_add_values                                                 *
 *                                                                                  *
 * Purpose: Sends values to the history storage                                     *
 *                                                                                  *
 * Parameters: history - [IN] the values to store                                   *
 *                                                                                  *
 * Comments: add history values to the configured storage backends                  *
 *                                                                                  *
 ************************************************************************************/
// 这个函数的作用是将值发送到已配置的存储后端
int	zbx_history_add_values(const zbx_vector_ptr_t *history)
{
	int	i, flags = 0, ret = SUCCEED;

	zabbix_log(LOG_LEVEL_DEBUG, "In %s()", __func__);
	
	for (i = 0; i < ITEM_VALUE_TYPE_MAX; i++)
	{
        // 增加判断条件
		if (NULL == CONFIG_HISTORY_STORAGE_URL || NULL == strstr(CONFIG_HISTORY_STORAGE_OPTS, opts[i]))
		{
			zbx_history_iface_t	*writer = &history_ifaces[i];
			if (0 < writer->add_values(writer, history))
				flags |= (1 << i);
		}
		else
		{
			// 将监控历史数据添加到es后端存储
			zbx_history_iface_t	*writer_es = &history_ifaces_es[i]; // 修改为使用全局数组history_ifaces_es
			if (0 < writer_es->add_values(writer_es, history))
				flags |= (1 << i);
			
			// 同时将监控历史数据添加到sql后端存储
			zbx_history_iface_t	*writer = &history_ifaces[i];
			if (0 < writer->add_values(writer, history))
				flags |= (1 << i);				
		}

	}

	for (i = 0; i < ITEM_VALUE_TYPE_MAX; i++)
	{
        // 增加判断条件
		if (NULL == CONFIG_HISTORY_STORAGE_URL || NULL == strstr(CONFIG_HISTORY_STORAGE_OPTS, opts[i]))
		{
			zbx_history_iface_t	*writer = &history_ifaces[i];
			if (0 != (flags & (1 << i)))
				ret = writer->flush(writer);
		}
		else
		{
			// 将监控历史数据添加到es后端存储
			zbx_history_iface_t	*writer_es = &history_ifaces_es[i]; // 修改为使用全局数组history_ifaces_es
			if (0 != (flags & (1 << i)))
				ret = writer_es->flush(writer_es);
			
			// 同时将监控历史数据添加到sql后端存储
			zbx_history_iface_t	*writer = &history_ifaces[i];
			if (0 != (flags & (1 << i)))
				ret = writer->flush(writer);						
		}

	}

	zabbix_log(LOG_LEVEL_DEBUG, "End of %s()", __func__);

	return ret;
}

// 除了上面需要调整的地方，以下参数保持原样即可，为了控制篇幅就不再贴上来。 
...
...
...

调整源码后，编译并运行：

./configure --prefix=/opt/zabbixsvr5v1 --enable-server --enable-java --enable-ipv6 --with-oracle-include=/usr/include/oracle/21/client64 --with-oracle-header-path=/usr/lib/oracle/21/client64/lib/ --with-oracle-lib=/usr/lib/oracle/21/client64/lib/ --with-net-snmp --with-libcurl --with-openipmi --with-libxml2 --with-unixodbc --with-ssh2 

make

make install

cat > /opt/zabbixsvr5v1/zabbix_server.conf <<EOF
LogFile=/tmp/zabbix_serverv1.log
DBHost=192.168.88.20
DBName=pdb1_zabbix
DBUser=zbxuser
DBPassword=abc123456
DBPort=1521
HistoryStorageURL=192.168.88.24:9200
HistoryStorageTypes=uint,dbl,str,log,text
Timeout=4
LogSlowQueries=3000
AllowRoot=1
StatsAllowedIP=127.0.0.1
EOF

/opt/zabbixsvr5v1/sbin/zabbix_server -c /opt/zabbixsvr5v1/etc/zabbix_server.conf

接着准备两个前端，一个指向数据库查询，另一个指向ES查询：

// 指向ES查询的前端 ./nginx/html/conf/zabbix.conf.php
<?php
$DB['TYPE']                             = 'ORACLE';
$DB['SERVER']                   = '192.168.88.20';
$DB['PORT']                             = '1521';
$DB['DATABASE']                 = 'pdb1_zabbix';
$DB['USER']                             = 'zbxuser';
$DB['PASSWORD']                 = 'abc123456';
$DB['SCHEMA']                   = '';
$DB['ENCRYPTION']               = false;
$DB['KEY_FILE']                 = '';
$DB['CERT_FILE']                = '';
$DB['CA_FILE']                  = '';
$DB['VERIFY_HOST']              = false;
$DB['CIPHER_LIST']              = '';
$DB['DOUBLE_IEEE754']   = true;
$ZBX_SERVER                             = '192.168.88.8';
$ZBX_SERVER_PORT                = '10051';
$ZBX_SERVER_NAME                = 'zabbix-es';
$IMAGE_FORMAT_DEFAULT   = IMAGE_FORMAT_PNG;
$HISTORY['url'] = [
        'uint' => 'http://192.168.88.24:9200',
        'text' => 'http://192.168.88.24:9200',
        'dbl'  => 'http://192.168.88.24:9200',
        'str'  => 'http://192.168.88.24:9200',
        'log'  => 'http://192.168.88.24:9200',
];
$HISTORY['types'] = ['uint','dbl','str','log','text'];


// 指向关系型数据库查询的前端 ./nginx/html/conf/zabbix.conf.php
<?php
$DB['TYPE']                             = 'ORACLE';
$DB['SERVER']                   = '192.168.88.20';
$DB['PORT']                             = '1521';
$DB['DATABASE']                 = 'pdb1_zabbix';
$DB['USER']                             = 'zbxuser';
$DB['PASSWORD']                 = 'abc123456';
$DB['SCHEMA']                   = '';
$DB['ENCRYPTION']               = false;
$DB['KEY_FILE']                 = '';
$DB['CERT_FILE']                = '';
$DB['CA_FILE']                  = '';
$DB['VERIFY_HOST']              = false;
$DB['CIPHER_LIST']              = '';
$DB['DOUBLE_IEEE754']   = true;
$ZBX_SERVER                             = '192.168.88.8';
$ZBX_SERVER_PORT                = '10051';
$ZBX_SERVER_NAME                = 'zabbix-oracle';
$IMAGE_FORMAT_DEFAULT   = IMAGE_FORMAT_PNG;

看最终效果

kibana看索引：

从数据库看最新数据：

从ES看最新数据：

【K8S（专注于深入研究K8S相关的各种技术和知识分享。）】：https://mp.weixin.qq.com/mp/appmsgalbum?action=getalbum&__biz=MzUzMTkyODc4NQ==&scene=1&album_id=2474851867500544003&count=3#wechat_redirect

【Go&Py（涵盖了Go和Python两种流行的编程语言。）】： https://mp.weixin.qq.com/mp/appmsgalbum?action=getalbum&__biz=MzUzMTkyODc4NQ==&scene=1&album_id=2869340550028771330&count=3#wechat_redirect

【Ops（运维领域的讨论和交流。）】：https://mp.weixin.qq.com/mp/appmsgalbum?action=getalbum&__biz=MzUzMTkyODc4NQ==&scene=1&album_id=2869345486221262853&count=3#wechat_redirect

本栏目推荐文章
• Jmeter 如何连接mysql数据库？
• SQL Server 检测是不是数字型数据的两种方法
• 字节微服务HTTP框架Hertz使用与源码分析｜拥抱开
• 如何通过Navicat把MySQL远程数据库导入到本地数据库
• 和利时MACS6通过dat文件实现任意时间范围与测点的历史数据提取与科研分析
• 深入理解spring框架：剖析多线程模式下数据库连接
• 数据可视化是如何帮助大型企业提高效率的？
• redis 数据库
• openpyxl模块--------------------------------重复数据
• Django客户端应用1向服务端应用2发送POST请求并接收解析数据

数据 集群 源码 同时 数据库数据 集群 源码 同时 折线 实际 同时 数据 数据库 数据 多个 同时 relationship同时 数据 数据pageoffice同时 区域 监听器 同时 两个 数据库 数据 同时 问题f460 文件pageoffice同时 数据 数据源 要么 同时 动态 一行 多个 同时 数据