GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1963 from eeff/main

Browse Source 
fix: reset metrics when node stopped
This commit is contained in:
fengzeroz
2024-04-29 09:43:44 +08:00
committed by GitHub
parent 1e30eea3f6 0ac0e0aa86
commit d724c52cb2
5 changed files with 134 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/metrics.h
+41 -14
View File
@@ -41,16 +41,22 @@ typedef enum {
NEU_METRIC_TYPE_GAUAGE,
NEU_METRIC_TYPE_COUNTER_SET,
NEU_METRIC_TYPE_ROLLING_COUNTER,
NEU_METRIC_TYPE_FLAG_NO_RESET = 0x80,
} neu_metric_type_e;
#define NEU_METRIC_TYPE_MASK 0x0F
// node running state
#define NEU_METRIC_RUNNING_STATE "running_state"
#define NEU_METRIC_RUNNING_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
#define NEU_METRIC_RUNNING_STATE_TYPE \
(NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
#define NEU_METRIC_RUNNING_STATE_HELP "Node running state"
// node link state
#define NEU_METRIC_LINK_STATE "link_state"
#define NEU_METRIC_LINK_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
#define NEU_METRIC_LINK_STATE_TYPE \
(NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
#define NEU_METRIC_LINK_STATE_HELP "Node link state"
// last round trip time in millisends
@@ -86,13 +92,15 @@ typedef enum {
// maintained by neuron core
// number of tags in group
#define NEU_METRIC_TAGS_TOTAL "tags_total"
#define NEU_METRIC_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
#define NEU_METRIC_TAGS_TOTAL_TYPE \
(NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
#define NEU_METRIC_TAGS_TOTAL_HELP "Total number of tags in the node"
// maintained by neuron core
// number of tags in group
#define NEU_METRIC_GROUP_TAGS_TOTAL "group_tags_total"
#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE \
(NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
#define NEU_METRIC_GROUP_TAGS_TOTAL_HELP "Total number of tags in the group"
// number of messages sent in last group timer
@@ -301,9 +309,24 @@ 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);
static inline bool neu_metric_type_is_counter(neu_metric_type_e type)
{
return NEU_METRIC_TYPE_COUNTER == (type & NEU_METRIC_TYPE_MASK);
}
static inline bool neu_metric_type_is_rolling_counter(neu_metric_type_e type)
{
return NEU_METRIC_TYPE_ROLLING_COUNTER == (type & NEU_METRIC_TYPE_MASK);
}
static inline bool neu_metric_type_no_reset(neu_metric_type_e type)
{
return NEU_METRIC_TYPE_FLAG_NO_RESET & type;
}
static inline const char *neu_metric_type_str(neu_metric_type_e type)
{
if (NEU_METRIC_TYPE_COUNTER == type) {
if (neu_metric_type_is_counter(type)) {
return "counter";
} else {
return "gauge";
@@ -316,7 +339,7 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
if (neu_metric_type_is_rolling_counter(entry->type)) {
neu_rolling_counter_free(entry->rcnt);
}
free(entry);
@@ -445,9 +468,9 @@ static inline int neu_node_metrics_update(neu_node_metrics_t *node_metrics,
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
if (neu_metric_type_is_counter(entry->type)) {
entry->value += n;
} else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
} else if (neu_metric_type_is_rolling_counter(entry->type)) {
entry->value =
neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
} else {
@@ -465,9 +488,11 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
pthread_mutex_lock(&node_metrics->lock);
HASH_LOOP(hh, node_metrics->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
if (!neu_metric_type_no_reset(entry->type)) {
entry->value = 0;
if (neu_metric_type_is_rolling_counter(entry->type)) {
neu_rolling_counter_reset(entry->rcnt);
}
}
}
@@ -476,9 +501,11 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
{
HASH_LOOP(hh, g->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
if (!neu_metric_type_no_reset(entry->type)) {
entry->value = 0;
if (neu_metric_type_is_rolling_counter(entry->type)) {
neu_rolling_counter_reset(entry->rcnt);
}
}
}
}
plugins/restful/metric_handle.c
+4 -4
View File
@@ -181,7 +181,7 @@ static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
pthread_mutex_lock(&node_metrics->lock);
HASH_LOOP(hh, node_metrics->entries, e)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
if (neu_metric_type_is_rolling_counter(e->type)) {
// force clean stale value
e->value = neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
}
@@ -196,7 +196,7 @@ static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
{
HASH_LOOP(hh, g->entries, e)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
if (neu_metric_type_is_rolling_counter(e->type)) {
// force clean stale value
e->value =
neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
@@ -278,7 +278,7 @@ static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
pthread_mutex_lock(&n->lock);
HASH_FIND_STR(n->entries, r->name, e);
if (e) {
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
if (neu_metric_type_is_rolling_counter(e->type)) {
// force clean stale value
e->value =
neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
@@ -294,7 +294,7 @@ static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
{
HASH_FIND_STR(g->entries, r->name, e);
if (e) {
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
if (neu_metric_type_is_rolling_counter(e->type)) {
// force clean stale value
e->value = neu_rolling_counter_inc(e->rcnt,
global_timestamp, 0);
src/adapter/adapter.c
+9 -8
View File
@@ -426,13 +426,6 @@ static int adapter_update_metric(neu_adapter_t *adapter,
return neu_node_metrics_update(adapter->metrics, group, metric_name, n);
}
static inline void adapter_reset_metrics(neu_adapter_t *adapter)
{
if (NULL != adapter->metrics) {
neu_node_metrics_reset(adapter->metrics);
}
}
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data)
{
@@ -1389,6 +1382,10 @@ int neu_adapter_start(neu_adapter_t *adapter)
if (error == NEU_ERR_SUCCESS) {
adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
adapter_storage_state(adapter->name, adapter->state);
if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
neu_adapter_driver_start_group_timer(
(neu_adapter_driver_t *) adapter);
}
}
return error;
@@ -1427,7 +1424,11 @@ int neu_adapter_stop(neu_adapter_t *adapter)
if (error == NEU_ERR_SUCCESS) {
adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
adapter_storage_state(adapter->name, adapter->state);
adapter_reset_metrics(adapter);
if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
neu_adapter_driver_stop_group_timer(
(neu_adapter_driver_t *) adapter);
}
neu_adapter_reset_metrics(adapter);
}
return error;
src/adapter/adapter_internal.h
+7
View File
@@ -96,6 +96,13 @@ int neu_adapter_set_setting(neu_adapter_t *adapter, const char *config);
int neu_adapter_get_setting(neu_adapter_t *adapter, char **config);
neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter);
static inline void neu_adapter_reset_metrics(neu_adapter_t *adapter)
{
if (NULL != adapter->metrics) {
neu_node_metrics_reset(adapter->metrics);
}
}
int neu_adapter_register_group_metric(neu_adapter_t *adapter,
const char *group_name, const char *name,
const char *help, neu_metric_type_e type,
src/adapter/driver/driver.c
+73 -85
View File
@@ -102,8 +102,11 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta);
static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
static void store_write_tag(group_t *group, to_be_write_tag_t *tag);
static group_t * find_group(neu_adapter_driver_t *driver, const char *name);
static void store_write_tag(group_t *group, to_be_write_tag_t *tag);
static inline void start_group_timer(neu_adapter_driver_t *driver,
group_t * grp);
static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp);
static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
{
@@ -324,9 +327,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
HASH_DEL(driver->groups, el);
neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group));
neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
neu_event_del_timer(driver->driver_events, el->read);
neu_event_del_timer(driver->driver_events, el->write);
stop_group_timer(driver, el);
if (el->grp.group_free != NULL) {
el->grp.group_free(&el->grp);
}
@@ -349,34 +350,67 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
return 0;
}
static inline void start_group_timer(neu_adapter_driver_t *driver, group_t *grp)
{
uint32_t interval = neu_group_get_interval(grp->group);
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = (void *) grp,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
grp->read = neu_event_add_timer(driver->driver_events, param);
struct timespec t1 = {
.tv_sec = 0,
.tv_nsec = 1000 * 1000 * 20,
};
struct timespec t2 = { 0 };
nanosleep(&t1, &t2);
param.type = NEU_EVENT_TIMER_NOBLOCK;
param.cb = report_callback;
grp->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
param.second = 0;
param.millisecond = 3;
param.cb = write_callback;
grp->write = neu_event_add_timer(driver->driver_events, param);
}
void neu_adapter_driver_start_group_timer(neu_adapter_driver_t *driver)
{
group_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, driver->groups, el, tmp)
{
uint32_t interval = neu_group_get_interval(el->group);
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = el,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
start_group_timer(driver, el);
neu_adapter_update_group_metric(
&driver->adapter, neu_group_get_name(el->group),
NEU_METRIC_GROUP_TAGS_TOTAL, neu_group_tag_size(el->group));
}
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
el->read = neu_event_add_timer(driver->driver_events, param);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAGS_TOTAL, driver->tag_cnt, NULL);
}
struct timespec t1 = {
.tv_sec = 0,
.tv_nsec = 1000 * 1000 * 20,
};
struct timespec t2 = { 0 };
nanosleep(&t1, &t2);
param.type = NEU_EVENT_TIMER_NOBLOCK;
param.cb = report_callback;
el->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp)
{
if (grp->report) {
neu_adapter_del_timer((neu_adapter_t *) driver, grp->report);
grp->report = NULL;
}
if (grp->read) {
neu_event_del_timer(driver->driver_events, grp->read);
grp->read = NULL;
}
if (grp->write) {
neu_event_del_timer(driver->driver_events, grp->write);
grp->write = NULL;
}
}
@@ -384,13 +418,7 @@ void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver)
{
group_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, driver->groups, el, tmp)
{
neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
el->report = NULL;
neu_event_del_timer(driver->driver_events, el->read);
el->read = NULL;
}
HASH_ITER(hh, driver->groups, el, tmp) { stop_group_timer(driver, el); }
}
void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
@@ -890,13 +918,6 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
if (find == NULL) {
find = calloc(1, sizeof(group_t));
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = (void *) find,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
pthread_mutex_init(&find->wt_mtx, NULL);
pthread_mutex_init(&find->apps_mtx, NULL);
@@ -910,25 +931,9 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
neu_group_split_static_tags(find->group, &find->static_tags,
&find->grp.tags);
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
find->read = neu_event_add_timer(driver->driver_events, param);
struct timespec t1 = {
.tv_sec = 0,
.tv_nsec = 1000 * 1000 * 20,
};
struct timespec t2 = { 0 };
nanosleep(&t1, &t2);
param.type = NEU_EVENT_TIMER_NOBLOCK;
param.cb = report_callback;
find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
param.second = 0;
param.millisecond = 3;
param.cb = write_callback;
find->write = neu_event_add_timer(driver->driver_events, param);
if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
start_group_timer(driver, find);
}
REGISTER_GROUP_METRIC(&driver->adapter, find->name,
NEU_METRIC_GROUP_TAGS_TOTAL,
@@ -975,8 +980,9 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
}
// stop the timer first to avoid race condition
neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
neu_event_del_timer(driver->driver_events, find->read);
if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
stop_group_timer(driver, find);
}
// a diminutive value should keep the interval untouched
if (interval < NEU_GROUP_INTERVAL_LIMIT) {
@@ -1005,31 +1011,12 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
}
}
neu_event_timer_param_t param = {
.second = interval / 1000,
.millisecond = interval % 1000,
.usr_data = (void *) find,
.type = NEU_EVENT_TIMER_NOBLOCK,
};
neu_group_set_interval(find->group, interval);
// restore the timers
param.type = driver->adapter.module->timer_type;
param.cb = read_callback;
find->read = neu_event_add_timer(driver->driver_events, param);
struct timespec t1 = {
.tv_sec = 0,
.tv_nsec = 1000 * 1000 * 20,
};
struct timespec t2 = { 0 };
nanosleep(&t1, &t2);
param.type = NEU_EVENT_TIMER_NOBLOCK;
param.cb = report_callback;
find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
start_group_timer(driver, find);
}
return ret;
}
@@ -1045,9 +1032,10 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(find->group));
neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
neu_event_del_timer(driver->driver_events, find->read);
neu_event_del_timer(driver->driver_events, find->write);
if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
stop_group_timer(driver, find);
}
if (find->grp.group_free != NULL) {
find->grp.group_free(&find->grp);
}