diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 672cf9fa..0627632e 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -52,6 +52,24 @@ typedef enum { 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[] = { @@ -130,6 +148,7 @@ 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 { @@ -193,6 +212,7 @@ 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) @@ -464,6 +484,10 @@ 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/plugins/monitor/metric_handle.c b/plugins/monitor/metric_handle.c index 3984e020..5b969280 100644 --- a/plugins/monitor/metric_handle.c +++ b/plugins/monitor/metric_handle.c @@ -21,6 +21,7 @@ #include "utils/log.h" #include "metric_handle.h" +#include "monitor.h" #include "restful/handle.h" #include "restful/http.h" @@ -64,6 +65,8 @@ static int response(nng_aio *aio, char *content, enum nng_http_status status) if (content != NULL && strlen(content) > 0) { nng_http_res_copy_data(res, content, strlen(content)); + } else { + nng_http_res_set_data(res, NULL, 0); } nng_http_res_set_status(res, status); @@ -78,13 +81,79 @@ static int response(nng_aio *aio, char *content, enum nng_http_status status) return 0; } +static inline bool parse_metrics_catgory(const char *s, size_t len, + neu_metrics_category_e *cat) +{ + char *domain[] = { + [NEU_METRICS_CATEGORY_GLOBAL] = "global", + [NEU_METRICS_CATEGORY_DRIVER] = "driver", + [NEU_METRICS_CATEGORY_APP] = "app", + [NEU_METRICS_CATEGORY_ALL] = NULL, + }; + + if (NULL == s || NULL == cat) { + return false; + } + + for (int i = 0; i < NEU_METRICS_CATEGORY_ALL; ++i) { + if (strlen(domain[i]) == len && 0 == strncmp(s, domain[i], len)) { + *cat = i; + return true; + } + } + + return false; +} + +static inline void gen_global_metrics(const neu_metrics_t *metrics, + char ** result) +{ + 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); +} + void handle_get_metric(nng_aio *aio) { - char *result = NULL; + char * result = NULL; + neu_plugin_t *plugin = neu_monitor_get_plugin(); + + neu_metrics_category_e cat = NEU_METRICS_CATEGORY_ALL; + size_t cat_param_len = 0; + const char *cat_param = http_get_param(aio, "category", &cat_param_len); + if (NULL != cat_param && + !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; + } + + 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; + } + switch (cat) { + case NEU_METRICS_CATEGORY_GLOBAL: + gen_global_metrics(&plugin->metrics, &result); + break; + case NEU_METRICS_CATEGORY_DRIVER: + case NEU_METRICS_CATEGORY_APP: + case NEU_METRICS_CATEGORY_ALL: + break; + } + pthread_mutex_unlock(&plugin->mutex); - neu_asprintf(&result, METRIC_GLOBAL_TMPL, 0, 0, 0, 0, 0, 0, 0, 0); if (NULL == result) { - nng_aio_finish(aio, NNG_EINTERNAL); + response(aio, NULL, NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR); return; } diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 290bfc54..5a607271 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -21,9 +21,6 @@ #include #include -#include -#include - #include "metric_handle.h" #include "monitor.h" #include "restful/handle.h" @@ -31,10 +28,7 @@ #define neu_plugin_module default_monitor_plugin_module -struct neu_plugin { - neu_plugin_common_t common; - nng_http_server * api_server; -}; +static struct neu_plugin *g_monitor_plugin_; static nng_http_server *server_init() { @@ -99,6 +93,11 @@ static neu_plugin_t *monitor_plugin_open(void) neu_plugin_common_init(&plugin->common); + if (0 != pthread_mutex_init(&plugin->mutex, NULL)) { + free(plugin); + return NULL; + } + plugin->api_server = server_init(); if (NULL == plugin->api_server) { nlog_error("fail to create monitor server"); @@ -114,6 +113,7 @@ static neu_plugin_t *monitor_plugin_open(void) return NULL; } + g_monitor_plugin_ = plugin; return plugin; } @@ -123,17 +123,45 @@ static int monitor_plugin_close(neu_plugin_t *plugin) nng_http_server_stop(plugin->api_server); nng_http_server_release(plugin->api_server); + pthread_mutex_destroy(&plugin->mutex); free(plugin); nlog_info("Success to free plugin: %s", neu_plugin_module.module_name); 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; - (void) plugin; + 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); nlog_info("Initialize plugin: %s", neu_plugin_module.module_name); return rv; @@ -143,7 +171,15 @@ static int monitor_plugin_uninit(neu_plugin_t *plugin) { int rv = 0; - (void) plugin; + 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; + } nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name); return rv; @@ -171,6 +207,12 @@ 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)); @@ -214,3 +256,8 @@ const neu_plugin_module_t default_monitor_plugin_module = { .kind = NEU_PLUGIN_KIND_SYSTEM, .type = NEU_NA_TYPE_APP, }; + +neu_plugin_t *neu_monitor_get_plugin() +{ + return g_monitor_plugin_; +} diff --git a/plugins/monitor/monitor.h b/plugins/monitor/monitor.h index c0b549c4..dfc1d41b 100644 --- a/plugins/monitor/monitor.h +++ b/plugins/monitor/monitor.h @@ -20,8 +20,26 @@ #ifndef NEU_PLUGIN_MONITOR_H #define NEU_PLUGIN_MONITOR_H +#include + +#include +#include + +#include "adapter.h" +#include "event/event.h" #include "plugin.h" +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; +}; + extern const neu_plugin_module_t default_monitor_plugin_module; +neu_plugin_t * neu_monitor_get_plugin(); #endif diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 0d291e95..f3e3de31 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -321,6 +321,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) 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_STAT: @@ -1031,6 +1032,9 @@ 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/core/manager.c b/src/core/manager.c index bea77924..f252f113 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -16,7 +16,9 @@ * 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 @@ -40,6 +42,7 @@ #include "manager.h" #include "manager_internal.h" +#include "utils/time.h" // definition for adapter names #define DEFAULT_DASHBOARD_ADAPTER_NAME DEFAULT_DASHBOARD_PLUGIN_NAME @@ -53,6 +56,7 @@ 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, @@ -72,6 +76,7 @@ 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(); @@ -206,6 +211,24 @@ 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); @@ -604,6 +627,25 @@ 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 bd0250e3..dadf12ce 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -32,6 +32,7 @@ #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 5605b756..029f7970 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -311,3 +311,46 @@ 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 b24ea7fc..dbec7525 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -62,4 +62,6 @@ 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