GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 22:59:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #855 from eeff/eeff/refactor

Browse Source 
refactor: monitor global metrics generation
This commit is contained in:
fengzeroz
2022-10-24 13:46:49 +08:00
committed by GitHub
parent 801ab8659e 04da8039f8
commit 8606d82b76
14 changed files with 177 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -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
include/neuron/adapter.h
-6
View File
@@ -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)
include/neuron/metrics.h
+20
View File
@@ -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
plugins/modbus/modbus_tcp.c
+1
View File
@@ -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)
plugins/monitor/metric_handle.c
+25 -24
View File
@@ -17,6 +17,10 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <stdio.h>
#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);
}
plugins/monitor/monitor.c
+1 -35
View File
@@ -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));
plugins/monitor/monitor.h
-1
View File
@@ -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;
plugins/restful/normal_handle.c
-1
View File
@@ -28,7 +28,6 @@
#include <time.h>
#include <unistd.h>
#include "adapter/adapter_internal.h"
#include "persist/persist.h"
#include "user.h"
#include "utils/asprintf.h"
src/adapter/adapter.c
+10 -20
View File
@@ -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;
src/adapter/metrics.c
+116
View File
@@ -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 <dirent.h>
#include <pthread.h>
#include <unistd.h>
#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_);
}
src/core/manager.c
+3 -52
View File
@@ -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 <dirent.h>
#include <stdlib.h>
#include <sys/types.h>
#include <unistd.h>
#include <nng/nng.h>
@@ -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, &param.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;
src/core/manager_internal.h
-1
View File
@@ -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;
src/core/node_manager.c
-43
View File
@@ -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;
}
src/core/node_manager.h
-2
View File
@@ -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