@cryptotaxi247 / netdata / commits / e6cb18b0c

Add publish latency to the aclk-state command (#20769)

add mqtt publish latency report to aclk-state cli command

Stelios Fragkakis committed Aug 6, 2025 at 18:15 UTC e6cb18b0c222ba10f9817b705f0f49fad7c2892b

4 files changed +23 -8

src/aclk/aclk.c

+11 -1

@@ -1061,6 +1061,9 @@ static void fill_alert_status_for_host(BUFFER *wb, RRDHOST *host)

1061 aclk_host_config->snapshot_count);

1062 }

1063

1064 +

1065 +extern usec_t publish_latency;

1066 +

1067 char *aclk_state(void)

1068 {

1069 BUFFER *wb = buffer_create(1024, &netdata_buffers_statistics.buffers_aclk);

@@ -1080,7 +1083,10 @@ char *aclk_state(void)

1083 else {

1084 const char *cloud_base_url = cloud_config_url_get();

1085 char *aclk_proxy = (char *)aclk_get_proxy(NULL, true);

1083 - buffer_sprintf(wb, "Yes\nClaimed Id: %s\nCloud URL: %s\nACLK Proxy: %s\n", claim_id.str, cloud_base_url ? cloud_base_url : "null", aclk_proxy ? aclk_proxy : "none");

1086 + usec_t latency = __atomic_load_n(&publish_latency, __ATOMIC_RELAXED);

1087 + char latency_str[64];

1088 + duration_snprintf(latency_str, sizeof(latency_str), (int64_t) latency, "us", true);

1089 + buffer_sprintf(wb, "Yes\nClaimed Id: %s\nCloud URL: %s\nACLK Proxy: %s\nPublish Latency: %s\n", claim_id.str, cloud_base_url ? cloud_base_url : "null", aclk_proxy ? aclk_proxy : "none", latency_str);

1090 }

1091

1092 buffer_sprintf(wb, "Online: %s\nReconnect count: %d\nBanned By Cloud: %s\n", aclk_online() ? "Yes" : "No", aclk_connection_counter > 0 ? (aclk_connection_counter - 1) : 0, aclk_disable_runtime ? "Yes" : "No");

@@ -1211,6 +1217,10 @@ char *aclk_state_json(void)

1217 tmp = aclk_proxy ? json_object_new_string(aclk_proxy) : NULL;

1218 json_object_object_add(msg, "aclk_proxy", tmp);

1219

1220 + usec_t latency = __atomic_load_n(&publish_latency, __ATOMIC_RELAXED);

1221 + tmp =json_object_new_int64((int64_t) latency);

1222 + json_object_object_add(msg, "publish_latency_us", tmp);

1223 +

1224 tmp = json_object_new_boolean(aclk_online());

1225 json_object_object_add(msg, "online", tmp);

1226

src/aclk/mqtt_websockets/mqtt_ng.c

+7 -2

@@ -5,7 +5,7 @@

5 #endif

7 #include "libnetdata/libnetdata.h"

8 -void pulse_aclk_sent_message_acked(usec_t usec, size_t len);

8 +void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len);

10 #include "common_internal.h"

11 #include "mqtt_constants.h"

@@ -251,6 +251,8 @@ struct mqtt_ng_client {

251 size_t max_msg_size;

252 };

253

254 +usec_t publish_latency;

255 +

256 unsigned char pingreq[] = { MQTT_CPT_PINGREQ << 4, 0x00 };

257

258 struct buffer_fragment ping_frag = {

@@ -622,6 +624,7 @@ struct mqtt_ng_client *mqtt_ng_init(struct mqtt_ng_init *settings)

624 client->msg_callback = settings->msg_callback;

625 spinlock_init(&client->pending_packets.spinlock);

626 client->pending_packets.JudyL = NULL;

627 + __atomic_store_n(&publish_latency, 0, __ATOMIC_RELEASE);

628

629 return client;

630 }

@@ -1165,7 +1168,9 @@ static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)

1168 UNLOCK_HDR_BUFFER(&client->main_buffer);

1169 return 1;

1170 }

1168 - pulse_aclk_sent_message_acked(frag->sent_monotonic_ut, frag->len);

1171 + usec_t latency = now_monotonic_usec() - frag->sent_monotonic_ut;

1172 + pulse_aclk_sent_message_acked(latency, frag->len);

1173 + __atomic_store_n(&publish_latency, latency, __ATOMIC_RELEASE);

1174 mark_message_for_gc(frag);

1175

1176 size_t used = BUFFER_BYTES_USED(&client->main_buffer.hdr_buffer);

src/daemon/pulse/pulse-network.c

+4 -4

@@ -92,12 +92,12 @@ static inline size_t aclk_time_histogram_slot(struct aclk_time_histogram *h, use

92 return low - 1;

93 }

95 -void pulse_aclk_sent_message_acked(usec_t sent_ut, size_t len __maybe_unused) {

96 - if(!sent_ut) return;

95 +void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len __maybe_unused) {

96 + if(!publish_latency) return;

98 - usec_t usec = now_monotonic_usec() - sent_ut;

98 +// usec_t usec = now_monotonic_usec() - publish_latency;

100 - size_t slot = aclk_time_histogram_slot(&aclk_time_heatmap, usec);

100 + size_t slot = aclk_time_histogram_slot(&aclk_time_heatmap, publish_latency);

101 internal_fatal(slot >= _countof(aclk_time_heatmap.array), "hey!");

102

103 __atomic_add_fetch(&aclk_time_heatmap.array[slot].count, 1, __ATOMIC_RELAXED);

src/daemon/pulse/pulse-network.h

+1 -1

@@ -16,7 +16,7 @@ void pulse_statsd_sent_bytes(size_t bytes);

16 void pulse_stream_received_bytes(size_t bytes);

17 void pulse_stream_sent_bytes(size_t bytes);

19 -void pulse_aclk_sent_message_acked(usec_t usec, size_t len);

19 +void pulse_aclk_sent_message_acked(usec_t publish_latency, size_t len);

21 #ifdef PULSE_INTERNALS

22 void aclk_time_histogram_init(void);