GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 21:41:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #834 from eeff/eeff/dev

Browse Source 
feat(monitor): handling global metrics generation
This commit is contained in:
fengzeroz
2022-10-13 11:06:53 +08:00
committed by GitHub
parent 9f6b2657fa 68ee92dbb0
commit 905ebeb826
9 changed files with 262 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+24
View File
@@ -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)
plugins/monitor/metric_handle.c
+72 -3
View File
@@ -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;
}
plugins/monitor/monitor.c
+56 -9
View File
@@ -21,9 +21,6 @@
#include <stdlib.h>
#include <string.h>
#include <nng/nng.h>
#include <nng/supplemental/http/http.h>
#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_;
}
plugins/monitor/monitor.h
+18
View File
@@ -20,8 +20,26 @@
#ifndef NEU_PLUGIN_MONITOR_H
#define NEU_PLUGIN_MONITOR_H
#include <pthread.h>
#include <nng/nng.h>
#include <nng/supplemental/http/http.h>
#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
src/adapter/adapter.c
+4
View File
@@ -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;
src/core/manager.c
+42
View File
@@ -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 <dirent.h>
#include <stdlib.h>
#include <sys/types.h>
#include <unistd.h>
#include <nng/nng.h>
@@ -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;
src/core/manager_internal.h
+1
View File
@@ -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;
src/core/node_manager.c
+43
View File
@@ -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;
}
src/core/node_manager.h
+2
View File
@@ -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