From b5321aff06d6ea8d730d62aec2ffd8e9271c1ffc Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Thu, 14 Apr 2022 20:12:10 +0200 Subject: Adding upstream version 1.34.0. Signed-off-by: Daniel Baumann --- aclk/aclk_stats.c | 114 +++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 79 insertions(+), 35 deletions(-) (limited to 'aclk/aclk_stats.c') diff --git a/aclk/aclk_stats.c b/aclk/aclk_stats.c index a7d4a4709..a9f0a923c 100644 --- a/aclk/aclk_stats.c +++ b/aclk/aclk_stats.c @@ -2,9 +2,18 @@ #include "aclk_stats.h" +#include "aclk_query.h" + netdata_mutex_t aclk_stats_mutex = NETDATA_MUTEX_INITIALIZER; -int query_thread_count; +struct { + int query_thread_count; +#ifdef ENABLE_NEW_CLOUD_PROTOCOL + unsigned int proto_hdl_cnt; + uint32_t *aclk_proto_rx_msgs_sample; + RRDDIM **rx_msg_dims; +#endif +} aclk_stats_cfg; // there is only 1 stats thread at a time // data ACLK stats need per query thread struct aclk_qt_data { @@ -13,6 +22,7 @@ struct aclk_qt_data { uint32_t *aclk_queries_per_thread = NULL; uint32_t *aclk_queries_per_thread_sample = NULL; +uint32_t *aclk_proto_rx_msgs_sample = NULL; struct aclk_metrics aclk_metrics = { .online = 0, @@ -113,39 +123,21 @@ static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample) static void aclk_stats_cloud_req_type(struct aclk_metrics_per_sample *per_sample) { static RRDSET *st = NULL; - static RRDDIM *rd_type_http = NULL; - static RRDDIM *rd_type_alarm_upd = NULL; - static RRDDIM *rd_type_metadata_info = NULL; - static RRDDIM *rd_type_metadata_alarms = NULL; - static RRDDIM *rd_type_chart_new = NULL; - static RRDDIM *rd_type_chart_del = NULL; - static RRDDIM *rd_type_register_node = NULL; - static RRDDIM *rd_type_node_upd = NULL; + static RRDDIM *dims[ACLK_QUERY_TYPE_COUNT]; if (unlikely(!st)) { st = rrdset_create_localhost( "netdata", "aclk_processed_query_type", NULL, "aclk", NULL, "Query thread commands processed by their type", "cmd/s", "netdata", "stats", 200006, localhost->rrd_update_every, RRDSET_TYPE_STACKED); - rd_type_http = rrddim_add(st, "http", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_alarm_upd = rrddim_add(st, "alarm update", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_metadata_info = rrddim_add(st, "info metadata", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_metadata_alarms = rrddim_add(st, "alarms metadata", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_chart_new = rrddim_add(st, "chart new", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_chart_del = rrddim_add(st, "chart delete", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_register_node = rrddim_add(st, "register node", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); - rd_type_node_upd = rrddim_add(st, "node update", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); + for (int i = 0; i < ACLK_QUERY_TYPE_COUNT; i++) + dims[i] = rrddim_add(st, aclk_query_get_name(i), NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); + } else rrdset_next(st); - rrddim_set_by_pointer(st, rd_type_http, per_sample->query_type_http); - rrddim_set_by_pointer(st, rd_type_alarm_upd, per_sample->query_type_alarm_upd); - rrddim_set_by_pointer(st, rd_type_metadata_info, per_sample->query_type_metadata_info); - rrddim_set_by_pointer(st, rd_type_metadata_alarms, per_sample->query_type_metadata_alarms); - rrddim_set_by_pointer(st, rd_type_chart_new, per_sample->query_type_chart_new); - rrddim_set_by_pointer(st, rd_type_chart_del, per_sample->query_type_chart_del); - rrddim_set_by_pointer(st, rd_type_register_node, per_sample->query_type_register_node); - rrddim_set_by_pointer(st, rd_type_node_upd, per_sample->query_type_node_upd); + for (int i = 0; i < ACLK_QUERY_TYPE_COUNT; i++) + rrddim_set_by_pointer(st, dims[i], per_sample->queries_per_type[i]); rrdset_done(st); } @@ -202,7 +194,7 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread) "netdata", "aclk_query_threads", NULL, "aclk", NULL, "Queries Processed Per Thread", "req/s", "netdata", "stats", 200009, localhost->rrd_update_every, RRDSET_TYPE_STACKED); - for (int i = 0; i < query_thread_count; i++) { + for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) { if (snprintfz(dim_name, MAX_DIM_NAME, "Query %d", i) < 0) error("snprintf encoding error"); aclk_qt_data[i].dim = rrddim_add(st, dim_name, NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); @@ -210,7 +202,7 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread) } else rrdset_next(st); - for (int i = 0; i < query_thread_count; i++) { + for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) { rrddim_set_by_pointer(st, aclk_qt_data[i].dim, queries_per_thread[i]); } @@ -245,8 +237,57 @@ static void aclk_stats_query_time(struct aclk_metrics_per_sample *per_sample) rrdset_done(st); } +#ifdef ENABLE_NEW_CLOUD_PROTOCOL +const char *rx_handler_get_name(size_t i); +static void aclk_stats_newproto_rx(uint32_t *rx_msgs_sample) +{ + static RRDSET *st = NULL; + + if (unlikely(!st)) { + st = rrdset_create_localhost( + "netdata", "aclk_protobuf_rx_types", NULL, "aclk", NULL, "Received new cloud architecture messages by their type.", "msg/s", + "netdata", "stats", 200010, localhost->rrd_update_every, RRDSET_TYPE_STACKED); + + for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++) { + aclk_stats_cfg.rx_msg_dims[i] = rrddim_add(st, rx_handler_get_name(i), NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE); + } + } else + rrdset_next(st); + + for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++) + rrddim_set_by_pointer(st, aclk_stats_cfg.rx_msg_dims[i], rx_msgs_sample[i]); + + rrdset_done(st); +} +#endif + +void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt) +{ +#ifndef ENABLE_NEW_CLOUD_PROTOCOL + UNUSED(proto_hdl_cnt); +#endif + + aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data)); + aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t)); + aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t)); + + memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample)); + +#ifdef ENABLE_NEW_CLOUD_PROTOCOL + aclk_stats_cfg.proto_hdl_cnt = proto_hdl_cnt; + aclk_stats_cfg.aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample)); + aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample)); + aclk_stats_cfg.rx_msg_dims = callocz(proto_hdl_cnt, sizeof(RRDDIM*)); +#endif +} + void aclk_stats_thread_cleanup() { +#ifdef ENABLE_NEW_CLOUD_PROTOCOL + freez(aclk_stats_cfg.rx_msg_dims); + freez(aclk_proto_rx_msgs_sample); + freez(aclk_stats_cfg.aclk_proto_rx_msgs_sample); +#endif freez(aclk_qt_data); freez(aclk_queries_per_thread); freez(aclk_queries_per_thread_sample); @@ -256,17 +297,12 @@ void *aclk_stats_main_thread(void *ptr) { struct aclk_stats_thread *args = ptr; - query_thread_count = args->query_thread_count; - aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data)); - aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t)); - aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t)); + aclk_stats_cfg.query_thread_count = args->query_thread_count; heartbeat_t hb; heartbeat_init(&hb); usec_t step_ut = localhost->rrd_update_every * USEC_PER_SEC; - memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample)); - struct aclk_metrics_per_sample per_sample; struct aclk_metrics permanent; @@ -282,11 +318,15 @@ void *aclk_stats_main_thread(void *ptr) // to not hold lock longer than necessary, especially not to hold it // during database rrd* operations memcpy(&per_sample, &aclk_metrics_per_sample, sizeof(struct aclk_metrics_per_sample)); +#ifdef ENABLE_NEW_CLOUD_PROTOCOL + memcpy(aclk_stats_cfg.aclk_proto_rx_msgs_sample, aclk_proto_rx_msgs_sample, sizeof(*aclk_proto_rx_msgs_sample) * aclk_stats_cfg.proto_hdl_cnt); + memset(aclk_proto_rx_msgs_sample, 0, sizeof(*aclk_proto_rx_msgs_sample) * aclk_stats_cfg.proto_hdl_cnt); +#endif memcpy(&permanent, &aclk_metrics, sizeof(struct aclk_metrics)); memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample)); - memcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * query_thread_count); - memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * query_thread_count); + memcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count); + memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count); ACLK_STATS_UNLOCK; aclk_stats_collect(&per_sample, &permanent); @@ -302,6 +342,10 @@ void *aclk_stats_main_thread(void *ptr) aclk_stats_query_threads(aclk_queries_per_thread_sample); aclk_stats_query_time(&per_sample); + +#ifdef ENABLE_NEW_CLOUD_PROTOCOL + aclk_stats_newproto_rx(aclk_stats_cfg.aclk_proto_rx_msgs_sample); +#endif } return 0; -- cgit v1.2.3