diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h index 09334433..0bfd993d 100644 --- a/include/neuron/metrics.h +++ b/include/neuron/metrics.h @@ -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, 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/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",