diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h index 53284396..0bfd993d 100644 --- a/include/neuron/metrics.h +++ b/include/neuron/metrics.h @@ -28,12 +28,14 @@ extern "C" { #include "define.h" #include "type.h" +#include "utils/rolling_counter.h" #include "utils/uthash.h" typedef enum { NEU_METRIC_TYPE_COUNTER, NEU_METRIC_TYPE_GAUAGE, NEU_METRIC_TYPE_COUNTER_SET, + NEU_METRIC_TYPE_ROLLING_COUNTER, } neu_metric_type_e; // node running state @@ -131,6 +133,96 @@ typedef enum { #define NEU_METRIC_RECV_MSGS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER #define NEU_METRIC_RECV_MSGS_TOTAL_HELP "Total number of messages received" +// number of trans data message within the last 5 seconds +#define NEU_METRIC_TRANS_DATA_5S "last_5s_trans_data_msgs" +#define NEU_METRIC_TRANS_DATA_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_TRANS_DATA_5S_HELP \ + "Number of internal trans data message within the last 5 seconds" + +// number of trans data message within the last 30 seconds +#define NEU_METRIC_TRANS_DATA_30S "last_30s_trans_data_msgs" +#define NEU_METRIC_TRANS_DATA_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_TRANS_DATA_30S_HELP \ + "Number of internal trans data message within the last 30 seconds" + +// number of trans data message within the last 60 seconds +#define NEU_METRIC_TRANS_DATA_60S "last_60s_trans_data_msgs" +#define NEU_METRIC_TRANS_DATA_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_TRANS_DATA_60S_HELP \ + "Number of internal trans data message within the last 60 seconds" + +// number of bytes sent within the last 5 seconds +#define NEU_METRIC_SEND_BYTES_5S "last_5s_send_bytes" +#define NEU_METRIC_SEND_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_SEND_BYTES_5S_HELP \ + "Number of bytes sent within the last 5 seconds" + +// number of bytes sent within the last 30 seconds +#define NEU_METRIC_SEND_BYTES_30S "last_30s_send_bytes" +#define NEU_METRIC_SEND_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_SEND_BYTES_30S_HELP \ + "Number of bytes sent within the last 30 seconds" + +// number of bytes sent within the last 60 seconds +#define NEU_METRIC_SEND_BYTES_60S "last_60s_send_bytes" +#define NEU_METRIC_SEND_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_SEND_BYTES_60S_HELP \ + "Number of bytes sent within the last 60 seconds" + +// number of bytes received within the last 5 seconds +#define NEU_METRIC_RECV_BYTES_5S "last_5s_recv_bytes" +#define NEU_METRIC_RECV_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_BYTES_5S_HELP \ + "Number of bytes received within the last 5 seconds" + +// number of bytes received within the last 30 seconds +#define NEU_METRIC_RECV_BYTES_30S "last_30s_recv_bytes" +#define NEU_METRIC_RECV_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_BYTES_30S_HELP \ + "Number of bytes received within the last 30 seconds" + +// number of bytes received within the last 60 seconds +#define NEU_METRIC_RECV_BYTES_60S "last_60s_recv_bytes" +#define NEU_METRIC_RECV_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_BYTES_60S_HELP \ + "Number of bytes received within the last 60 seconds" + +// number of messages received within the last 5 seconds +#define NEU_METRIC_RECV_MSGS_5S "last_5s_recv_msgs" +#define NEU_METRIC_RECV_MSGS_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_MSGS_5S_HELP \ + "Number of messages received within the last 5 seconds" + +// number of messages received within the last 30 seconds +#define NEU_METRIC_RECV_MSGS_30S "last_30s_recv_msgs" +#define NEU_METRIC_RECV_MSGS_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_MSGS_30S_HELP \ + "Number of messages received within the last 30 seconds" + +// number of messages received within the last 60 seconds +#define NEU_METRIC_RECV_MSGS_60S "last_60s_recv_msgs" +#define NEU_METRIC_RECV_MSGS_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_RECV_MSGS_60S_HELP \ + "Number of messages received within the last 60 seconds" + +// number of disconnection within the last 60 seconds +#define NEU_METRIC_DISCONNECTION_60S "last_60s_disconnections" +#define NEU_METRIC_DISCONNECTION_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_DISCONNECTION_60S_HELP \ + "Number of disconnection within the last 60 seconds" + +// number of disconnection within the last 600 seconds +#define NEU_METRIC_DISCONNECTION_600S "last_600s_disconnections" +#define NEU_METRIC_DISCONNECTION_600S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_DISCONNECTION_600S_HELP \ + "Number of disconnection within the last 600 seconds" + +// number of disconnection within the last 1800 seconds +#define NEU_METRIC_DISCONNECTION_1800S "last_1800s_disconnections" +#define NEU_METRIC_DISCONNECTION_1800S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER +#define NEU_METRIC_DISCONNECTION_1800S_HELP \ + "Number of disconnection within the last 1800 seconds" + typedef enum { NEU_METRICS_CATEGORY_GLOBAL, NEU_METRICS_CATEGORY_DRIVER, @@ -140,11 +232,12 @@ typedef enum { // metric entry typedef struct { - const char * name; // NOTE: should points to string literal - const char * help; // NOTE: should points to string literal - neu_metric_type_e type; // - uint64_t value; // - UT_hash_handle hh; // ordered by name + const char * name; // NOTE: should points to string literal + const char * help; // NOTE: should points to string literal + neu_metric_type_e type; // + uint64_t value; // + neu_rolling_counter_t *rcnt; // + UT_hash_handle hh; // ordered by name } neu_metric_entry_t; // group metrics @@ -207,6 +300,9 @@ 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) { + neu_rolling_counter_free(entry->rcnt); + } free(entry); } diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index d223bce9..f94d6a5b 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -36,6 +36,14 @@ extern "C" { #define NEURON_PLUGIN_VER_1_0 \ NEU_VERSION(NEU_VERSION_MAJOR, NEU_VERSION_MINOR, NEU_VERSION_FIX) +#define NEU_PLUGIN_REGISTER_METRIC(plugin, name, init) \ + plugin->common.adapter_callbacks->register_metric( \ + plugin->common.adapter, name, name##_HELP, name##_TYPE, init) + +#define NEU_PLUGIN_UPDATE_METRIC(plugin, name, val, grp) \ + plugin->common.adapter_callbacks->update_metric(plugin->common.adapter, \ + name, val, grp) + extern int64_t global_timestamp; typedef struct neu_plugin_common { diff --git a/include/neuron/utils/rolling_counter.h b/include/neuron/utils/rolling_counter.h new file mode 100644 index 00000000..c0a7c010 --- /dev/null +++ b/include/neuron/utils/rolling_counter.h @@ -0,0 +1,106 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2024 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. + **/ +#ifndef NEURON_UTILS_ROLLING_COUNTER_H +#define NEURON_UTILS_ROLLING_COUNTER_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include +#include +#include +#include +#include + +/** Rolling counter. + * + * This counter is for counting values within some latest time span, like + * network bytes sent within the last 5 seconds etc. + */ +typedef struct { + uint64_t val; // accumulator + uint64_t ts; // head time stamp in milliseconds + uint32_t res : 22; // time resolution in milliseconds + uint32_t hd : 5; // head position + uint32_t n : 5; // number of counters + uint32_t counts[]; // bins of counters +} neu_rolling_counter_t; + +/** Create rolling counter. + * + * @param span time span in milliseconds + */ +static inline neu_rolling_counter_t *neu_rolling_counter_new(unsigned span) +{ + unsigned n = span <= 6000 ? 4 : span <= 32000 ? 8 : span <= 64000 ? 16 : 32; + assert(span / n < (1 << 22)); // should not overflow ti + + neu_rolling_counter_t *counter = (neu_rolling_counter_t *) calloc( + 1, sizeof(*counter) + sizeof(counter->counts[0]) * n); + if (counter) { + counter->res = span / n; + counter->n = n; + } + return counter; +} + +/** Destructs the rolling counter. + */ +static inline void neu_rolling_counter_free(neu_rolling_counter_t *counter) +{ + if (counter) { + free(counter); + } +} + +/** Increment the rolling counter and return the value; + * + * @param ts time stamp in milliseconds, should be monotonic + * @param dt delta value to increment by + */ +static inline uint64_t neu_rolling_counter_inc(neu_rolling_counter_t *counter, + uint64_t ts, unsigned dt) +{ + uint64_t step = (ts - counter->ts) / counter->res; + for (unsigned i = 0; i < step && i < counter->n; ++i) { + counter->hd = (counter->hd + 1) & (counter->n - 1); + counter->val -= counter->counts[counter->hd]; + counter->counts[counter->hd] = 0; + } + counter->val += dt; + counter->counts[counter->hd] += dt; + counter->ts += step * counter->res; + return counter->val; +} + +/** Return the counter value. + * + * NOTE: may return stale value if the counter is not updated frequent enough. + */ +static inline uint64_t neu_rolling_counter_value(neu_rolling_counter_t *counter) +{ + return counter->val; +} + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 55ff8738..7f62eacd 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -276,15 +276,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic, neu_plugin_t *plugin = data; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - if (0 == errcode) { - update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1, - NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, len, NULL); } else { - update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, - 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1, + NULL); } free(payload); @@ -293,16 +292,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic, static inline int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic, char *payload, size_t payload_len) { - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; int rv = neu_mqtt_client_publish(plugin->client, qos, topic, (uint8_t *) payload, (uint32_t) payload_len, plugin, publish_cb); if (0 != rv) { plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos); - update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, - 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1, + NULL); free(payload); rv = NEU_ERR_MQTT_PUBLISH_FAILURE; } @@ -320,9 +317,13 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic, (void) qos; (void) topic; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL); char *json_str = malloc(len + 1); if (NULL == json_str) { @@ -407,9 +408,13 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic, (void) qos; (void) topic; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL); char *json_str = malloc(len + 1); if (NULL == json_str) { diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index e734dce4..0c7d668f 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -37,6 +37,9 @@ static void disconnect_cb(void *data) { neu_plugin_t *plugin = data; plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL); plog_notice(plugin, "plugin `%s` disconnected", neu_plugin_module.module_name); } @@ -60,13 +63,26 @@ static int mqtt_plugin_close(neu_plugin_t *plugin) static int mqtt_plugin_init(neu_plugin_t *plugin, bool load) { (void) load; + + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_CACHED_MSGS_NUM, 0); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000); + plog_notice(plugin, "initialize plugin `%s` success", neu_plugin_module.module_name); - neu_adapter_register_metric_cb_t register_metric = - plugin->common.adapter_callbacks->register_metric; - register_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM, - NEU_METRIC_CACHED_MSGS_NUM_HELP, - NEU_METRIC_CACHED_MSGS_NUM_TYPE, 0); return NEU_ERR_SUCCESS; } @@ -333,15 +349,12 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, { neu_err_code_e error = NEU_ERR_SUCCESS; - neu_adapter_update_metric_cb_t update_metric = - plugin->common.adapter_callbacks->update_metric; - // update cached messages number per seconds if (NULL != plugin->client && (global_timestamp - plugin->cache_metric_update_ts) >= 1000) { - update_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM, - neu_mqtt_client_get_cached_msgs_num(plugin->client), - NULL); + NEU_PLUGIN_UPDATE_METRIC( + plugin, NEU_METRIC_CACHED_MSGS_NUM, + neu_mqtt_client_get_cached_msgs_num(plugin->client), NULL); plugin->cache_metric_update_ts = global_timestamp; } @@ -353,6 +366,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, error = handle_read_response(plugin, head->ctx, data); break; case NEU_REQRESP_TRANS_DATA: { + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL); error = handle_trans_data(plugin, data); break; } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 931514d4..46fec10e 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -415,6 +415,9 @@ static int adapter_update_metric(neu_adapter_t *adapter, if (NEU_METRIC_TYPE_COUNTER == entry->type) { entry->value += n; + } else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) { + entry->value = + neu_rolling_counter_inc(entry->rcnt, global_timestamp, n); } else { entry->value = n; } @@ -1535,6 +1538,9 @@ int neu_adapter_update_group_metric(neu_adapter_t *adapter, if (NEU_METRIC_TYPE_COUNTER == entry->type) { entry->value += n; + } else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) { + entry->value = + neu_rolling_counter_inc(entry->rcnt, global_timestamp, n); } else { entry->value = n; } diff --git a/src/base/metrics.c b/src/base/metrics.c index 1d19b717..deaffb85 100644 --- a/src/base/metrics.c +++ b/src/base/metrics.c @@ -262,10 +262,19 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name, return -1; } - entry->name = name; - entry->type = type; - entry->help = help; - entry->value = init; + if (NEU_METRIC_TYPE_ROLLING_COUNTER == type) { + entry->rcnt = neu_rolling_counter_new(init); + if (NULL == entry->rcnt) { + free(entry); + return -1; + } + } else { + entry->value = init; + } + + entry->name = name; + entry->type = type; + entry->help = help; HASH_ADD_STR(*entries, name, entry); return 0; } diff --git a/tests/ft/metrics/test_metrics.py b/tests/ft/metrics/test_metrics.py index 22e1528b..6c950467 100644 --- a/tests/ft/metrics/test_metrics.py +++ b/tests/ft/metrics/test_metrics.py @@ -102,8 +102,23 @@ class TestMetrics: "send_msg_errors_total": (0, {}), "recv_msgs_total": (0, {}), "cached_msgs": (0, {}), + "last_5s_trans_data_msgs": (0, {}), + "last_30s_trans_data_msgs": (0, {}), + "last_60s_trans_data_msgs": (0, {}), + "last_5s_send_bytes": (0, {}), + "last_30s_send_bytes": (0, {}), + "last_60s_send_bytes": (0, {}), + "last_5s_recv_bytes": (0, {}), + "last_30s_recv_bytes": (0, {}), + "last_60s_recv_bytes": (0, {}), + "last_5s_recv_msgs": (0, {}), + "last_30s_recv_msgs": (0, {}), + "last_60s_recv_msgs": (0, {}), + "last_60s_disconnections": (0, {}), + "last_600s_disconnections": (0, {}), + "last_1800s_disconnections": (0, {}), } - + assert_metrics(resp.content.decode('utf-8'), expected_metrics) @description(given="neuron is started", diff --git a/tests/ut/CMakeLists.txt b/tests/ut/CMakeLists.txt index 21b3a743..f4feb6a2 100644 --- a/tests/ut/CMakeLists.txt +++ b/tests/ut/CMakeLists.txt @@ -57,6 +57,13 @@ target_include_directories(async_queue_test PRIVATE ${CMAKE_SOURCE_DIR}/include) target_link_libraries(async_queue_test neuron-base gtest_main gtest) +add_executable(rolling_counter_test rolling_counter_test.cc) +target_include_directories(rolling_counter_test PRIVATE + ${CMAKE_SOURCE_DIR}/src + ${CMAKE_SOURCE_DIR}/include +) +target_link_libraries(rolling_counter_test neuron-base gtest_main gtest) + include(GoogleTest) gtest_discover_tests(json_test) gtest_discover_tests(http_test) @@ -64,4 +71,5 @@ gtest_discover_tests(jwt_test) gtest_discover_tests(base64_test) gtest_discover_tests(tag_sort_test) gtest_discover_tests(modbus_test) -gtest_discover_tests(async_queue_test) \ No newline at end of file +gtest_discover_tests(async_queue_test) +gtest_discover_tests(rolling_counter_test) diff --git a/tests/ut/rolling_counter_test.cc b/tests/ut/rolling_counter_test.cc new file mode 100644 index 00000000..fb72c458 --- /dev/null +++ b/tests/ut/rolling_counter_test.cc @@ -0,0 +1,45 @@ +#include + +#include "utils/log.h" +#include "utils/rolling_counter.h" + +zlog_category_t *neuron = NULL; +TEST(RollingCounterTest, neu_rolling_counter_inc) +{ + uint64_t ts = 0; + neu_rolling_counter_t *counter = neu_rolling_counter_new(4000); + EXPECT_NE(nullptr, counter); + + for (int i = 1; i <= 4; ++i) { + ts += 1000; + EXPECT_EQ(i, neu_rolling_counter_inc(counter, ts, 1)); + } + + // wrap around + for (int i = 1; i <= 4; ++i) { + ts += 1000; + EXPECT_EQ(4, neu_rolling_counter_inc(counter, ts, 1)); + } + + // increment without time stamp update + EXPECT_EQ(4, neu_rolling_counter_value(counter)); + for (int i = 1; i <= 10; ++i) { + EXPECT_EQ(4 + i, neu_rolling_counter_inc(counter, ts, 1)); + } + + EXPECT_EQ(4 + 10, neu_rolling_counter_value(counter)); + + // stale value will be clean + ts += 4000; + EXPECT_EQ(0, neu_rolling_counter_inc(counter, ts, 0)); + + neu_rolling_counter_free(counter); +} + +int main(int argc, char **argv) +{ + zlog_init("./config/dev.conf"); + neuron = zlog_get_category("neuron"); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}