feat(aclk): Add detailed pulse metrics for ACLK telemetry (#20802)
* feat(aclk): Add detailed pulse metrics for ACLK telemetry Adds several new charts to the internal `pulse` telemetry system to provide deeper insight into ACLK performance. New metrics include: - Per-iteration min/avg/max PUBACK latency. - Max wait times for the send queue and PUBACK queue. - Split send queue wait times into "unsent" and "partial". All timing charts are presented in milliseconds for readability. * remove _Atomic
Costa Tsaousis committed
Aug 11, 2025 at 20:55 UTC
64ede9565aa741b2610144af6318a7198cf4ad13
3 files changed
+206
src/aclk/mqtt_websockets/common_public.h
+8
@@ -4,6 +4,7 @@
4
#define MQTT_WEBSOCKETS_COMMON_PUBLIC_H
5
6
#include <stddef.h>
7
+#include <stdint.h>
8
9
/* free_fnc_t in general (in whatever function or struct it is used)
10
* decides how the related data will be handled.
@@ -31,6 +32,13 @@ struct mqtt_ng_stats {
32
size_t tx_buffer_size;
33
// part of transaction buffer that containes mesages we can free alredy during the garbage colleciton step
34
size_t tx_buffer_reclaimable;
35
+ // maximum time (in microseconds) a QoS1 message is currently waiting for PUBACK
36
+ uint64_t max_puback_wait_us;
37
+ // maximum time (in microseconds) a message has been waiting in the send queue without starting transmission
38
+ uint64_t max_send_queue_wait_us;
39
+ // split of send-queue wait into unsent and partial (diagnostics)
40
+ uint64_t max_unsent_wait_us;
41
+ uint64_t max_partial_wait_us;
42
};
43
44
#endif /* MQTT_WEBSOCKETS_COMMON_PUBLIC_H */
src/aclk/mqtt_websockets/mqtt_ng.c
+58
@@ -41,6 +41,8 @@ struct buffer_fragment {
41
uint16_t packet_id;
42
void (*free_fnc)(void *ptr);
43
unsigned char *data;
44
+ // timestamp (monotonic usec) of when this MQTT packet was enqueued into the transaction buffer (set on HEAD)
45
+ usec_t enqueued_monotonic_ut;
46
usec_t sent_monotonic_ut;
47
struct buffer_fragment *next;
48
};
@@ -1151,6 +1153,9 @@ int mqtt_ng_generate_publish(struct transaction_buffer *trx_buf,
1153
if (!qos)
1154
trx_buf->hdr_buffer.tail_frag->flags |= BUFFER_FRAG_GARBAGE_COLLECT_ON_SEND;
1155
transaction_buffer_transaction_commit(trx_buf)
1156
+ // mark enqueue time on the HEAD fragment (after commit) so we measure from commit time
1157
+ if (mqtt_msg)
1158
+ mqtt_msg->enqueued_monotonic_ut = now_monotonic_usec();
1159
return MQTT_NG_MSGGEN_OK;
1160
fail_rollback:
1161
transaction_buffer_transaction_rollback(trx_buf, mqtt_msg);
@@ -2262,20 +2267,73 @@ void mqtt_ng_get_stats(struct mqtt_ng_client *client, struct mqtt_ng_stats *stat
2267
2268
stats->tx_bytes_queued = 0;
2269
stats->tx_buffer_reclaimable = 0;
2270
+ stats->max_puback_wait_us = 0;
2271
+ stats->max_send_queue_wait_us = 0;
2272
+ stats->max_unsent_wait_us = 0;
2273
+ stats->max_partial_wait_us = 0;
2274
2275
+ // First pass: compute buffer usage/queued bytes and max send-queue wait time (unsent messages)
2276
LOCK_HDR_BUFFER(&client->main_buffer);
2277
stats->tx_buffer_used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);
2278
stats->tx_buffer_free = BUFFER_BYTES_AVAILABLE(&client->main_buffer.hdr_buffer);
2279
stats->tx_buffer_size = client->main_buffer.hdr_buffer.size;
2280
struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
2281
+ usec_t now_ut = now_monotonic_usec();
2282
while (frag) {
2283
stats->tx_bytes_queued += frag->len - frag->sent;
2284
if (frag_is_marked_for_gc(frag))
2285
stats->tx_buffer_reclaimable += FRAG_SIZE_IN_BUFFER(frag);
2286
2287
+ // For HEAD fragments that are not fully sent yet (unsent or partially sent),
2288
+ // track max send-queue wait time. Prefer the enqueue timestamp; if missing, fall back to first-send.
2289
+ if ((frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->sent < frag->len) {
2290
+ usec_t base = frag->enqueued_monotonic_ut ? frag->enqueued_monotonic_ut : frag->sent_monotonic_ut;
2291
+ if (base) {
2292
+ usec_t waited = now_ut - base;
2293
+ uint64_t w = (uint64_t)waited;
2294
+ if (frag->sent == 0) {
2295
+ if (w > stats->max_unsent_wait_us) stats->max_unsent_wait_us = w;
2296
+ } else {
2297
+ if (w > stats->max_partial_wait_us) stats->max_partial_wait_us = w;
2298
+ }
2299
+ if (w > stats->max_send_queue_wait_us) stats->max_send_queue_wait_us = w;
2300
+ } else {
2301
+ // Throttled debug if enqueue time is missing on an unsent HEAD
2302
+ if (frag->sent == 0) {
2303
+ static time_t last_warn = 0;
2304
+ time_t now_s = now_monotonic_sec();
2305
+ if (now_s - last_warn > 60) {
2306
+ nd_log(NDLS_DAEMON, NDLP_DEBUG, "ACLK: Missing enqueue timestamp on unsent MQTT packet head");
2307
+ last_warn = now_s;
2308
+ }
2309
+ }
2310
+ }
2311
+ }
2312
+
2313
frag = frag->next;
2314
}
2315
UNLOCK_HDR_BUFFER(&client->main_buffer);
2316
+
2317
+ // Second pass: compute max PUBACK wait time by correlating HEAD fragments with pending packet IDs
2318
+ spinlock_lock(&client->pending_packets.spinlock);
2319
+ LOCK_HDR_BUFFER(&client->main_buffer);
2320
+ frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);
2321
+ while (frag) {
2322
+ if ((frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD) && frag->packet_id) {
2323
+ Pvoid_t *Pvalue = JudyLGet(client->pending_packets.JudyL, (Word_t)frag->packet_id, PJE0);
2324
+ if (Pvalue) {
2325
+ // message is still pending PUBACK
2326
+ if (frag->sent_monotonic_ut) {
2327
+ usec_t waited = now_ut - frag->sent_monotonic_ut;
2328
+ if ((uint64_t)waited > stats->max_puback_wait_us)
2329
+ stats->max_puback_wait_us = (uint64_t)waited;
2330
+ }
2331
+ }
2332
+ }
2333
+ frag = frag->next;
2334
+ }
2335
+ UNLOCK_HDR_BUFFER(&client->main_buffer);
2336
+ spinlock_unlock(&client->pending_packets.spinlock);
2337
}
2338
2339
int mqtt_ng_set_topic_alias(struct mqtt_ng_client *client, const char *topic)
src/daemon/pulse/pulse-network.c
+140
@@ -92,6 +92,12 @@ static inline size_t aclk_time_histogram_slot(struct aclk_time_histogram *h, use
92
return low - 1;
93
}
94
95
+// Per-iteration PUBACK latency accumulators (microseconds)
96
+static uint64_t aclk_ack_count = 0;
97
+static uint64_t aclk_ack_sum_us = 0;
98
+static uint64_t aclk_ack_min_us = UINT64_MAX;
99
+static uint64_t aclk_ack_max_us = 0;
100
+
101
void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len __maybe_unused) {
102
if(!publish_latency) return;
103
@@ -101,6 +107,23 @@ void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len __maybe_un
107
internal_fatal(slot >= _countof(aclk_time_heatmap.array), "hey!");
108
109
__atomic_add_fetch(&aclk_time_heatmap.array[slot].count, 1, __ATOMIC_RELAXED);
110
+
111
+ // Track per-iteration min/avg/max in microseconds using atomics
112
+ __atomic_add_fetch(&aclk_ack_count, 1, __ATOMIC_RELAXED);
113
+ __atomic_add_fetch(&aclk_ack_sum_us, (uint64_t)publish_latency, __ATOMIC_RELAXED);
114
+
115
+ // atomic min update
116
+ uint64_t cur_min = __atomic_load_n(&aclk_ack_min_us, __ATOMIC_RELAXED);
117
+ while (publish_latency < cur_min &&
118
+ !__atomic_compare_exchange_n(&aclk_ack_min_us, &cur_min, (uint64_t)publish_latency, true, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) {
119
+ ;
120
+ }
121
+ // atomic max update
122
+ uint64_t cur_max = __atomic_load_n(&aclk_ack_max_us, __ATOMIC_RELAXED);
123
+ while (publish_latency > cur_max &&
124
+ !__atomic_compare_exchange_n(&aclk_ack_max_us, &cur_max, (uint64_t)publish_latency, true, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) {
125
+ ;
126
+ }
127
}
128
129
static void pulse_aclk_time_heatmap(void) {
@@ -372,5 +395,122 @@ void pulse_network_do(bool extended __maybe_unused) {
395
rrddim_set_by_pointer(st_aclk_messages, rd_out, (collected_number)t.mqtt.tx_messages_sent);
396
rrdset_done(st_aclk_messages);
397
}
398
+
399
+ if(extended) {
400
+ // Bytes queued for send
401
+ static RRDSET *st_aclk_bytes = NULL;
402
+ static RRDDIM *rd_bytes = NULL;
403
+ if (unlikely(!st_aclk_bytes)) {
404
+ st_aclk_bytes = rrdset_create_localhost(
405
+ "netdata",
406
+ "network_aclk_send_queue_bytes",
407
+ NULL,
408
+ PULSE_NETWORK_CHART_FAMILY,
409
+ "netdata.network_aclk_send_queue_bytes",
410
+ "Netdata ACLK Send Queue Bytes",
411
+ "bytes",
412
+ "netdata",
413
+ "pulse",
414
+ PULSE_NETWORK_CHART_PRIORITY + 2,
415
+ localhost->rrd_update_every,
416
+ RRDSET_TYPE_LINE);
417
+ rrdlabels_add(st_aclk_bytes->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);
418
+ rd_bytes = rrddim_add(st_aclk_bytes, "bytes", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
419
+ }
420
+ rrddim_set_by_pointer(st_aclk_bytes, rd_bytes, (collected_number)t.mqtt.tx_bytes_queued);
421
+ rrdset_done(st_aclk_bytes);
422
+ }
423
+
424
+ if(extended) {
425
+ // Max wait times for PUBACK and send queue (convert us->s via divisor)
426
+ static RRDSET *st_aclk_puback_wait = NULL;
427
+ static RRDDIM *rd_puback_max = NULL;
428
+ if (unlikely(!st_aclk_puback_wait)) {
429
+ st_aclk_puback_wait = rrdset_create_localhost(
430
+ "netdata",
431
+ "network_aclk_puback_wait",
432
+ NULL,
433
+ PULSE_NETWORK_CHART_FAMILY,
434
+ "netdata.network_aclk_puback_wait",
435
+ "Netdata ACLK PUBACK Max Wait",
436
+ "seconds",
437
+ "netdata",
438
+ "pulse",
439
+ PULSE_NETWORK_CHART_PRIORITY + 3,
440
+ localhost->rrd_update_every,
441
+ RRDSET_TYPE_LINE);
442
+ rrdlabels_add(st_aclk_puback_wait->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);
443
+ rd_puback_max = rrddim_add(st_aclk_puback_wait, "max", NULL, 1, USEC_PER_SEC, RRD_ALGORITHM_ABSOLUTE);
444
+ }
445
+ rrddim_set_by_pointer(st_aclk_puback_wait, rd_puback_max, (collected_number)t.mqtt.max_puback_wait_us);
446
+ rrdset_done(st_aclk_puback_wait);
447
+ }
448
+
449
+ if(extended) {
450
+ static RRDSET *st_aclk_send_wait = NULL;
451
+ static RRDDIM *rd_overall = NULL, *rd_unsent = NULL, *rd_partial = NULL;
452
+ if (unlikely(!st_aclk_send_wait)) {
453
+ st_aclk_send_wait = rrdset_create_localhost(
454
+ "netdata",
455
+ "network_aclk_send_queue_wait",
456
+ NULL,
457
+ PULSE_NETWORK_CHART_FAMILY,
458
+ "netdata.network_aclk_send_queue_wait",
459
+ "Netdata ACLK Send Queue Wait (Overall/Unsent/Partial)",
460
+ "seconds",
461
+ "netdata",
462
+ "pulse",
463
+ PULSE_NETWORK_CHART_PRIORITY + 3,
464
+ localhost->rrd_update_every,
465
+ RRDSET_TYPE_LINE);
466
+ rrdlabels_add(st_aclk_send_wait->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);
467
+ rd_overall = rrddim_add(st_aclk_send_wait, "overall", NULL, 1, USEC_PER_SEC, RRD_ALGORITHM_ABSOLUTE);
468
+ rd_unsent = rrddim_add(st_aclk_send_wait, "unsent", NULL, 1, USEC_PER_SEC, RRD_ALGORITHM_ABSOLUTE);
469
+ rd_partial = rrddim_add(st_aclk_send_wait, "partial", NULL, 1, USEC_PER_SEC, RRD_ALGORITHM_ABSOLUTE);
470
+ }
471
+ rrddim_set_by_pointer(st_aclk_send_wait, rd_overall, (collected_number)t.mqtt.max_send_queue_wait_us);
472
+ rrddim_set_by_pointer(st_aclk_send_wait, rd_unsent, (collected_number)t.mqtt.max_unsent_wait_us);
473
+ rrddim_set_by_pointer(st_aclk_send_wait, rd_partial, (collected_number)t.mqtt.max_partial_wait_us);
474
+ rrdset_done(st_aclk_send_wait);
475
+ }
476
+ }
477
+
478
+ // Publish PUBACK latency min/avg/max per iteration
479
+ {
480
+ static RRDSET *st_stats = NULL;
481
+ static RRDDIM *rd_min = NULL, *rd_avg = NULL, *rd_max = NULL;
482
+
483
+ // pull and reset accumulators
484
+ uint64_t count = __atomic_exchange_n(&aclk_ack_count, 0, __ATOMIC_RELAXED);
485
+ uint64_t sum_us = __atomic_exchange_n(&aclk_ack_sum_us, 0, __ATOMIC_RELAXED);
486
+ uint64_t min_us = __atomic_exchange_n(&aclk_ack_min_us, UINT64_MAX, __ATOMIC_RELAXED);
487
+ uint64_t max_us = __atomic_exchange_n(&aclk_ack_max_us, 0, __ATOMIC_RELAXED);
488
+
489
+ uint64_t avg_us = (count ? (sum_us / count) : 0);
490
+ if (min_us == UINT64_MAX) min_us = 0;
491
+
492
+ if (unlikely(!st_stats)) {
493
+ st_stats = rrdset_create_localhost(
494
+ "netdata",
495
+ "aclk_puback_latency_stats",
496
+ NULL,
497
+ PULSE_NETWORK_CHART_FAMILY,
498
+ "netdata.aclk_puback_latency_stats",
499
+ "Netdata ACLK PubACK Latency (Min/Avg/Max)",
500
+ "milliseconds",
501
+ "netdata",
502
+ "pulse",
503
+ PULSE_NETWORK_CHART_PRIORITY + 1,
504
+ localhost->rrd_update_every,
505
+ RRDSET_TYPE_LINE);
506
+ rd_min = rrddim_add(st_stats, "min", NULL, 1, USEC_PER_MS, RRD_ALGORITHM_ABSOLUTE);
507
+ rd_avg = rrddim_add(st_stats, "avg", NULL, 1, USEC_PER_MS, RRD_ALGORITHM_ABSOLUTE);
508
+ rd_max = rrddim_add(st_stats, "max", NULL, 1, USEC_PER_MS, RRD_ALGORITHM_ABSOLUTE);
509
+ }
510
+
511
+ rrddim_set_by_pointer(st_stats, rd_min, (collected_number)min_us);
512
+ rrddim_set_by_pointer(st_stats, rd_avg, (collected_number)avg_us);
513
+ rrddim_set_by_pointer(st_stats, rd_max, (collected_number)max_us);
514
+ rrdset_done(st_stats);
515
}
516
}