Adds chart for incoming proto msgs in new cloud protocol (#11969)
Timotej S committed
Feb 28, 2022 at 09:45 UTC
6c2073510413f4a2784dc8113ea057e9fa716b65
5 files changed
+95
-15
aclk/aclk.c
+3
-1
@@ -761,8 +761,9 @@ void *aclk_main(void *ptr)
761
return NULL;
762
}
763
764
+ unsigned int proto_hdl_cnt;
765
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
765
- aclk_init_rx_msg_handlers();
766
+ proto_hdl_cnt = aclk_init_rx_msg_handlers();
767
#endif
768
769
// This thread is unusual in that it cannot be cancelled by cancel_main_threads()
@@ -802,6 +803,7 @@ void *aclk_main(void *ptr)
803
stats_thread = callocz(1, sizeof(struct aclk_stats_thread));
804
stats_thread->thread = mallocz(sizeof(netdata_thread_t));
805
stats_thread->query_thread_count = query_threads.count;
806
+ aclk_stats_thread_prepare(query_threads.count, proto_hdl_cnt);
807
netdata_thread_create(
808
stats_thread->thread, ACLK_STATS_THREAD_NAME, NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread,
809
stats_thread);
aclk/aclk_rx_msgs.c
+14
-2
@@ -457,9 +457,15 @@ new_cloud_rx_msg_t *find_rx_handler_by_hash(simple_hash_t hash)
457
return NULL;
458
}
459
460
-void aclk_init_rx_msg_handlers(void)
460
+const char *rx_handler_get_name(size_t i)
461
{
462
- for (int i = 0; rx_msgs[i].fnc; i++) {
462
+ return rx_msgs[i].name;
463
+}
464
+
465
+unsigned int aclk_init_rx_msg_handlers(void)
466
+{
467
+ int i;
468
+ for (i = 0; rx_msgs[i].fnc; i++) {
469
simple_hash_t hash = simple_hash(rx_msgs[i].name);
470
new_cloud_rx_msg_t *hdl = find_rx_handler_by_hash(hash);
471
if (unlikely(hdl)) {
@@ -469,6 +475,7 @@ void aclk_init_rx_msg_handlers(void)
475
}
476
rx_msgs[i].name_hash = hash;
477
}
478
+ return i;
479
}
480
481
void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len)
@@ -489,6 +496,11 @@ void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t
496
}
497
return;
498
}
499
+ if (aclk_stats_enabled) {
500
+ ACLK_STATS_LOCK;
501
+ aclk_proto_rx_msgs_sample[msg_descriptor-rx_msgs]++;
502
+ ACLK_STATS_UNLOCK;
503
+ }
504
if (msg_descriptor->fnc(msg, msg_len)) {
505
error("Error processing message of type '%s'", message_type);
506
if (aclk_stats_enabled) {
aclk/aclk_rx_msgs.h
+2
-1
@@ -11,7 +11,8 @@
11
int aclk_handle_cloud_cmd_message(char *payload);
12
13
#ifdef ENABLE_NEW_CLOUD_PROTOCOL
14
-void aclk_init_rx_msg_handlers(void);
14
+const char *rx_handler_get_name(size_t i);
15
+unsigned int aclk_init_rx_msg_handlers(void);
16
void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len);
17
#endif
18
aclk/aclk_stats.c
+71
-11
@@ -6,7 +6,14 @@
6
7
netdata_mutex_t aclk_stats_mutex = NETDATA_MUTEX_INITIALIZER;
8
9
-int query_thread_count;
9
+struct {
10
+ int query_thread_count;
11
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
12
+ unsigned int proto_hdl_cnt;
13
+ uint32_t *aclk_proto_rx_msgs_sample;
14
+ RRDDIM **rx_msg_dims;
15
+#endif
16
+} aclk_stats_cfg; // there is only 1 stats thread at a time
17
18
// data ACLK stats need per query thread
19
struct aclk_qt_data {
@@ -15,6 +22,7 @@ struct aclk_qt_data {
22
23
uint32_t *aclk_queries_per_thread = NULL;
24
uint32_t *aclk_queries_per_thread_sample = NULL;
25
+uint32_t *aclk_proto_rx_msgs_sample = NULL;
26
27
struct aclk_metrics aclk_metrics = {
28
.online = 0,
@@ -186,7 +194,7 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread)
194
"netdata", "aclk_query_threads", NULL, "aclk", NULL, "Queries Processed Per Thread", "req/s",
195
"netdata", "stats", 200009, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
196
189
- for (int i = 0; i < query_thread_count; i++) {
197
+ for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) {
198
if (snprintfz(dim_name, MAX_DIM_NAME, "Query %d", i) < 0)
199
error("snprintf encoding error");
200
aclk_qt_data[i].dim = rrddim_add(st, dim_name, NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
@@ -194,7 +202,7 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread)
202
} else
203
rrdset_next(st);
204
197
- for (int i = 0; i < query_thread_count; i++) {
205
+ for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) {
206
rrddim_set_by_pointer(st, aclk_qt_data[i].dim, queries_per_thread[i]);
207
}
208
@@ -229,8 +237,57 @@ static void aclk_stats_query_time(struct aclk_metrics_per_sample *per_sample)
237
rrdset_done(st);
238
}
239
240
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
241
+const char *rx_handler_get_name(size_t i);
242
+static void aclk_stats_newproto_rx(uint32_t *rx_msgs_sample)
243
+{
244
+ static RRDSET *st = NULL;
245
+
246
+ if (unlikely(!st)) {
247
+ st = rrdset_create_localhost(
248
+ "netdata", "aclk_protobuf_rx_types", NULL, "aclk", NULL, "Received new cloud architecture messages by their type.", "msg/s",
249
+ "netdata", "stats", 200010, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
250
+
251
+ for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++) {
252
+ aclk_stats_cfg.rx_msg_dims[i] = rrddim_add(st, rx_handler_get_name(i), NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
253
+ }
254
+ } else
255
+ rrdset_next(st);
256
+
257
+ for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++)
258
+ rrddim_set_by_pointer(st, aclk_stats_cfg.rx_msg_dims[i], rx_msgs_sample[i]);
259
+
260
+ rrdset_done(st);
261
+}
262
+#endif
263
+
264
+void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt)
265
+{
266
+#ifndef ENABLE_NEW_CLOUD_PROTOCOL
267
+ UNUSED(proto_hdl_cnt);
268
+#endif
269
+
270
+ aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data));
271
+ aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t));
272
+ aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t));
273
+
274
+ memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
275
+
276
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
277
+ aclk_stats_cfg.proto_hdl_cnt = proto_hdl_cnt;
278
+ aclk_stats_cfg.aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample));
279
+ aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample));
280
+ aclk_stats_cfg.rx_msg_dims = callocz(proto_hdl_cnt, sizeof(RRDDIM*));
281
+#endif
282
+}
283
+
284
void aclk_stats_thread_cleanup()
285
{
286
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
287
+ freez(aclk_stats_cfg.rx_msg_dims);
288
+ freez(aclk_proto_rx_msgs_sample);
289
+ freez(aclk_stats_cfg.aclk_proto_rx_msgs_sample);
290
+#endif
291
freez(aclk_qt_data);
292
freez(aclk_queries_per_thread);
293
freez(aclk_queries_per_thread_sample);
@@ -240,17 +297,12 @@ void *aclk_stats_main_thread(void *ptr)
297
{
298
struct aclk_stats_thread *args = ptr;
299
243
- query_thread_count = args->query_thread_count;
244
- aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data));
245
- aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t));
246
- aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t));
300
+ aclk_stats_cfg.query_thread_count = args->query_thread_count;
301
302
heartbeat_t hb;
303
heartbeat_init(&hb);
304
usec_t step_ut = localhost->rrd_update_every * USEC_PER_SEC;
305
252
- memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
253
-
306
struct aclk_metrics_per_sample per_sample;
307
struct aclk_metrics permanent;
308
@@ -266,11 +318,15 @@ void *aclk_stats_main_thread(void *ptr)
318
// to not hold lock longer than necessary, especially not to hold it
319
// during database rrd* operations
320
memcpy(&per_sample, &aclk_metrics_per_sample, sizeof(struct aclk_metrics_per_sample));
321
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
322
+ 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);
323
+ memset(aclk_proto_rx_msgs_sample, 0, sizeof(*aclk_proto_rx_msgs_sample) * aclk_stats_cfg.proto_hdl_cnt);
324
+#endif
325
memcpy(&permanent, &aclk_metrics, sizeof(struct aclk_metrics));
326
memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
327
272
- memcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * query_thread_count);
273
- memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * query_thread_count);
328
+ memcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count);
329
+ memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count);
330
ACLK_STATS_UNLOCK;
331
332
aclk_stats_collect(&per_sample, &permanent);
@@ -286,6 +342,10 @@ void *aclk_stats_main_thread(void *ptr)
342
aclk_stats_query_threads(aclk_queries_per_thread_sample);
343
344
aclk_stats_query_time(&per_sample);
345
+
346
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
347
+ aclk_stats_newproto_rx(aclk_stats_cfg.aclk_proto_rx_msgs_sample);
348
+#endif
349
}
350
351
return 0;
aclk/aclk_stats.h
+5
@@ -60,9 +60,14 @@ extern struct aclk_metrics_per_sample {
60
volatile uint32_t cloud_q_process_max;
61
} aclk_metrics_per_sample;
62
63
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
64
+extern uint32_t *aclk_proto_rx_msgs_sample;
65
+#endif
66
+
67
extern uint32_t *aclk_queries_per_thread;
68
69
void *aclk_stats_main_thread(void *ptr);
70
+void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt);
71
void aclk_stats_thread_cleanup();
72
void aclk_stats_upd_online(int online);
73