diff --git a/CMakeLists.txt b/CMakeLists.txt index 51841329..13e94f22 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -103,6 +103,7 @@ set(NEURON_SOURCES src/core/storage.c src/adapter/storage.c src/adapter/adapter.c + src/adapter/metrics.c src/adapter/driver/cache.c src/adapter/driver/driver.c plugins/restful/handle.c diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 2ae098ce..82b9de3f 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -91,7 +91,6 @@ typedef enum neu_reqresp_type { NEU_REQRESP_TRANS_DATA, NEU_REQRESP_NODES_STATE, NEU_REQRESP_NODE_DELETED, - NEU_REQRESP_METRICS, } neu_reqresp_type_e; typedef enum { @@ -153,7 +152,6 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA", [NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODES_STATE", [NEU_REQRESP_NODE_DELETED] = "NEU_REQRESP_NODE_DELETED", - [NEU_REQRESP_METRICS] = "NEU_REQRESP_METRICS", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) @@ -416,10 +414,6 @@ typedef struct { char node[NEU_NODE_NAME_LEN]; } neu_reqresp_node_deleted_t; -typedef struct { - neu_metrics_t *metrics; -} neu_reqresp_metrics_t; - void *neu_msg_gen(neu_reqresp_head_t *header, void *data); inline static void neu_msg_exchange(neu_reqresp_head_t *header) diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h index 976b3641..6eebb69e 100644 --- a/include/neuron/metrics.h +++ b/include/neuron/metrics.h @@ -30,6 +30,9 @@ extern "C" { #include "type.h" #include "utils/uthash.h" +// forward declaration for neu_adapter_t +typedef struct neu_adapter neu_adapter_t; + typedef enum { NEU_METRIC_TYPE_COUNTER, NEU_METRIC_TYPE_GAUAGE, @@ -91,6 +94,7 @@ typedef struct { char * name; // node name neu_metric_entry_t * entries; // node metric entries neu_group_metrics_t *group_metrics; // group metrics + neu_adapter_t * adapter; // UT_hash_handle hh; // ordered by name } neu_node_metrics_t; @@ -106,6 +110,15 @@ typedef struct { neu_node_metrics_t *node_metrics; } neu_metrics_t; +static inline const char *neu_metric_type_str(neu_metric_type_e type) +{ + if (NEU_METRIC_TYPE_COUNTER == type) { + return "counter"; + } else { + return "gauge"; + } +} + static inline void neu_metric_entry_free(neu_metric_entry_t *entry) { free(entry); @@ -150,6 +163,13 @@ static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics) free(node_metrics); } +void neu_metrics_init(); +void neu_metrics_add_node(const neu_adapter_t *adapter); +void neu_metrics_del_node(const neu_adapter_t *adapter); + +typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data); +void neu_metrics_visist(neu_metrics_cb_t cb, void *data); + #ifdef __cplusplus } #endif diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index 8e8b311d..db167670 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/plugins/modbus/modbus_tcp.c @@ -63,6 +63,7 @@ const neu_plugin_module_t neu_plugin_module = { .kind = NEU_PLUGIN_KIND_SYSTEM, .type = NEU_NA_TYPE_DRIVER, .single = false, + .display = true, }; static neu_plugin_t *driver_open(void) diff --git a/plugins/monitor/metric_handle.c b/plugins/monitor/metric_handle.c index b7f5f18f..785d8035 100644 --- a/plugins/monitor/metric_handle.c +++ b/plugins/monitor/metric_handle.c @@ -17,6 +17,10 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ +#include + +#include "define.h" +#include "metrics.h" #include "utils/asprintf.h" #include "utils/log.h" @@ -108,19 +112,21 @@ static inline bool parse_metrics_catgory(const char *s, size_t len, } static inline void gen_global_metrics(const neu_metrics_t *metrics, - char ** result) + FILE * stream) { - neu_asprintf(result, METRIC_GLOBAL_TMPL, metrics->core_dumped, - metrics->uptime_seconds, metrics->north_nodes, - metrics->north_running_nodes, - metrics->north_disconnected_nodes, metrics->south_nodes, - metrics->south_running_nodes, - metrics->south_disconnected_nodes); + fprintf(stream, METRIC_GLOBAL_TMPL, metrics->core_dumped, + metrics->uptime_seconds, metrics->north_nodes, + metrics->north_running_nodes, metrics->north_disconnected_nodes, + metrics->south_nodes, metrics->south_running_nodes, + metrics->south_disconnected_nodes); } void handle_get_metric(nng_aio *aio) { + int status = NNG_HTTP_STATUS_OK; char * result = NULL; + size_t len = 0; + FILE * stream = NULL; neu_plugin_t *plugin = neu_monitor_get_plugin(); neu_metrics_category_e cat = NEU_METRICS_CATEGORY_ALL; @@ -130,35 +136,30 @@ void handle_get_metric(nng_aio *aio) !parse_metrics_catgory(cat_param, cat_param_len, &cat)) { plog_info(plugin, "invalid metrics category: %.*s", (int) cat_param_len, cat_param); - response(aio, NULL, NNG_HTTP_STATUS_BAD_REQUEST); - return; + status = NNG_HTTP_STATUS_BAD_REQUEST; + goto end; } - pthread_mutex_lock(&plugin->mutex); - if (plugin->metrics_updating) { - // a metrics request to core is in progress - // in the assumption that this case is rare, we just finish error - pthread_mutex_unlock(&plugin->mutex); - nng_aio_finish(aio, NNG_EBUSY); - response(aio, NULL, NNG_HTTP_STATUS_SERVICE_UNAVAILABLE); - return; + stream = open_memstream(&result, &len); + if (NULL == stream) { + status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR; + goto end; } + switch (cat) { case NEU_METRICS_CATEGORY_GLOBAL: - gen_global_metrics(&plugin->metrics, &result); + neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream); break; case NEU_METRICS_CATEGORY_DRIVER: case NEU_METRICS_CATEGORY_APP: case NEU_METRICS_CATEGORY_ALL: break; } - pthread_mutex_unlock(&plugin->mutex); - if (NULL == result) { - response(aio, NULL, NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR); - return; +end: + if (NULL != stream) { + fclose(stream); } - - response(aio, result, NNG_HTTP_STATUS_OK); + response(aio, NNG_HTTP_STATUS_OK == status ? result : NULL, status); free(result); } diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 5a607271..7a7e4b34 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -130,38 +130,10 @@ static int monitor_plugin_close(neu_plugin_t *plugin) return rv; } -static inline int send_metrics_req(neu_plugin_t *plugin) -{ - neu_reqresp_head_t header = { - .type = NEU_REQRESP_METRICS, - }; - neu_reqresp_metrics_t cmd = { .metrics = &plugin->metrics }; - - pthread_mutex_lock(&plugin->mutex); - plugin->metrics_updating = true; - pthread_mutex_unlock(&plugin->mutex); - - return neu_plugin_op(plugin, header, &cmd); -} - -static int monitor_timer_cb(void *data) -{ - neu_plugin_t *plugin = data; - return send_metrics_req(plugin); -} - static int monitor_plugin_init(neu_plugin_t *plugin) { int rv = 0; - - plugin->events = neu_event_new(); - neu_event_timer_param_t param = { - .second = 2, - .millisecond = 0, - .cb = monitor_timer_cb, - .usr_data = plugin, - }; - plugin->timer = neu_event_add_timer(plugin->events, param); + (void) plugin; nlog_info("Initialize plugin: %s", neu_plugin_module.module_name); return rv; @@ -207,12 +179,6 @@ static int monitor_plugin_request(neu_plugin_t * plugin, nlog_warn("recv error code: %d", resp_err->error); break; } - case NEU_REQRESP_METRICS: { - pthread_mutex_lock(&plugin->mutex); - plugin->metrics_updating = false; - pthread_mutex_unlock(&plugin->mutex); - break; - } default: nlog_fatal("recv unhandle msg: %s", neu_reqresp_type_string(header->type)); diff --git a/plugins/monitor/monitor.h b/plugins/monitor/monitor.h index dfc1d41b..8ec5aca9 100644 --- a/plugins/monitor/monitor.h +++ b/plugins/monitor/monitor.h @@ -33,7 +33,6 @@ struct neu_plugin { neu_plugin_common_t common; pthread_mutex_t mutex; bool metrics_updating; - neu_metrics_t metrics; neu_events_t * events; neu_event_timer_t * timer; nng_http_server * api_server; diff --git a/plugins/restful/normal_handle.c b/plugins/restful/normal_handle.c index a429214a..311b13cb 100644 --- a/plugins/restful/normal_handle.c +++ b/plugins/restful/normal_handle.c @@ -28,7 +28,6 @@ #include #include -#include "adapter/adapter_internal.h" #include "persist/persist.h" #include "user.h" #include "utils/asprintf.h" diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 2da50e3a..1f4d546e 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -55,9 +55,6 @@ 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, @@ -103,11 +100,14 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info) adapter->cb_funs.update_metric = callback_funs.update_metric; adapter->module = info->module; - adapter->metrics = calloc(1, sizeof(*adapter->metrics)); + if (adapter->module->display) { + 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); + adapter->metrics->type = info->module->type; + adapter->metrics->name = adapter->name; + adapter->metrics->adapter = adapter; + neu_metrics_add_node(adapter); } rv = nng_pair1_open(&adapter->sock); @@ -333,22 +333,17 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) nng_recvmsg(adapter->sock, &msg, 0); header = (neu_reqresp_head_t *) nng_msg_body(msg); - switch (header->type) { - case NEU_REQRESP_NODES_STATE: - case NEU_REQRESP_METRICS: + if (header->type == NEU_REQRESP_NODES_STATE) { nlog_debug("adapter(%s) recv msg from: %s, type: %s", adapter->name, header->sender, neu_reqresp_type_string(header->type)); - break; - default: + } else { nlog_info("adapter(%s) recv msg from: %s, type: %s", adapter->name, header->sender, neu_reqresp_type_string(header->type)); - break; } switch (header->type) { case NEU_RESP_GET_DRIVER_GROUP: case NEU_REQRESP_NODE_DELETED: - case NEU_REQRESP_METRICS: case NEU_RESP_GET_SUB_DRIVER_TAGS: case NEU_REQ_UPDATE_LICENSE: case NEU_RESP_GET_NODE_STATE: @@ -717,10 +712,8 @@ 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_metrics_del_node(adapter); neu_node_metrics_free(adapter->metrics); - pthread_rwlock_unlock(&g_node_metrics_mtx_); } if (adapter->name != NULL) { @@ -1046,9 +1039,6 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_REQRESP_NODE_DELETED: data_size = sizeof(neu_reqresp_node_deleted_t); break; - case NEU_REQRESP_METRICS: - data_size = sizeof(neu_reqresp_metrics_t); - break; case NEU_RESP_GET_DRIVER_GROUP: data_size = sizeof(neu_resp_get_driver_group_t); break; diff --git a/src/adapter/metrics.c b/src/adapter/metrics.c new file mode 100644 index 00000000..cdedc13c --- /dev/null +++ b/src/adapter/metrics.c @@ -0,0 +1,116 @@ +/** + * 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. + **/ + +#include +#include +#include + +#include "adapter.h" +#include "adapter_internal.h" +#include "metrics.h" +#include "utils/time.h" + +static pthread_rwlock_t g_metrics_mtx_ = PTHREAD_RWLOCK_INITIALIZER; +static neu_metrics_t g_metrics_; +static uint64_t g_start_ts_; + +static bool has_core_dumps() +{ + DIR *dp = opendir("core"); + if (NULL == dp) { + return false; + } + + bool found = false; + for (struct dirent *de = NULL; NULL != (de = readdir(dp));) { + if (0 == strncmp("core", de->d_name, 4)) { + found = true; + break; + } + } + + closedir(dp); + return found; +} + +void neu_metrics_init() +{ + pthread_rwlock_wrlock(&g_metrics_mtx_); + if (0 == g_start_ts_) { + g_start_ts_ = neu_time_ms(); + } + pthread_rwlock_unlock(&g_metrics_mtx_); +} + +void neu_metrics_add_node(const neu_adapter_t *adapter) +{ + pthread_rwlock_wrlock(&g_metrics_mtx_); + HASH_ADD_STR(g_metrics_.node_metrics, name, adapter->metrics); + pthread_rwlock_unlock(&g_metrics_mtx_); +} + +void neu_metrics_del_node(const neu_adapter_t *adapter) +{ + pthread_rwlock_wrlock(&g_metrics_mtx_); + HASH_DEL(g_metrics_.node_metrics, adapter->metrics); + pthread_rwlock_unlock(&g_metrics_mtx_); +} + +void neu_metrics_visist(neu_metrics_cb_t cb, void *data) +{ + bool core_dumped = has_core_dumps(); + uint64_t uptime_seconds = (neu_time_ms() - g_start_ts_) / 1000; + pthread_rwlock_rdlock(&g_metrics_mtx_); + g_metrics_.core_dumped = core_dumped; + g_metrics_.uptime_seconds = uptime_seconds; + + g_metrics_.north_nodes = 0; + g_metrics_.north_running_nodes = 0; + g_metrics_.north_disconnected_nodes = 0; + g_metrics_.south_nodes = 0; + g_metrics_.south_running_nodes = 0; + g_metrics_.south_disconnected_nodes = 0; + + neu_node_metrics_t *n; + for (n = g_metrics_.node_metrics; NULL != n; n = n->hh.next) { + neu_plugin_common_t *common = + neu_plugin_to_plugin_common(n->adapter->plugin); + + if (NEU_NA_TYPE_DRIVER == n->adapter->module->type) { + ++g_metrics_.south_nodes; + if (NEU_NODE_RUNNING_STATE_RUNNING == n->adapter->state) { + ++g_metrics_.south_running_nodes; + } + if (NEU_NODE_LINK_STATE_DISCONNECTED == common->link_state) { + ++g_metrics_.south_disconnected_nodes; + } + } else if (NEU_NA_TYPE_APP == n->adapter->module->type) { + ++g_metrics_.north_nodes; + if (NEU_NODE_RUNNING_STATE_RUNNING == n->adapter->state) { + ++g_metrics_.north_running_nodes; + } + if (NEU_NODE_LINK_STATE_DISCONNECTED == common->link_state) { + ++g_metrics_.north_disconnected_nodes; + } + } + } + + cb(&g_metrics_, data); + pthread_rwlock_unlock(&g_metrics_mtx_); +} diff --git a/src/core/manager.c b/src/core/manager.c index c7e1b686..900c911d 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -16,9 +16,7 @@ * along with this program; if not, write to the Free Software Foundation, * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#include #include -#include #include #include @@ -42,7 +40,6 @@ #include "manager.h" #include "manager_internal.h" -#include "utils/time.h" // definition for adapter names #define DEFAULT_DASHBOARD_ADAPTER_NAME DEFAULT_DASHBOARD_PLUGIN_NAME @@ -56,7 +53,6 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, nng_pipe pipe); inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *ndoe); -static bool has_core_dumps(); static void start_static_adapter(neu_manager_t *manager, const char *name); static int report_nodes_state(void *usr_data); static void start_single_adapter(neu_manager_t *manager, const char *name, @@ -76,7 +72,6 @@ neu_manager_t *neu_manager_create() .cb = report_nodes_state, }; - manager->start_ts = neu_time_ms(); manager->events = neu_event_new(); manager->plugin_manager = neu_plugin_manager_create(); manager->node_manager = neu_node_manager_create(); @@ -94,6 +89,7 @@ neu_manager_t *neu_manager_create() nng_socket_get_int(manager->socket, NNG_OPT_RECVFD, ¶m.fd); manager->loop = neu_event_add_io(manager->events, param); + neu_metrics_init(); start_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME); start_static_adapter(manager, DEFAULT_MONITOR_PLUGIN_NAME); @@ -190,16 +186,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) header = (neu_reqresp_head_t *) nng_msg_body(msg); - switch (header->type) { - case NEU_REQRESP_METRICS: - nlog_debug("manager recv msg from: %s to %s, type: %s", header->sender, - header->receiver, neu_reqresp_type_string(header->type)); - break; - default: - nlog_info("manager recv msg from: %s to %s, type: %s", header->sender, - header->receiver, neu_reqresp_type_string(header->type)); - break; - } + nlog_info("manager recv msg from: %s to %s, type: %s", header->sender, + header->receiver, neu_reqresp_type_string(header->type)); switch (header->type) { case NEU_REQRESP_TRANS_DATA: { neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1]; @@ -216,24 +204,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } - case NEU_REQRESP_METRICS: { - neu_reqresp_metrics_t *cmd = (neu_reqresp_metrics_t *) &header[1]; - if (NULL == cmd->metrics || - 0 != - neu_node_manager_get_metrics(manager->node_manager, - cmd->metrics)) { - break; - } - cmd->metrics->core_dumped = has_core_dumps(); - cmd->metrics->uptime_seconds = - (neu_time_ms() - manager->start_ts) / 1000; - - header->type = NEU_REQRESP_METRICS; - strcpy(header->receiver, header->sender); - reply(manager, header, cmd); - - break; - } case NEU_REQ_UPDATE_LICENSE: { UT_array *pipes = neu_node_manager_get_pipes(manager->node_manager, NEU_NA_TYPE_DRIVER); @@ -630,25 +600,6 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, } } -static bool has_core_dumps() -{ - DIR *dp = opendir("core"); - if (NULL == dp) { - return false; - } - - bool found = false; - for (struct dirent *de = NULL; NULL != (de = readdir(dp));) { - if (0 == strncmp("core", de->d_name, 4)) { - found = true; - break; - } - } - - closedir(dp); - return found; -} - static void start_static_adapter(neu_manager_t *manager, const char *name) { neu_adapter_t * adapter = NULL; diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 72d258cf..29593b74 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -32,7 +32,6 @@ #include "subscribe.h" typedef struct neu_manager { - uint64_t start_ts; nng_socket socket; neu_events_t * events; neu_event_io_t * loop; diff --git a/src/core/node_manager.c b/src/core/node_manager.c index e7dac47b..86385932 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -316,46 +316,3 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr) return states; } - -int neu_node_manager_get_metrics(neu_node_manager_t *mgr, - neu_metrics_t * metrics_p) -{ - node_entity_t *el = NULL, *tmp = NULL; - - metrics_p->north_nodes = 0; - metrics_p->north_running_nodes = 0; - metrics_p->north_disconnected_nodes = 0; - metrics_p->south_nodes = 0; - metrics_p->south_running_nodes = 0; - metrics_p->south_disconnected_nodes = 0; - - HASH_ITER(hh, mgr->nodes, el, tmp) - { - if (el->is_static || !el->display) { - continue; - } - - neu_plugin_common_t *common = - neu_plugin_to_plugin_common(el->adapter->plugin); - - if (NEU_NA_TYPE_DRIVER == el->adapter->module->type) { - ++metrics_p->south_nodes; - if (NEU_NODE_RUNNING_STATE_RUNNING == el->adapter->state) { - ++metrics_p->south_running_nodes; - } - if (NEU_NODE_LINK_STATE_DISCONNECTED == common->link_state) { - ++metrics_p->south_disconnected_nodes; - } - } else if (NEU_NA_TYPE_APP == el->adapter->module->type) { - ++metrics_p->north_nodes; - if (NEU_NODE_RUNNING_STATE_RUNNING == el->adapter->state) { - ++metrics_p->north_running_nodes; - } - if (NEU_NODE_LINK_STATE_DISCONNECTED == common->link_state) { - ++metrics_p->north_disconnected_nodes; - } - } - } - - return 0; -} diff --git a/src/core/node_manager.h b/src/core/node_manager.h index dbec7525..b24ea7fc 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -62,6 +62,4 @@ nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name); // neu_nodes_state_t array UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr); -int neu_node_manager_get_metrics(neu_node_manager_t *mgr, - neu_metrics_t * metrics_p); #endif \ No newline at end of file