diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h index 12765c37..50e05664 100644 --- a/include/neuron/metrics.h +++ b/include/neuron/metrics.h @@ -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); + } } } } diff --git a/plugins/restful/metric_handle.c b/plugins/restful/metric_handle.c index d18884b0..daea80a9 100644 --- a/plugins/restful/metric_handle.c +++ b/plugins/restful/metric_handle.c @@ -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); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index f5531c69..df067bf3 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 156f4349..7afa19c8 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -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, diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index bd0adf4f..f294a9f3 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -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); }