From f197f574a05891c61a92b64ff34a038eccd9b5ac Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Sun, 4 Feb 2024 16:49:00 +0800 Subject: [PATCH 1/3] feat(ekuiper): add rolling metrics --- plugins/ekuiper/plugin_ekuiper.c | 24 ++++++++++++++++++++++++ plugins/ekuiper/read_write.c | 31 ++++++++++++++++--------------- 2 files changed, 40 insertions(+), 15 deletions(-) diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 46df341b..c3156c9b 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -71,6 +71,10 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg) nng_mtx_lock(plugin->mtx); plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; nng_mtx_unlock(plugin->mtx); + + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL); } static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load) @@ -96,6 +100,22 @@ static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load) plugin->recv_aio = recv_aio; + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000); + plog_notice(plugin, "plugin initialized"); return rv; } @@ -293,6 +313,10 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, case NEU_REQRESP_TRANS_DATA: { neu_reqresp_trans_data_t *trans_data = data; + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL); + if (disconnected) { plog_debug(plugin, "not connected"); } else { diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c index 725a767f..7b032248 100644 --- a/plugins/ekuiper/read_write.c +++ b/plugins/ekuiper/read_write.c @@ -30,9 +30,6 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data) { int rv = 0; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - char * json_str = NULL; json_read_resp_t resp = { .plugin = plugin, @@ -59,15 +56,18 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data) rv = nng_sendmsg(plugin->sock, msg, NNG_FLAG_NONBLOCK); // TODO: use aio to send message if (0 == rv) { - update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, json_len, - NULL); - update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1, - NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len, + NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, json_len, + NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, json_len, + NULL); } else { plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv)); nng_msg_free(msg); - update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, - 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1, + NULL); } } @@ -80,9 +80,6 @@ void recv_data_callback(void *arg) char * json_str = NULL; json_write_req_t *req = NULL; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - rv = nng_aio_result(plugin->recv_aio); if (0 != rv) { plog_error(plugin, "nng_recv error: %s", nng_strerror(rv)); @@ -93,9 +90,13 @@ void recv_data_callback(void *arg) json_str = nng_msg_body(msg); json_len = nng_msg_len(msg); plog_debug(plugin, "<< %.*s", (int) json_len, json_str); - update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, json_len, - NULL); - update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, json_len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, json_len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, json_len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL); if (json_decode_write_req(json_str, json_len, &req) < 0) { plog_error(plugin, "fail decode write request json: %.*s", (int) nng_msg_len(msg), json_str); From e8ab18f0252b540f0431f734efe4e83a8047a7a7 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Sun, 4 Feb 2024 18:40:08 +0800 Subject: [PATCH 2/3] fix(metrics): correct registered rolling counter entry ref count --- src/base/metrics.c | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/base/metrics.c b/src/base/metrics.c index deaffb85..60f95ed5 100644 --- a/src/base/metrics.c +++ b/src/base/metrics.c @@ -263,8 +263,8 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name, } if (NEU_METRIC_TYPE_ROLLING_COUNTER == type) { - entry->rcnt = neu_rolling_counter_new(init); - if (NULL == entry->rcnt) { + // only allocate rolling counter for nonzero time span + if (init > 0 && NULL == (entry->rcnt = neu_rolling_counter_new(init))) { free(entry); return -1; } @@ -311,9 +311,11 @@ int neu_metrics_register_entry(const char *name, const char *help, neu_metric_entry_t *e = NULL; pthread_rwlock_wrlock(&g_metrics_mtx_); + // use `value` field as reference counter, initialize to zero + // and we don't need to allocate rolling counter for register entries rv = neu_metric_entries_add(&g_metrics_.registered_metrics, name, help, - type, 1); - if (1 == rv) { + type, 0); + if (-1 != rv) { HASH_FIND_STR(g_metrics_.registered_metrics, name, e); ++e->value; rv = 0; From 7ce97290bf0c34c061766e65e1f139efc809277a Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Sun, 4 Feb 2024 19:51:51 +0800 Subject: [PATCH 3/3] fix(core): memory leak --- src/core/manager_internal.c | 1 + 1 file changed, 1 insertion(+) diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 7e303883..ce17f257 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -91,6 +91,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name, 0 != (ret = neu_adapter_set_setting(adapter, setting))) { neu_node_manager_del(manager->node_manager, node_name); neu_adapter_uninit(adapter); + neu_adapter_destroy(adapter); return ret; }