diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index f555ac94..9b3b3827 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -22,8 +22,10 @@ #include #include "errcodes.h" +#include "event/event.h" #include "metric_handle.h" #include "monitor.h" +#include "mqtt_handle.h" #include "utils/http_handler.h" #include "utils/log.h" @@ -159,6 +161,83 @@ static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client, return rv; } +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); +} + +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_info(plugin, "heartbeat timer stopped"); + } +} + +static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval) +{ + 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, + }; + + neu_event_timer_t *timer = neu_event_add_timer(plugin->events, param); + if (NULL == timer) { + plog_error(plugin, "neu_event_add_timer fail"); + return NEU_ERR_EINTERNAL; + } + + if (plugin->heartbeat_timer) { + neu_event_del_timer(plugin->events, plugin->heartbeat_timer); + } + plugin->heartbeat_timer = timer; + + plog_info(plugin, "start_hearbeat_timer interval: %" PRIu64, interval); + + return 0; +} + +static void stop_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; + } + + if (NULL != plugin->mqtt_client) { + neu_mqtt_client_close(plugin->mqtt_client); + neu_mqtt_client_free(plugin->mqtt_client); + plugin->mqtt_client = NULL; + } +} + static neu_plugin_t *monitor_plugin_open(void) { neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t)); @@ -218,27 +297,7 @@ static int monitor_plugin_uninit(neu_plugin_t *plugin) { int rv = 0; - if (NULL != plugin->timer) { - neu_event_del_timer(plugin->events, plugin->timer); - plugin->events = NULL; - } - - 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; - } - - if (NULL != plugin->mqtt_client) { - neu_mqtt_client_close(plugin->mqtt_client); - neu_mqtt_client_free(plugin->mqtt_client); - plugin->mqtt_client = NULL; - } + stop_timers_and_mqtt(plugin); nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name); return rv; @@ -249,7 +308,6 @@ 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 }; - bool started = false; rv = monitor_config_parse(plugin, setting, &config); if (0 != rv) { @@ -257,6 +315,13 @@ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting) return NEU_ERR_NODE_SETTING_INVALID; } + if (config.heartbeat_interval == 0) { + // heartbeat disabled, no need for mqtt functionality + stop_timers_and_mqtt(plugin); + monitor_config_fini(&config); + return 0; + } + if (NULL == plugin->config) { plugin->config = calloc(1, sizeof(*plugin->config)); if (NULL == plugin->config) { @@ -267,8 +332,7 @@ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting) } if (plugin->mqtt_client && neu_mqtt_client_is_open(plugin->mqtt_client)) { - started = true; - rv = neu_mqtt_client_close(plugin->mqtt_client); + rv = neu_mqtt_client_close(plugin->mqtt_client); if (0 != rv) { plog_error(plugin, "neu_mqtt_client_close fail"); rv = NEU_ERR_EINTERNAL; @@ -289,17 +353,22 @@ static int monitor_plugin_config(neu_plugin_t *plugin, const char *setting) goto error; } - if (started && 0 != neu_mqtt_client_open(plugin->mqtt_client)) { + 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->config->host) { - // already configured - monitor_config_fini(plugin->config); + 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` moved, do not call monitor_config_fini plog_info(plugin, "config plugin `%s` success", plugin_name); return 0; @@ -321,6 +390,10 @@ static int monitor_plugin_request(neu_plugin_t * plugin, 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; + } default: nlog_fatal("recv unhandle msg: %s", neu_reqresp_type_string(header->type)); @@ -337,12 +410,25 @@ static int monitor_plugin_start(neu_plugin_t *plugin) if (NULL == plugin->mqtt_client) { plog_info(plugin, "mqtt client is NULL"); - } else 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 != 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_info(plugin, "start plugin `%s` success", plugin_name); } else { plog_error(plugin, "start plugin `%s` failed, error %d", plugin_name, @@ -359,6 +445,8 @@ static int monitor_plugin_stop(neu_plugin_t *plugin) plog_info(plugin, "mqtt client closed"); } + stop_heartbeart_timer(plugin); + plugin->started = false; plog_info(plugin, "stop plugin `%s` success", neu_plugin_module.module_name); return 0; diff --git a/plugins/monitor/monitor.h b/plugins/monitor/monitor.h index 82122edf..726e33ce 100644 --- a/plugins/monitor/monitor.h +++ b/plugins/monitor/monitor.h @@ -34,9 +34,9 @@ struct neu_plugin { neu_plugin_common_t common; pthread_mutex_t mutex; - bool metrics_updating; + bool started; neu_events_t * events; - neu_event_timer_t * timer; + neu_event_timer_t * heartbeat_timer; nng_http_server * api_server; monitor_config_t * config; neu_mqtt_client_t * mqtt_client; diff --git a/plugins/monitor/monitor_config.c b/plugins/monitor/monitor_config.c index 0153ebe3..2ba6afed 100644 --- a/plugins/monitor/monitor_config.c +++ b/plugins/monitor/monitor_config.c @@ -256,6 +256,10 @@ error: void monitor_config_fini(monitor_config_t *config) { + if (NULL == config) { + return; + } + free(config->client_id); free(config->heartbeat_topic); free(config->host);