From e53226fc4e912be26193e86bf656364b8a6dea20 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 18 Oct 2022 22:56:48 +0800 Subject: [PATCH 1/2] fix(monitor): correct global metrics type --- plugins/monitor/metric_handle.c | 52 +++++++++++++++++---------------- 1 file changed, 27 insertions(+), 25 deletions(-) diff --git a/plugins/monitor/metric_handle.c b/plugins/monitor/metric_handle.c index 5b969280..b7f5f18f 100644 --- a/plugins/monitor/metric_handle.c +++ b/plugins/monitor/metric_handle.c @@ -25,31 +25,33 @@ #include "restful/handle.h" #include "restful/http.h" -#define METRIC_GLOBAL_TMPL \ - "# HELP core_dumped Whether there is any core dump\n" \ - "# TYPE core_dumped counter\n" \ - "core_dumped %d\n" \ - "# HELP uptime_seconds Uptime in seconds\n" \ - "# TYPE uptime_seconds counter\n" \ - "uptime_seconds %" PRIu64 "\n" \ - "# HELP north_nodes Number of north nodes\n" \ - "# TYPE north_nodes counter\n" \ - "north_nodes %zu\n" \ - "# HELP north_running_nodes Number of north nodes in running state\n" \ - "# TYPE north_running_nodes counter\n" \ - "north_running_nodes %zu\n" \ - "# HELP north_disconnected_nodes Number of north nodes disconnected\n" \ - "# TYPE north_disconnected_nodes counter\n" \ - "north_disconnected_nodes %zu\n" \ - "# HELP south_nodes Number of south nodes\n" \ - "# TYPE south_nodes counter\n" \ - "south_nodes %zu\n" \ - "# HELP south_running_nodes Number of south nodes in running state\n" \ - "# TYPE south_running_nodes counter\n" \ - "south_running_nodes %zu\n" \ - "# HELP south_disconnected_nodes Number of south nodes disconnected\n" \ - "# TYPE south_disconnected_nodes counter\n" \ - "south_disconnected_nodes %zu\n" +// clang-format off +#define METRIC_GLOBAL_TMPL \ + "# HELP core_dumped Whether there is any core dump\n" \ + "# TYPE core_dumped gauge\n" \ + "core_dumped %d\n" \ + "# HELP uptime_seconds Uptime in seconds\n" \ + "# TYPE uptime_seconds counter\n" \ + "uptime_seconds %" PRIu64 "\n" \ + "# HELP north_nodes_total Number of north nodes\n" \ + "# TYPE north_nodes_total gauge\n" \ + "north_nodes_total %zu\n" \ + "# HELP north_running_nodes_total Number of north nodes in running state\n" \ + "# TYPE north_running_nodes_total gauge\n" \ + "north_running_nodes_total %zu\n" \ + "# HELP north_disconnected_nodes_total Number of north nodes disconnected\n" \ + "# TYPE north_disconnected_nodes_total gauge\n" \ + "north_disconnected_nodes_total %zu\n" \ + "# HELP south_nodes_total Number of south nodes\n" \ + "# TYPE south_nodes_total gauge\n" \ + "south_nodes_total %zu\n" \ + "# HELP south_running_nodes_total Number of south nodes in running state\n" \ + "# TYPE south_running_nodes_total gauge\n" \ + "south_running_nodes_total %zu\n" \ + "# HELP south_disconnected_nodes_total Number of south nodes disconnected\n" \ + "# TYPE south_disconnected_nodes_total gauge\n" \ + "south_disconnected_nodes_total %zu\n" +// clang-format on static int response(nng_aio *aio, char *content, enum nng_http_status status) { From 27d7b2d4d5bd57d210b3051de0e7339337aafda3 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 20 Oct 2022 16:39:18 +0800 Subject: [PATCH 2/2] refactor: node metrics data structures --- include/neuron/adapter.h | 63 ++----------- include/neuron/metrics.h | 157 +++++++++++++++++++++++++++++++ plugins/ekuiper/plugin_ekuiper.c | 4 - plugins/ekuiper/read_write.c | 16 ++-- plugins/modbus/modbus_req.c | 18 ++-- src/adapter/adapter.c | 139 +++++++++++++++++---------- src/adapter/adapter_internal.h | 15 +-- src/adapter/driver/driver.c | 15 ++- src/core/node_manager.c | 7 +- 9 files changed, 287 insertions(+), 147 deletions(-) create mode 100644 include/neuron/metrics.h diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 23cfffbb..b667ee8e 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -27,65 +27,10 @@ extern "C" { #include #include "define.h" +#include "metrics.h" #include "tag.h" #include "type.h" -#define NEU_NODE_STAT_RTT_MAX 9999 - -/** Kinds of node statistics counters. - * - * Some counters are maintained by the neuron core, while others should be - * maintained by the plugin code using the adapter callback `stat_acc`. - */ -typedef enum { - // kinds of counters maintained by the plugins - NEU_NODE_STAT_BYTES_SENT, // number of bytes sent - NEU_NODE_STAT_BYTES_RECV, // number of bytes received - NEU_NODE_STAT_MSGS_SENT, // number of messages sent - NEU_NODE_STAT_MSGS_RECV, // number of messages received - NEU_NODE_STAT_AVG_RTT, // average round trip time (ms) - - // kinds of counters maintained by the driver core - NEU_NODE_STAT_TAG_TOT_CNT, // number of tag read including errors - NEU_NODE_STAT_TAG_ERR_CNT, // number of tag read errors - - NEU_NODE_STAT_MAX, -} neu_node_stat_e; - -typedef enum { - NEU_METRICS_CATEGORY_GLOBAL, - NEU_METRICS_CATEGORY_DRIVER, - NEU_METRICS_CATEGORY_APP, - NEU_METRICS_CATEGORY_ALL, -} neu_metrics_category_e; - -typedef struct { - bool core_dumped; - uint64_t uptime_seconds; - size_t north_nodes; - size_t north_running_nodes; - size_t north_disconnected_nodes; - size_t south_nodes; - size_t south_running_nodes; - size_t south_disconnected_nodes; -} neu_metrics_t; - -static inline const char *neu_node_stat_string(neu_node_stat_e s) -{ - static const char *map[] = { - [NEU_NODE_STAT_BYTES_SENT] = "bytes_sent", - [NEU_NODE_STAT_BYTES_RECV] = "bytes_received", - [NEU_NODE_STAT_MSGS_SENT] = "messages_sent", - [NEU_NODE_STAT_MSGS_RECV] = "messages_received", - [NEU_NODE_STAT_AVG_RTT] = "average_rtt", - [NEU_NODE_STAT_TAG_TOT_CNT] = "tag_total_count", - [NEU_NODE_STAT_TAG_ERR_CNT] = "tag_error_count", - [NEU_NODE_STAT_MAX] = NULL, - }; - - return map[s]; -} - typedef struct { neu_node_running_state_e running; neu_node_link_state_e link; @@ -494,11 +439,15 @@ typedef struct neu_adapter neu_adapter_t; typedef struct neu_adapter_driver neu_adapter_driver_t; typedef struct neu_adapter_app neu_adapter_app_t; +typedef int (*neu_adapter_update_metric_cb_t)(neu_adapter_t *adapter, + const char * metric_name, + uint64_t n); + typedef struct adapter_callbacks { int (*command)(neu_adapter_t *adapter, neu_reqresp_head_t head, void *data); int (*response)(neu_adapter_t *adapter, neu_reqresp_head_t *head, void *data); - void (*stat_acc)(neu_adapter_t *adapter, neu_node_stat_e s, uint64_t n); + neu_adapter_update_metric_cb_t update_metric; union { struct { diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h new file mode 100644 index 00000000..976b3641 --- /dev/null +++ b/include/neuron/metrics.h @@ -0,0 +1,157 @@ +/** + * 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. + **/ + +#ifndef NEU_METRICS_H +#define NEU_METRICS_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "define.h" +#include "type.h" +#include "utils/uthash.h" + +typedef enum { + NEU_METRIC_TYPE_COUNTER, + NEU_METRIC_TYPE_GAUAGE, +} neu_metric_type_e; + +// last round trip time in millisends +#define NEU_METRIC_LAST_RTT_MS "last_rtt_ms" +#define NEU_METRIC_LAST_RTT_MS_TYPE NEU_METRIC_TYPE_GAUAGE +#define NEU_METRIC_LAST_RTT_MS_HELP \ + "Last request round trip time in milliseconds" + +// number of bytes sent +#define NEU_METRIC_SEND_BYTES "send_bytes" +#define NEU_METRIC_SEND_BYTES_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_SEND_BYTES_HELP "Total number of bytes sent" + +// number of bytes received +#define NEU_METRIC_RECV_BYTES "recv_bytes" +#define NEU_METRIC_RECV_BYTES_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_RECV_BYTES_HELP "Total number of bytes received" + +// number of tag read including errors +#define NEU_METRIC_TAG_READS_TOTAL "tag_reads_total" +#define NEU_METRIC_TAG_READS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_TAG_READS_TOTAL_HELP \ + "Total number of tag reads including errors" + +// number of tag read errors +#define NEU_METRIC_TAG_READ_ERRORS_TOTAL "tag_read_errors_total" +#define NEU_METRIC_TAG_READ_ERRORS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_TAG_READ_ERRORS_TOTAL_HELP "Total number of tag read errors" + +typedef enum { + NEU_METRICS_CATEGORY_GLOBAL, + NEU_METRICS_CATEGORY_DRIVER, + NEU_METRICS_CATEGORY_APP, + NEU_METRICS_CATEGORY_ALL, +} neu_metrics_category_e; + +// metric entry +typedef struct { + const char * name; // NOTE: should points to string literal + const char * help; // NOTE: should points to string literal + neu_metric_type_e type; // + uint64_t value; // + UT_hash_handle hh; // ordered by name +} neu_metric_entry_t; + +// group metrics +typedef struct { + char * name; // group name + neu_metric_entry_t *entries; // group metric entries + UT_hash_handle hh; // ordered by name +} neu_group_metrics_t; + +// node metrics +typedef struct { + neu_node_type_e type; // node type + char * name; // node name + neu_metric_entry_t * entries; // node metric entries + neu_group_metrics_t *group_metrics; // group metrics + UT_hash_handle hh; // ordered by name +} neu_node_metrics_t; + +typedef struct { + bool core_dumped; + uint64_t uptime_seconds; + size_t north_nodes; + size_t north_running_nodes; + size_t north_disconnected_nodes; + size_t south_nodes; + size_t south_running_nodes; + size_t south_disconnected_nodes; + neu_node_metrics_t *node_metrics; +} neu_metrics_t; + +static inline void neu_metric_entry_free(neu_metric_entry_t *entry) +{ + free(entry); +} + +static inline void neu_group_metrics_free(neu_group_metrics_t *group_metrics) +{ + if (NULL == group_metrics) { + return; + } + + neu_metric_entry_t *e = NULL, *tmp = NULL; + HASH_ITER(hh, group_metrics->entries, e, tmp) + { + HASH_DEL(group_metrics->entries, e); + neu_metric_entry_free(e); + } + free(group_metrics->name); + free(group_metrics); +} + +static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics) +{ + if (NULL == node_metrics) { + return; + } + + neu_metric_entry_t *e = NULL, *etmp = NULL; + HASH_ITER(hh, node_metrics->entries, e, etmp) + { + HASH_DEL(node_metrics->entries, e); + neu_metric_entry_free(e); + } + + neu_group_metrics_t *g = NULL, *gtmp = NULL; + HASH_ITER(hh, node_metrics->group_metrics, g, gtmp) + { + HASH_DEL(node_metrics->group_metrics, g); + neu_group_metrics_free(g); + } + + free(node_metrics); +} + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 1bd481c0..c2e29304 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -54,8 +54,6 @@ static void pipe_add_cb(nng_pipe p, nng_pipe_ev ev, void *arg) neu_plugin_t *plugin = arg; nng_mtx_lock(plugin->mtx); plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED; - plugin->common.adapter_callbacks->stat_acc(plugin->common.adapter, - NEU_NODE_STAT_AVG_RTT, 0); nng_mtx_unlock(plugin->mtx); } @@ -66,8 +64,6 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg) neu_plugin_t *plugin = arg; nng_mtx_lock(plugin->mtx); plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; - plugin->common.adapter_callbacks->stat_acc( - plugin->common.adapter, NEU_NODE_STAT_AVG_RTT, NEU_NODE_STAT_RTT_MAX); nng_mtx_unlock(plugin->mtx); } diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c index 89d4ce36..3d0264a5 100644 --- a/plugins/ekuiper/read_write.c +++ b/plugins/ekuiper/read_write.c @@ -29,8 +29,9 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data) { int rv = 0; - void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) = - plugin->common.adapter_callbacks->stat_acc; + + neu_adapter_update_metric_cb_t update_metric = + plugin->common.adapter_callbacks->update_metric; char * json_str = NULL; json_read_resp_t resp = { @@ -58,8 +59,7 @@ 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) { - stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_SENT, 1); - stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_SENT, json_len); + update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, json_len); } else { plog_error(plugin, "nng cannot send msg"); nng_msg_free(msg); @@ -74,8 +74,9 @@ void recv_data_callback(void *arg) size_t json_len = 0; char * json_str = NULL; json_write_req_t *req = NULL; - void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) = - plugin->common.adapter_callbacks->stat_acc; + + neu_adapter_update_metric_cb_t update_metric = + plugin->common.adapter_callbacks->update_metric; rv = nng_aio_result(plugin->recv_aio); if (0 != rv) { @@ -87,8 +88,7 @@ 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); - stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_RECV, 1); - stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_RECV, json_len); + update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, json_len); 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); diff --git a/plugins/modbus/modbus_req.c b/plugins/modbus/modbus_req.c index 9f1e7852..84bdbd52 100644 --- a/plugins/modbus/modbus_req.c +++ b/plugins/modbus/modbus_req.c @@ -49,14 +49,14 @@ void modbus_conn_disconnected(void *data, int fd) int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes) { neu_plugin_t *plugin = (neu_plugin_t *) ctx; - void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) = - plugin->common.adapter_callbacks->stat_acc; + + neu_adapter_update_metric_cb_t update_metric = + plugin->common.adapter_callbacks->update_metric; plog_send_protocol(plugin, bytes, n_byte); int ret = neu_conn_send(plugin->conn, bytes, n_byte); if (ret > 0) { - stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_SENT, 1); - stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_SENT, ret); + update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, ret); } return ret; } @@ -303,19 +303,17 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size) uint8_t * recv_buf = calloc(response_size, 1); neu_protocol_unpack_buf_t pbuf = { 0 }; ssize_t ret = 0; - void (*stat_acc)(neu_adapter_t *, neu_node_stat_e, uint64_t) = - plugin->common.adapter_callbacks->stat_acc; + + neu_adapter_update_metric_cb_t update_metric = + plugin->common.adapter_callbacks->update_metric; ret = neu_conn_recv(plugin->conn, recv_buf, response_size); if (ret > 0) { - stat_acc(plugin->common.adapter, NEU_NODE_STAT_BYTES_RECV, ret); + update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, ret); neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret); plog_recv_protocol(plugin, recv_buf, ret); ret = modbus_stack_recv(plugin->stack, &pbuf); - if (ret > 0) { - stat_acc(plugin->common.adapter, NEU_NODE_STAT_MSGS_RECV, 1); - } } free(recv_buf); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index a759444d..c46c4acc 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -40,23 +41,39 @@ #include "plugin.h" #include "storage.h" -static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data); -static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, +static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data); +static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, + void *data); +static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); -static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, - void *data); -static void adapter_stat_acc(neu_adapter_t *adapter, neu_node_stat_e s, - uint64_t n); +static int adapter_register_metric(neu_adapter_t *adapter, const char *name, + neu_metric_type_e type, const char *help, + uint64_t init); +static int adapter_update_metric(neu_adapter_t *adapter, + const char *metric_name, uint64_t n); inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); static int update_timestamp(void *usr_data); +static pthread_rwlock_t g_node_metrics_mtx_ = PTHREAD_RWLOCK_INITIALIZER; +static neu_metrics_t g_node_metrics_; + static const adapter_callbacks_t callback_funs = { - .command = adapter_command, - .response = adapter_response, - .stat_acc = adapter_stat_acc, + .command = adapter_command, + .response = adapter_response, + .update_metric = adapter_update_metric, }; +#define REGISTER_METRIC(adapter, name, init) \ + adapter_register_metric(adapter, name, name##_TYPE, name##_HELP, init); + +#define REGISTER_DRIVER_METRICS(adapter) \ + REGISTER_METRIC(adapter, NEU_METRIC_LAST_RTT_MS, 9999); \ + REGISTER_METRIC(adapter, NEU_METRIC_SEND_BYTES, 0); \ + REGISTER_METRIC(adapter, NEU_METRIC_RECV_BYTES, 0); \ + REGISTER_METRIC(adapter, NEU_METRIC_TAG_READS_TOTAL, 0); \ + REGISTER_METRIC(adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, 0) + neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) { int rv = 0; @@ -77,16 +94,21 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) break; } - adapter->name = strdup(info->name); - adapter->events = neu_event_new(); - adapter->state = NEU_NODE_RUNNING_STATE_INIT; - adapter->handle = info->handle; - adapter->cb_funs.command = callback_funs.command; - adapter->cb_funs.response = callback_funs.response; - adapter->cb_funs.stat_acc = callback_funs.stat_acc; - adapter->module = info->module; + adapter->name = strdup(info->name); + adapter->events = neu_event_new(); + adapter->state = NEU_NODE_RUNNING_STATE_INIT; + adapter->handle = info->handle; + adapter->cb_funs.command = callback_funs.command; + adapter->cb_funs.response = callback_funs.response; + adapter->cb_funs.update_metric = callback_funs.update_metric; + adapter->module = info->module; - adapter->stat.avg_rtt = NEU_NODE_STAT_RTT_MAX; + adapter->metrics = calloc(1, sizeof(*adapter->metrics)); + if (NULL != adapter->metrics) { + adapter->metrics->type = info->module->type; + adapter->metrics->name = adapter->name; + HASH_ADD_STR(g_node_metrics_.node_metrics, name, adapter->metrics); + } rv = nng_pair1_open(&adapter->sock); assert(rv == 0); @@ -97,6 +119,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) switch (info->module->type) { case NEU_NA_TYPE_DRIVER: + REGISTER_DRIVER_METRICS(adapter); neu_adapter_driver_init((neu_adapter_driver_t *) adapter); break; case NEU_NA_TYPE_APP: @@ -166,41 +189,46 @@ neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter) return adapter->module->type; } -static void adapter_stat_acc(neu_adapter_t *adapter, neu_node_stat_e s, - uint64_t n) +static int adapter_register_metric(neu_adapter_t *adapter, const char *name, + neu_metric_type_e type, const char *help, + uint64_t init) { - switch (s) { - case NEU_NODE_STAT_BYTES_SENT: - adapter->stat.bytes_sent += n; - break; - case NEU_NODE_STAT_BYTES_RECV: - adapter->stat.bytes_recv += n; - break; - case NEU_NODE_STAT_MSGS_SENT: - adapter->stat.msgs_sent += n; - break; - case NEU_NODE_STAT_MSGS_RECV: - adapter->stat.msgs_recv += n; - break; - case NEU_NODE_STAT_AVG_RTT: { - if (NEU_NODE_STAT_RTT_MAX <= n || - adapter->stat.avg_rtt == NEU_NODE_STAT_RTT_MAX) { - adapter->stat.avg_rtt = n; - break; - } - - // exponential moving average with alpha = 0.3 - const double a = 0.3; - adapter->stat.avg_rtt = adapter->stat.avg_rtt * (1 - a) + n * a; - break; + if (NULL == adapter->metrics) { + return -1; } - // these are maintained by neuron core - case NEU_NODE_STAT_TAG_TOT_CNT: - case NEU_NODE_STAT_TAG_ERR_CNT: - default: - assert(!"please supply a valid statistics counter kind"); + neu_metric_entry_t *entry = calloc(1, sizeof(*entry)); + if (NULL == entry) { + return -1; } + + entry->name = name; + entry->type = type; + entry->help = help; + entry->value = init; + HASH_ADD_STR(adapter->metrics->entries, name, entry); + return 0; +} + +static int adapter_update_metric(neu_adapter_t *adapter, + const char *metric_name, uint64_t n) +{ + neu_metric_entry_t *entry = NULL; + if (NULL != adapter->metrics) { + HASH_FIND_STR(adapter->metrics->entries, metric_name, entry); + } + + if (NULL == entry) { + return -1; + } + + if (NEU_METRIC_TYPE_COUNTER == entry->type) { + entry->value += n; + } else { + entry->value = n; + } + + return 0; } static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, @@ -402,8 +430,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_GET_NODE_STATE: { neu_resp_get_node_state_t resp = { 0 }; + neu_metric_entry_t *e = NULL; + if (NULL != adapter->metrics) { + HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e); + } + resp.avg_rtt = NULL != e ? e->value : 0; resp.state = neu_adapter_get_state(adapter); - resp.avg_rtt = adapter->stat.avg_rtt; header->type = NEU_RESP_GET_NODE_STATE; neu_msg_exchange(header); reply(adapter, header, &resp); @@ -680,6 +712,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter) adapter->module->intf_funs->close(adapter->plugin); + if (NULL != adapter->metrics) { + pthread_rwlock_wrlock(&g_node_metrics_mtx_); + HASH_DEL(g_node_metrics_.node_metrics, adapter->metrics); + neu_node_metrics_free(adapter->metrics); + pthread_rwlock_unlock(&g_node_metrics_mtx_); + } + if (adapter->name != NULL) { free(adapter->name); } diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index dfbc6b64..21ea695c 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -52,19 +52,8 @@ struct neu_adapter { neu_event_timer_t *timer; int recv_fd; - // statistics counters - union { - struct { - uint64_t bytes_sent; // number of bytes sent over network - uint64_t bytes_recv; // number of bytes received over network - uint64_t msgs_sent; // number of messages sent - uint64_t msgs_recv; // number of messages received - uint64_t avg_rtt; // average round trip time in milliseconds - uint64_t tag_tot_cnt; // number of tag read including errors - uint64_t tag_err_cnt; // number of tag read errors - }; - uint64_t data[NEU_NODE_STAT_MAX]; - } stat; + // metrics + neu_node_metrics_t *metrics; }; typedef void (*adapter_handler)(neu_adapter_t * adapter, diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 35442b5b..2703e39f 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -99,15 +99,22 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_driver_cache_update(driver->cache, group, t->name, driver->adapter.timestamp, value); } - driver->adapter.stat.tag_tot_cnt += utarray_len(tags); - driver->adapter.stat.tag_err_cnt += utarray_len(tags); + driver->adapter.cb_funs.update_metric(&driver->adapter, + NEU_METRIC_TAG_READS_TOTAL, + utarray_len(tags)); + driver->adapter.cb_funs.update_metric( + &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, + utarray_len(tags)); utarray_free(tags); } } else { neu_driver_cache_update(driver->cache, group, tag, driver->adapter.timestamp, value); - driver->adapter.stat.tag_tot_cnt++; - driver->adapter.stat.tag_err_cnt += (NEU_TYPE_ERROR == value.type); + driver->adapter.cb_funs.update_metric(&driver->adapter, + NEU_METRIC_TAG_READS_TOTAL, 1); + driver->adapter.cb_funs.update_metric(&driver->adapter, + NEU_METRIC_TAG_READ_ERRORS_TOTAL, + NEU_TYPE_ERROR == value.type); } nlog_debug( "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64, diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 029f7970..80d4e04a 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -303,7 +303,12 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr) state.state.running = el->adapter->state; state.state.link = neu_plugin_to_plugin_common(el->adapter->plugin)->link_state; - state.avg_rtt = el->adapter->stat.avg_rtt; + neu_metric_entry_t *e = NULL; + if (NULL != el->adapter->metrics) { + HASH_FIND_STR(el->adapter->metrics->entries, + NEU_METRIC_LAST_RTT_MS, e); + } + state.avg_rtt = NULL != e ? e->value : 0; utarray_push_back(states, &state); }