diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index cb9bb94b..0627632e 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -52,6 +52,13 @@ 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; 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