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

feat(mqtt): add rolling metrics

Browse Source
This commit is contained in:
eeff
2024-02-01 22:42:18 +08:00
parent 6bea8c254e
commit 353dbcd18c
4 changed files with 155 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/metrics.h
+90
View File
@@ -133,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,
plugins/mqtt/mqtt_handle.c
+22 -17
View File
@@ -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) {
plugins/mqtt/mqtt_plugin.c
+27 -11
View File
@@ -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;
}
tests/ft/metrics/test_metrics.py
+16 -1
View File
@@ -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",