GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1847 from eeff/eeff/dev

Browse Source 
feat(ekuiper): add rolling metrics
This commit is contained in:
fengzeroz
2024-02-05 15:49:34 +08:00
committed by GitHub
parent 7bfe936ae9 7ce97290bf
commit e943fe0bb5
4 changed files with 47 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/ekuiper/plugin_ekuiper.c
+24
View File
@@ -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 {
plugins/ekuiper/read_write.c
+16 -15
View File
@@ -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);
src/base/metrics.c
+6 -4
View File
@@ -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;
src/core/manager_internal.c
+1
View File
@@ -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;
}