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

feat(monitor): handling driver metrics generation

Browse Source
This commit is contained in:
eeff
2022-10-24 18:32:03 +08:00
parent 04da8039f8
commit 1f0445712e
5 changed files with 197 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/metrics.h
+8
View File
@@ -108,6 +108,7 @@ typedef struct {
size_t south_running_nodes;
size_t south_disconnected_nodes;
neu_node_metrics_t *node_metrics;
neu_metric_entry_t *registered_metrics;
} neu_metrics_t;
static inline const char *neu_metric_type_str(neu_metric_type_e type)
@@ -119,6 +120,10 @@ static inline const char *neu_metric_type_str(neu_metric_type_e type)
}
}
int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
const char *help, neu_metric_type_e type,
uint64_t init);
static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
{
free(entry);
@@ -166,6 +171,9 @@ static inline void neu_node_metrics_free(neu_node_metrics_t *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);
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type);
void neu_metrics_unregister_entry(const char *name);
typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
include/neuron/utils/utextend.h
+10 -1
View File
@@ -58,4 +58,13 @@ static inline UT_array *utarray_clone(UT_array *array)
#define DL_REVERSE_FOREACH_END(head, tail, el) \
while (((el) = (el)->prev) != (head))
#endif
#ifdef NO_DECLTYPE
#define HASH_LOOP(hh, head, el) \
for ((el) = (head); (el) != NULL; \
(*(char **) (&(el))) = (char *) (el)->hh.next)
#else
#define HASH_LOOP(hh, head, el) \
for ((el) = (head); (el) != NULL; (el) = DECLTYPE(el)((el)->hh.next))
#endif
#endif
plugins/monitor/metric_handle.c
+88
View File
@@ -121,6 +121,71 @@ static inline void gen_global_metrics(const neu_metrics_t *metrics,
metrics->south_disconnected_nodes);
}
static inline void
gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
{
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, node_metrics->entries, e)
{
fprintf(stream,
"# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\"} %" PRIu64 "\n",
e->name, e->help, e->name, neu_metric_type_str(e->type),
e->name, node_metrics->name, e->value);
}
}
static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
FILE *stream)
{
neu_metric_entry_t *e = NULL, *r = NULL;
neu_node_metrics_t *n = NULL;
HASH_LOOP(hh, metrics->registered_metrics, r)
{
bool commented = false;
HASH_LOOP(hh, metrics->node_metrics, n)
{
if (!(type_filter & n->type)) {
continue;
}
if (!commented) {
commented = true;
fprintf(stream, "# HELP %s %s\n# TYPE %s %s\n", r->name,
r->help, r->name, neu_metric_type_str(r->type));
}
HASH_FIND_STR(n->entries, r->name, e);
if (e) {
fprintf(stream, "%s{node=\"%s\"} %" PRIu64 "\n", e->name,
n->name, e->value);
}
}
}
}
struct context {
int filter;
int * status;
FILE * stream;
const char *node;
};
static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx)
{
if (ctx->node[0]) {
neu_node_metrics_t *n = NULL;
HASH_FIND_STR(metrics->node_metrics, ctx->node, n);
if (NULL == n || 0 == (ctx->filter & n->type)) {
*ctx->status = NNG_HTTP_STATUS_NOT_FOUND;
return;
}
gen_single_node_metrics(n, ctx->stream);
} else {
gen_all_node_metrics(metrics, ctx->filter, ctx->stream);
}
}
void handle_get_metric(nng_aio *aio)
{
int status = NNG_HTTP_STATUS_OK;
@@ -140,19 +205,42 @@ void handle_get_metric(nng_aio *aio)
goto end;
}
char node_name[NEU_NODE_NAME_LEN] = { 0 };
ssize_t rv = http_get_param_str(aio, "node", node_name, sizeof(node_name));
if (-1 == rv || rv >= NEU_NODE_NAME_LEN ||
(0 < rv && NEU_METRICS_CATEGORY_GLOBAL == cat)) {
status = NNG_HTTP_STATUS_BAD_REQUEST;
goto end;
}
stream = open_memstream(&result, &len);
if (NULL == stream) {
status = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR;
goto end;
}
struct context ctx = {
.status = &status,
.stream = stream,
.node = node_name,
};
switch (cat) {
case NEU_METRICS_CATEGORY_GLOBAL:
neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream);
break;
case NEU_METRICS_CATEGORY_DRIVER:
ctx.filter = NEU_NA_TYPE_DRIVER;
neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx);
break;
case NEU_METRICS_CATEGORY_APP:
ctx.filter = NEU_NA_TYPE_APP;
neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx);
break;
case NEU_METRICS_CATEGORY_ALL:
ctx.filter = NEU_NA_TYPE_DRIVER | NEU_NA_TYPE_APP;
neu_metrics_visist((neu_metrics_cb_t) gen_global_metrics, stream);
neu_metrics_visist((neu_metrics_cb_t) gen_node_metrics, &ctx);
break;
}
src/adapter/adapter.c
+19 -24
View File
@@ -47,7 +47,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
neu_metric_type_e type, const char *help,
const char *help, neu_metric_type_e type,
uint64_t init);
static int adapter_update_metric(neu_adapter_t *adapter,
const char *metric_name, uint64_t n);
@@ -62,7 +62,7 @@ static const adapter_callbacks_t callback_funs = {
};
#define REGISTER_METRIC(adapter, name, init) \
adapter_register_metric(adapter, name, name##_TYPE, name##_HELP, init);
adapter_register_metric(adapter, name, name##_HELP, name##_TYPE, init);
#define REGISTER_DRIVER_METRICS(adapter) \
REGISTER_METRIC(adapter, NEU_METRIC_LAST_RTT_MS, 9999); \
@@ -100,16 +100,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
adapter->cb_funs.update_metric = callback_funs.update_metric;
adapter->module = info->module;
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;
adapter->metrics->adapter = adapter;
neu_metrics_add_node(adapter);
}
rv = nng_pair1_open(&adapter->sock);
assert(rv == 0);
nng_socket_set_int(adapter->sock, NNG_OPT_RECVBUF, 128);
@@ -119,7 +109,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
switch (info->module->type) {
case NEU_NA_TYPE_DRIVER:
REGISTER_DRIVER_METRICS(adapter);
if (adapter->module->display) {
REGISTER_DRIVER_METRICS(adapter);
}
neu_adapter_driver_init((neu_adapter_driver_t *) adapter);
break;
case NEU_NA_TYPE_APP:
@@ -190,23 +182,26 @@ neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter)
}
static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
neu_metric_type_e type, const char *help,
const char *help, neu_metric_type_e type,
uint64_t init)
{
if (NULL == adapter->metrics) {
adapter->metrics = calloc(1, sizeof(*adapter->metrics));
if (NULL == adapter->metrics) {
return -1;
}
adapter->metrics->type = adapter->module->type;
adapter->metrics->name = adapter->name;
adapter->metrics->adapter = adapter;
neu_metrics_add_node(adapter);
}
if (0 > neu_metric_entries_add(&adapter->metrics->entries, name, help, type,
init)) {
return -1;
}
neu_metric_entry_t *entry = calloc(1, sizeof(*entry));
if (NULL == entry) {
return -1;
}
entry->name = name;
entry->type = type;
entry->help = help;
entry->value = init;
HASH_ADD_STR(adapter->metrics->entries, name, entry);
neu_metrics_register_entry(name, help, type);
return 0;
}
src/adapter/metrics.c
+72 -1
View File
@@ -24,6 +24,7 @@
#include "adapter.h"
#include "adapter_internal.h"
#include "metrics.h"
#include "utils/log.h"
#include "utils/time.h"
static pthread_rwlock_t g_metrics_mtx_ = PTHREAD_RWLOCK_INITIALIZER;
@@ -49,6 +50,45 @@ static bool has_core_dumps()
return found;
}
static inline void metrics_unregister_entry(const char *name)
{
neu_metric_entry_t *e = NULL;
HASH_FIND_STR(g_metrics_.registered_metrics, name, e);
if (0 == --e->value) {
HASH_DEL(g_metrics_.registered_metrics, e);
nlog_info("cai del entry:%s", e->name);
neu_metric_entry_free(e);
}
}
int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
const char *help, neu_metric_type_e type,
uint64_t init)
{
neu_metric_entry_t *entry = NULL;
HASH_FIND_STR(*entries, name, entry);
if (NULL != entry) {
if (entry->type != type || (0 != strcmp(entry->help, help))) {
nlog_error("metric entry %s(%d, %s) conflicts with (%d, %s)", name,
entry->type, entry->help, type, help);
return -1;
}
return 1;
}
entry = calloc(1, sizeof(*entry));
if (NULL == entry) {
return -1;
}
entry->name = name;
entry->type = type;
entry->help = help;
entry->value = init;
HASH_ADD_STR(*entries, name, entry);
return 0;
}
void neu_metrics_init()
{
pthread_rwlock_wrlock(&g_metrics_mtx_);
@@ -69,6 +109,36 @@ void neu_metrics_del_node(const neu_adapter_t *adapter)
{
pthread_rwlock_wrlock(&g_metrics_mtx_);
HASH_DEL(g_metrics_.node_metrics, adapter->metrics);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, adapter->metrics->entries, e)
{
metrics_unregister_entry(e->name);
}
pthread_rwlock_unlock(&g_metrics_mtx_);
}
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type)
{
int rv = 0;
neu_metric_entry_t *e = NULL;
pthread_rwlock_wrlock(&g_metrics_mtx_);
rv = neu_metric_entries_add(&g_metrics_.registered_metrics, name, help,
type, 1);
if (1 == rv) {
HASH_FIND_STR(g_metrics_.registered_metrics, name, e);
++e->value;
rv = 0;
}
pthread_rwlock_unlock(&g_metrics_mtx_);
return rv;
}
void neu_metrics_unregister_entry(const char *name)
{
pthread_rwlock_wrlock(&g_metrics_mtx_);
metrics_unregister_entry(name);
pthread_rwlock_unlock(&g_metrics_mtx_);
}
@@ -88,7 +158,8 @@ void neu_metrics_visist(neu_metrics_cb_t cb, void *data)
g_metrics_.south_disconnected_nodes = 0;
neu_node_metrics_t *n;
for (n = g_metrics_.node_metrics; NULL != n; n = n->hh.next) {
HASH_LOOP(hh, g_metrics_.node_metrics, n)
{
neu_plugin_common_t *common =
neu_plugin_to_plugin_common(n->adapter->plugin);