/** * NEURON IIoT System for Industry 4.0 * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. * * This program is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public * License as published by the Free Software Foundation; either * version 3 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 * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser 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. **/ /** * monitor.c - 监控插件主实现文件 * * 该文件实现了Neuron系统的监控插件,主要功能包括: * 1. 通过MQTT协议发送系统监控数据 * 2. 定时发送心跳消息监测节点状态 * 3. 处理各类系统事件并发送相应通知 * 4. 提供系统状态监控和告警功能 */ #include #include #include #include "errcodes.h" #include "event/event.h" #include "monitor.h" #include "mqtt_handle.h" #include "utils/http_handler.h" #include "utils/log.h" extern const neu_plugin_module_t neu_plugin_module; /** * 监控插件全局实例指针 * 用于在其他组件中访问监控插件实例 */ static struct neu_plugin *g_monitor_plugin_; /** * MQTT连接成功回调函数 * * @param data 插件实例指针 */ static void connect_cb(void *data) { neu_plugin_t *plugin = data; plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED; plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name); } /** * MQTT连接断开回调函数 * * @param data 插件实例指针 */ static void disconnect_cb(void *data) { neu_plugin_t *plugin = data; plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; plog_notice(plugin, "plugin `%s` disconnected", neu_plugin_module.module_name); } /** * 配置MQTT客户端参数 * * @param plugin 监控插件实例 * @param client MQTT客户端实例 * @param config 监控配置信息 * @return 0成功,-1失败 */ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client, const monitor_config_t *config) { int rv = 0; if (NULL == client) { return 0; } // 尽早设置日志类别以便于调试 rv = neu_mqtt_client_set_zlog_category(client, plugin->common.log); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_zlog_category fail"); return -1; } // 设置MQTT服务器地址和端口 rv = neu_mqtt_client_set_addr(client, config->host, config->port); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_host fail"); return -1; } // 设置MQTT客户端ID rv = neu_mqtt_client_set_id(client, config->client_id); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_id fail"); return -1; } // 设置连接成功回调函数 rv = neu_mqtt_client_set_connect_cb(client, connect_cb, plugin); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_connect_cb fail"); return -1; } // 设置连接断开回调函数 rv = neu_mqtt_client_set_disconnect_cb(client, disconnect_cb, plugin); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_disconnect_cb fail"); return -1; } // 如果配置了用户名,设置认证信息 if (NULL != config->username) { rv = neu_mqtt_client_set_user(client, config->username, config->password); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_user fail"); } } // 设置TLS/SSL加密参数 rv = neu_mqtt_client_set_tls(client, config->ssl, config->ca, config->cert, config->key, config->keypass); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_set_tsl fail"); return -1; } return rv; } /** * 心跳定时器回调函数 * * 定时请求获取所有节点状态信息,用于监控节点健康状况 * * @param data 插件实例指针 * @return 操作结果 */ static int heartbeat_timer_cb(void *data) { neu_plugin_t *plugin = data; neu_reqresp_head_t header = { .type = NEU_REQ_GET_NODES_STATE, // 请求获取所有节点状态 .ctx = plugin, }; neu_req_get_nodes_state_t cmd = {}; return neu_plugin_op(plugin, header, &cmd); } /** * 停止心跳定时器 * * @param plugin 监控插件实例 */ static inline void stop_heartbeart_timer(neu_plugin_t *plugin) { if (plugin->heartbeat_timer) { neu_event_del_timer(plugin->events, plugin->heartbeat_timer); plugin->heartbeat_timer = NULL; plog_notice(plugin, "heartbeat timer stopped"); } } /** * 启动心跳定时器 * * 根据配置的时间间隔,定时发送获取节点状态请求 * * @param plugin 监控插件实例 * @param interval 心跳间隔(秒) * @return 0成功,非零错误码 */ static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval) { neu_event_timer_t *timer = NULL; // 如果间隔为0,表示禁用心跳 if (0 == interval) { plog_info(plugin, "heartbeat disabled"); goto end; } // 如果事件循环不存在,创建一个 if (NULL == plugin->events) { plugin->events = neu_event_new(); if (NULL == plugin->events) { plog_error(plugin, "neu_event_new fail"); return NEU_ERR_EINTERNAL; } } // 配置定时器参数 neu_event_timer_param_t param = { .second = interval, .millisecond = 0, .cb = heartbeat_timer_cb, .usr_data = plugin, }; // 添加定时器 timer = neu_event_add_timer(plugin->events, param); if (NULL == timer) { plog_error(plugin, "neu_event_add_timer fail"); return NEU_ERR_EINTERNAL; } end: // 如果已存在定时器,先删除旧的 if (plugin->heartbeat_timer) { neu_event_del_timer(plugin->events, plugin->heartbeat_timer); } plugin->heartbeat_timer = timer; plog_notice(plugin, "start_hearbeat_timer interval: %" PRIu64, interval); return 0; } /** * 清理定时器和MQTT客户端资源 * * @param plugin 监控插件实例 */ static void uninit_timers_and_mqtt(neu_plugin_t *plugin) { // 停止心跳定时器 stop_heartbeart_timer(plugin); // 释放事件循环 if (NULL != plugin->events) { neu_event_close(plugin->events); plugin->events = NULL; } // 释放配置 if (NULL != plugin->config) { monitor_config_fini(plugin->config); free(plugin->config); plugin->config = NULL; } // 释放MQTT客户端 if (NULL != plugin->mqtt_client) { neu_mqtt_client_close(plugin->mqtt_client); neu_mqtt_client_free(plugin->mqtt_client); plugin->mqtt_client = NULL; } } /** * 监控插件打开函数 * * 创建并初始化插件实例 * * @return 创建的插件实例指针,失败返回NULL */ static neu_plugin_t *monitor_plugin_open(void) { neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t)); if (NULL == plugin) { return NULL; } // 初始化插件通用部分 neu_plugin_common_init(&plugin->common); // 设置全局插件实例指针 g_monitor_plugin_ = plugin; return plugin; } /** * 监控插件关闭函数 * * 释放插件资源 * * @param plugin 监控插件实例 * @return 操作结果 */ static int monitor_plugin_close(neu_plugin_t *plugin) { int rv = 0; plog_notice(plugin, "Success to free plugin: %s", neu_plugin_module.module_name); free(plugin); return rv; } /** * 监控插件初始化函数 * * @param plugin 监控插件实例 * @param load 是否为加载时初始化 * @return 操作结果 */ static int monitor_plugin_init(neu_plugin_t *plugin, bool load) { int rv = 0; (void) plugin; (void) load; plog_notice(plugin, "Initialize plugin: %s", neu_plugin_module.module_name); return rv; } /** * 监控插件反初始化函数 * * 清理插件资源 * * @param plugin 监控插件实例 * @return 操作结果 */ static int monitor_plugin_uninit(neu_plugin_t *plugin) { int rv = 0; // 清理定时器和MQTT客户端资源 uninit_timers_and_mqtt(plugin); plog_notice(plugin, "Uninitialize plugin: %s", neu_plugin_module.module_name); return rv; } /** * 监控插件配置函数 * * 解析并应用插件设置 * * @param plugin 监控插件实例 * @param setting JSON格式的配置字符串 * @return 操作结果,0成功,非零错误码 */ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting) { int rv = 0; const char *plugin_name = neu_plugin_module.module_name; monitor_config_t config = { 0 }; // 解析配置 rv = monitor_config_parse(plugin, setting, &config); if (0 != rv) { plog_error(plugin, "monitor_config_parse fail"); return NEU_ERR_NODE_SETTING_INVALID; } // 如果配置不存在,分配内存 if (NULL == plugin->config) { plugin->config = calloc(1, sizeof(*plugin->config)); if (NULL == plugin->config) { plog_error(plugin, "calloc monitor config fail"); rv = NEU_ERR_EINTERNAL; goto error; } } // 如果MQTT客户端不存在,创建一个 if (NULL == plugin->mqtt_client) { plugin->mqtt_client = neu_mqtt_client_new(NEU_MQTT_VERSION_V311); if (NULL == plugin->mqtt_client) { plog_error(plugin, "neu_mqtt_client_new fail"); rv = NEU_ERR_EINTERNAL; goto error; } } else if (neu_mqtt_client_is_open(plugin->mqtt_client)) { // 如果MQTT客户端已经打开,先关闭它 rv = neu_mqtt_client_close(plugin->mqtt_client); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_close fail"); rv = NEU_ERR_EINTERNAL; goto error; } } // 配置MQTT客户端 rv = config_mqtt_client(plugin, plugin->mqtt_client, &config); if (0 != rv) { rv = NEU_ERR_MQTT_INIT_FAILURE; goto error; } // 如果插件已启动,打开MQTT连接 if (plugin->started && 0 != neu_mqtt_client_open(plugin->mqtt_client)) { plog_error(plugin, "neu_mqtt_client_open fail"); rv = NEU_ERR_MQTT_CONNECT_FAILURE; goto error; } // 如果插件已启动,启动心跳定时器 if (plugin->started && 0 != start_hearbeat_timer(plugin, config.heartbeat_interval)) { plog_error(plugin, "start_hearbeat_timer fail"); rv = NEU_ERR_EINTERNAL; goto error; } // 更新配置 monitor_config_fini(plugin->config); memmove(plugin->config, &config, sizeof(config)); // 注意:config已移动,不要调用monitor_config_fini plog_notice(plugin, "config plugin `%s` success", plugin_name); return 0; error: plog_error(plugin, "config plugin `%s` fail", plugin_name); monitor_config_fini(&config); return rv; } /** * 监控插件请求处理函数 * * 处理各类系统消息和事件 * * @param plugin 监控插件实例 * @param header 消息头 * @param data 消息数据 * @return 操作结果 */ static int monitor_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *header, void *data) { (void) plugin; switch (header->type) { case NEU_RESP_ERROR: { // 处理错误响应 neu_resp_error_t *resp_err = (neu_resp_error_t *) data; nlog_warn("recv error code: %d", resp_err->error); break; } case NEU_RESP_GET_NODES_STATE: { // 处理节点状态响应 handle_nodes_state(header->ctx, (neu_resp_get_nodes_state_t *) data); break; } // 处理各类系统事件 case NEU_REQ_ADD_NODE_EVENT: // 添加节点事件 case NEU_REQ_DEL_NODE_EVENT: // 删除节点事件 case NEU_REQ_NODE_CTL_EVENT: // 节点控制事件 case NEU_REQ_NODE_SETTING_EVENT: // 节点设置事件 case NEU_REQ_ADD_GROUP_EVENT: // 添加组事件 case NEU_REQ_DEL_GROUP_EVENT: // 删除组事件 case NEU_REQ_UPDATE_GROUP_EVENT: // 更新组事件 case NEU_REQ_ADD_TAG_EVENT: // 添加标签事件 case NEU_REQ_DEL_TAG_EVENT: // 删除标签事件 case NEU_REQ_UPDATE_TAG_EVENT: { // 更新标签事件 handle_events(plugin, header->type, data); break; } default: nlog_warn("recv unsupported msg: %s", neu_reqresp_type_string(header->type)); break; } return 0; } /** * 监控插件启动函数 * * 打开MQTT连接并启动心跳定时器 * * @param plugin 监控插件实例 * @return 操作结果 */ static int monitor_plugin_start(neu_plugin_t *plugin) { int rv = 0; const char *plugin_name = neu_plugin_module.module_name; // 检查MQTT客户端是否存在 if (NULL == plugin->mqtt_client) { plog_notice(plugin, "mqtt client is NULL"); goto end; } // 打开MQTT连接 if (0 != neu_mqtt_client_open(plugin->mqtt_client)) { plog_error(plugin, "neu_mqtt_client_open fail"); rv = NEU_ERR_MQTT_CONNECT_FAILURE; goto end; } // 启动心跳定时器 if (0 != start_hearbeat_timer(plugin, plugin->config->heartbeat_interval)) { plog_error(plugin, "start_hearbeat_timer fail"); neu_mqtt_client_close(plugin->mqtt_client); rv = NEU_ERR_EINTERNAL; goto end; } end: if (0 == rv) { plugin->started = true; plog_notice(plugin, "start plugin `%s` success", plugin_name); } else { plog_error(plugin, "start plugin `%s` failed, error %d", plugin_name, rv); } return 0; } /** * 监控插件停止函数 * * 关闭MQTT连接并停止心跳定时器 * * @param plugin 监控插件实例 * @return 操作结果 */ static int monitor_plugin_stop(neu_plugin_t *plugin) { // 关闭MQTT连接 if (plugin->mqtt_client) { neu_mqtt_client_close(plugin->mqtt_client); plog_notice(plugin, "mqtt client closed"); } // 停止心跳定时器 stop_heartbeart_timer(plugin); plugin->started = false; plog_notice(plugin, "stop plugin `%s` success", neu_plugin_module.module_name); return 0; } /** * 监控插件接口函数表 */ static const neu_plugin_intf_funs_t plugin_intf_funs = { .open = monitor_plugin_open, // 打开插件 .close = monitor_plugin_close, // 关闭插件 .init = monitor_plugin_init, // 初始化插件 .uninit = monitor_plugin_uninit, // 反初始化插件 .start = monitor_plugin_start, // 启动插件 .stop = monitor_plugin_stop, // 停止插件 .setting = monitor_plugin_config, // 配置插件 .request = monitor_plugin_request, // 处理请求 }; #define DEFAULT_MONITOR_PLUGIN_DESCR \ "Builtin plugin for Neuron monitoring and alerting" #define DEFAULT_MONITOR_PLUGIN_DESCR_ZH "内置监控与告警插件" /** * 监控插件模块定义 */ const neu_plugin_module_t neu_plugin_module = { .version = NEURON_PLUGIN_VER_1_0, // 插件版本 .schema = "monitor", // 插件模式 .module_name = "Monitor", // 模块名称 .module_descr = DEFAULT_MONITOR_PLUGIN_DESCR, // 模块描述(英文) .module_descr_zh = DEFAULT_MONITOR_PLUGIN_DESCR_ZH, // 模块描述(中文) .intf_funs = &plugin_intf_funs, // 接口函数表 .kind = NEU_PLUGIN_KIND_SYSTEM, // 插件类型:系统插件 .type = NEU_NA_TYPE_APP, // 应用类型 .display = true, // 是否显示 .single = false, // 是否单例 .single_name = "monitor", // 单例名称 }; /** * 获取监控插件实例 * * @return 监控插件实例指针 */ neu_plugin_t *neu_monitor_get_plugin() { return g_monitor_plugin_; }