Fix the latency issue on the ACLK and suppress the diagnostics (#8992)
The on-connect payloads were large enough to trigger a massive increase in latency on the link and prevent chart updates due to head-of-line blocking. The default window detection in libwebsockets was under-reporting the size of the available window in the network. Overwritten with some sensible values. The large volume of ACLK per-message info-logging is not produced unless the agent is compiled with NETDATA_INTERNAL_CHECKS. The logging now includes latency measurements on the link. Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>
Andrew Moss committed
May 12, 2020 at 19:27 UTC
f20269d836f3eecfd67d566cd781618e8bc2e8ee
3 files changed
+26
-6
aclk/aclk_lws_wss_client.c
+1
-1
@@ -132,7 +132,7 @@ static inline void aclk_lws_wss_clear_io_buffers()
132
}
133
134
static const struct lws_protocols protocols[] = { { "aclk-wss", aclk_lws_wss_callback,
135
- sizeof(struct aclk_lws_wss_perconnect_data), 0, 0, 0, 0 },
135
+ sizeof(struct aclk_lws_wss_perconnect_data), 32768*4, 0, 0, 32768*4 },
136
{ NULL, NULL, 0, 0, 0, 0, 0 } };
137
138
static void aclk_lws_wss_log_divert(int level, const char *line)
aclk/agent_cloud_link.c
-2
@@ -1412,8 +1412,6 @@ void *aclk_main(void *ptr)
1412
aclk_lws_wss_destroy_context();
1413
aclk_force_reconnect = 0;
1414
}
1415
- //info("loop state first_init_%d connected=%d connecting=%d wq=%zu (%zu-bytes) rq=%zu",
1416
- // first_init, aclk_connected, aclk_connecting, write_q, write_q_bytes, read_q);
1415
if (unlikely(!netdata_exit && !aclk_connected && !aclk_force_reconnect)) {
1416
if (unlikely(!first_init)) {
1417
aclk_try_to_connect(aclk_hostname, aclk_port, port_num);
aclk/mqtt.c
+25
-3
@@ -13,6 +13,10 @@ inline const char *_link_strerror(int rc)
13
return mosquitto_strerror(rc);
14
}
15
16
+#ifdef NETDATA_INTERNAL_CHECKS
17
+static struct timeval sendTimes[1024];
18
+#endif
19
+
20
static struct mosquitto *mosq = NULL;
21
22
@@ -29,8 +33,14 @@ void publish_callback(struct mosquitto *mosq, void *obj, int rc)
33
UNUSED(mosq);
34
UNUSED(obj);
35
UNUSED(rc);
32
- info("Publish_callback: mid=%d", rc);
33
- // TODO: link this with a msg_id so it can be traced
36
+#ifdef NETDATA_INTERNAL_CHECKS
37
+ struct timeval now, *orig;
38
+ now_realtime_timeval(&now);
39
+ orig = &sendTimes[ rc & 0x3ff ];
40
+ int64_t diff = (now.tv_sec - orig->tv_sec) * USEC_PER_SEC + (now.tv_usec - orig->tv_usec);
41
+
42
+ info("Publish_callback: mid=%d latency=%" PRId64 "ms", rc, diff / 1000);
43
+#endif
44
return;
45
}
46
@@ -320,6 +330,7 @@ int _link_subscribe(char *topic, int qos)
330
int _link_send_message(char *topic, unsigned char *message, int *mid)
331
{
332
int rc;
333
+ size_t write_q, write_q_bytes, read_q;
334
335
rc = mosquitto_pub_topic_check(topic);
336
@@ -327,9 +338,20 @@ int _link_send_message(char *topic, unsigned char *message, int *mid)
338
return rc;
339
340
int msg_len = strlen((char*)message);
330
- info("Sending MQTT len=%d starts %02x %02x %02x", msg_len, message[0], message[1], message[2]);
341
+ lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);
342
rc = mosquitto_publish(mosq, mid, topic, msg_len, message, ACLK_QOS, 0);
343
344
+#ifdef NETDATA_INTERNAL_CHECKS
345
+ char msg_head[64];
346
+ memset(msg_head, 0, sizeof(msg_head));
347
+ strncpy(msg_head, (char*)message, 60);
348
+ for (size_t i = 0; i < sizeof(msg_head); i++)
349
+ if(msg_head[i] == '\n') msg_head[i] = ' ';
350
+ info("Sending MQTT len=%d mid=%d wq=%zu (%zu-bytes) readq=%zu: %s", msg_len,
351
+ *mid, write_q, write_q_bytes, read_q, msg_head);
352
+ now_realtime_timeval(&sendTimes[ *mid & 0x3ff ]);
353
+#endif
354
+
355
// TODO: Add better handling -- error will flood the logfile here
356
if (unlikely(rc != MOSQ_ERR_SUCCESS)) {
357
errno = 0;