@cryptotaxi247 / netdata-1 / commits / 87a0559ec

Improving the ACLK performance - initial changes (#8399)

Add an inspection point for VerneMQ in the local dev env. Remove the bottleneck in sending websocket messages, at the expense of increased CPU-load. Fixed the message encoding. Added support for stress testing - it is still enabled in the main loop so will fire stress-testing payloads when the ACLK is established. Next patch will integrate the socket polling properly to reduce the CPU overhead and remove the stress testing payloads.

Andrew Moss committed Mar 14, 2020 at 07:35 UTC 87a0559ec3f885ff64764ab1d48bb1cb52ce0ebe

9 files changed +192 -31

.codacy.yml

@@ -13,3 +13,4 @@ exclude_paths:

13 - web/gui/src/**

14 - web/gui/main.js

15 - tests/**

16 + - aclk/tests/**

aclk/aclk_lws_wss_client.c

+72 -10

@@ -10,14 +10,57 @@ struct aclk_lws_wss_perconnect_data {

10 int todo;

11 };

13 -struct lws_wss_packet_buffer {

14 - unsigned char *data;

15 - size_t data_size;

16 - struct lws_wss_packet_buffer *next;

17 -};

18 -

13 static struct aclk_lws_wss_engine_instance *engine_instance = NULL;

15 +void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len)

16 +{

17 + if (write_len != NULL && write_len_bytes != NULL)

18 + {

19 + *write_len = 0;

20 + *write_len_bytes = 0;

21 + if (engine_instance != NULL)

22 + {

23 + aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);

24 +

25 + struct lws_wss_packet_buffer *write_b;

26 + size_t w,wb;

27 + for(w=0, wb=0, write_b = engine_instance->write_buffer_head; write_b != NULL; write_b = write_b->next)

28 + {

29 + w++;

30 + wb += write_b->data_size;

31 + }

32 + *write_len = w;

33 + *write_len_bytes = wb;

34 + aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);

35 + }

36 + }

37 + else if (write_len != NULL)

38 + {

39 + *write_len = 0;

40 + if (engine_instance != NULL)

41 + {

42 + aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);

43 +

44 + struct lws_wss_packet_buffer *write_b;

45 + size_t w;

46 + for(w=0, write_b = engine_instance->write_buffer_head; write_b != NULL; write_b = write_b->next)

47 + w++;

48 + *write_len = w;

49 + aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);

50 + }

51 + }

52 + if (read_len != NULL)

53 + {

54 + *read_len = 0;

55 + if (engine_instance != NULL)

56 + {

57 + aclk_lws_mutex_lock(&engine_instance->read_buf_mutex);

58 + *read_len = lws_ring_get_count_waiting_elements(engine_instance->read_ringbuffer, NULL);

59 + aclk_lws_mutex_unlock(&engine_instance->read_buf_mutex);

60 + }

61 + }

62 +}

63 +

64 static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void *data, size_t size)

65 {

66 struct lws_wss_packet_buffer *new = callocz(1, sizeof(struct lws_wss_packet_buffer));

@@ -25,6 +68,7 @@ static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void *data

68 new->data = mallocz(LWS_PRE + size);

69 memcpy(new->data + LWS_PRE, data, size);

70 new->data_size = size;

71 + new->written = 0;

72 }

73 return new;

74 }

@@ -355,7 +399,7 @@ static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reas

399 {

400 UNUSED(user);

401 struct lws_wss_packet_buffer *data;

358 - int retval = 0;

402 + int retval = 0, rc;

403

404 // Callback servicing is forced when we are closed from above.

405 if (engine_instance->upstream_reconnect_request) {

@@ -372,12 +416,24 @@ static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reas

416 switch (reason) {

417 case LWS_CALLBACK_CLIENT_WRITEABLE:

418 aclk_lws_mutex_lock(&engine_instance->write_buf_mutex);

375 - data = lws_wss_packet_buffer_pop(&engine_instance->write_buffer_head);

419 + data = engine_instance->write_buffer_head;

420 if (likely(data)) {

377 - lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);

378 - lws_wss_packet_buffer_free(data);

421 + size_t bytes_left = data->data_size - data->written;

422 + if ( bytes_left > 65536 )

423 + bytes_left = 65536;

424 + rc = lws_write(wsi, data->data + LWS_PRE + data->written, bytes_left, LWS_WRITE_BINARY);

425 + error("lws_write(req=%u,written=%u) %zu of %zu",bytes_left, rc, data->written,data->data_size,rc);

426 + data->written += bytes_left;

427 + if (data->written == data->data_size)

428 + {

429 + lws_wss_packet_buffer_pop(&engine_instance->write_buffer_head);

430 + lws_wss_packet_buffer_free(data);

431 + }

432 if (engine_instance->write_buffer_head)

433 + {

434 + error("Req write");

435 lws_callback_on_writable(engine_instance->lws_wsi);

436 + }

437 }

438 aclk_lws_mutex_unlock(&engine_instance->write_buf_mutex);

439 return retval;

@@ -487,7 +543,13 @@ abort:

543 void aclk_lws_wss_service_loop()

544 {

545 if (engine_instance)

546 + {

547 + if (engine_instance->lws_wsi) {

548 + lws_cancel_service(engine_instance->lws_context);

549 + lws_callback_on_writable(engine_instance->lws_wsi);

550 + }

551 lws_service(engine_instance->lws_context, 0);

552 + }

553 }

554

555 // in case the MQTT connection disconnect while lws transport is still operational

aclk/aclk_lws_wss_client.h

+6 -1

@@ -31,7 +31,11 @@ struct aclk_lws_wss_engine_callbacks {

31 void (*connection_closed)();

32 };

34 -struct lws_wss_packet_buffer;

34 +struct lws_wss_packet_buffer {

35 + unsigned char *data;

36 + size_t data_size, written;

37 + struct lws_wss_packet_buffer *next;

38 +};

40 struct aclk_lws_wss_engine_instance {

41 //target host/port for connection

@@ -73,6 +77,7 @@ void aclk_lws_wss_mqtt_layer_disconect_notif();

77 void aclk_lws_connection_established();

78 void aclk_lws_connection_data_received();

79 void aclk_lws_connection_closed();

80 +void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);

83 #endif

aclk/agent_cloud_link.c

+35 -17

@@ -1281,7 +1281,7 @@ void aclk_get_challenge(char *aclk_hostname, char *aclk_port)

1281 return;

1282 }

1283 if (challenge.result == NULL ) {

1284 - error("Could not retrieve challenge from auth response");

1284 + error("Could not retrieve challenge from auth response: %s", payload);

1285 return;

1286 }

1287

@@ -1362,6 +1362,7 @@ static void aclk_try_to_connect(char *hostname, char *port, int port_num)

1362 *

1363 * @return It always returns NULL

1364 */

1365 +void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);

1366 void *aclk_main(void *ptr)

1367 {

1368 struct netdata_static_thread *query_thread;

@@ -1410,9 +1411,10 @@ void *aclk_main(void *ptr)

1411

1412 while (!netdata_exit) {

1413 static int first_init = 0;

1413 -

1414 - info("loop state first_init_%d connected=%d connecting=%d", first_init, aclk_connected, aclk_connecting);

1415 - sleep_usec(USEC_PER_MS * 500);

1414 + size_t write_q, write_q_bytes, read_q;

1415 + lws_wss_check_queues(&write_q, &write_q_bytes, &read_q);

1416 + info("loop state first_init_%d connected=%d connecting=%d wq=%zu (%zu-bytes) rq=%zu",

1417 + first_init, aclk_connected, aclk_connecting, write_q, read_q);

1418 if (unlikely(!aclk_connected)) {

1419 if (unlikely(!first_init)) {

1420 aclk_try_to_connect(aclk_hostname, aclk_port, port_num);

@@ -1439,7 +1441,10 @@ void *aclk_main(void *ptr)

1441 }

1442

1443 _link_event_loop();

1442 - sleep_usec(USEC_PER_MS * 100);

1444 + //sleep_usec(USEC_PER_MS * 50);

1445 + static int stress_counter = 0;

1446 + if (stress_counter++ % 100 == 0 && write_q==0)

1447 + aclk_send_stress_test(2000000);

1448

1449 // TODO: Move to on-connect

1450 if (unlikely(!aclk_subscribed)) {

@@ -1498,7 +1503,7 @@ int aclk_send_message(char *sub_topic, char *message, char *msg_id)

1503 }

1504

1505 ACLK_LOCK;

1501 - rc = _link_send_message(final_topic, message, &mid);

1506 + rc = _link_send_message(final_topic, (unsigned char *)message, &mid);

1507 // TODO: link the msg_id with the mid so we can trace it

1508 ACLK_UNLOCK;

1509

@@ -1603,8 +1608,6 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)

1608 debug(D_ACLK, "Sending v%d msgid [%s] type [%s] time [%ld]", ACLK_VERSION, msg_id, type, time_created);

1609 }

1610

1606 -//#define EYE_FRIENDLY

1607 -

1611 /*

1612 * Take a buffer, encode it and rewrite it

1613 *

@@ -1612,10 +1615,6 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id)

1615

1616 BUFFER *aclk_encode_response(BUFFER *contents)

1617 {

1615 -#ifdef EYE_FRIENDLY

1616 -

1617 - return contents;

1618 -#else

1618 char *tmp_buffer = mallocz(contents->len * 2);

1619 char *src, *dst;

1620

@@ -1652,7 +1651,6 @@ BUFFER *aclk_encode_response(BUFFER *contents)

1651

1652 freez(tmp_buffer);

1653 return contents;

1655 -#endif

1654 }

1655

1656 /*

@@ -1684,7 +1682,7 @@ void aclk_send_alarm_metadata()

1682 debug(D_ACLK, "Metadata %s with alarms_active has %zu bytes", msg_id, local_buffer->len);

1683

1684 buffer_sprintf(local_buffer, "\n}\n}");

1687 - aclk_send_message(ACLK_ALARMS_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);

1685 + aclk_send_message(ACLK_ALARMS_TOPIC, local_buffer->buffer, msg_id);

1686 debug(D_ACLK, "Metadata %s encoded has %zu bytes", msg_id, local_buffer->len);

1687

1688 freez(msg_id);

@@ -1711,7 +1709,7 @@ int aclk_send_info_metadata()

1709 buffer_sprintf(local_buffer, "\n}\n}");

1710 debug(D_ACLK, "Metadata %s with chart has %zu bytes", msg_id, local_buffer->len);

1711

1714 - aclk_send_message(ACLK_METADATA_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);

1712 + aclk_send_message(ACLK_METADATA_TOPIC, local_buffer->buffer, msg_id);

1713 debug(D_ACLK, "Metadata %s encoded has %zu bytes", msg_id, local_buffer->len);

1714 freez(msg_id);

1715

@@ -1719,10 +1717,30 @@ int aclk_send_info_metadata()

1717 return 0;

1718 }

1719

1720 +void aclk_send_stress_test(size_t size)

1721 +{

1722 + char *buffer = mallocz(size);

1723 + if (buffer != NULL)

1724 + {

1725 + for(size_t i=0; i<size; i++)

1726 + buffer[i] = 'x';

1727 + buffer[size-1] = 0;

1728 + time_t time_created = now_realtime_sec();

1729 + sprintf(buffer,"{\"type\":\"stress\", \"timestamp\":%ld,\"payload\":", time_created);

1730 + buffer[strlen(buffer)] = '"';

1731 + buffer[size-2] = '}';

1732 + buffer[size-3] = '"';

1733 + aclk_send_message(ACLK_METADATA_TOPIC, buffer, NULL);

1734 + error("Sending stress of size %zu at time %ld", size, time_created);

1735 + }

1736 + free(buffer);

1737 +}

1738 +

1739 // Send info metadata message to the cloud if the link is established

1740 // or on request

1741 int aclk_send_metadata()

1742 {

1743 +

1744 aclk_send_info_metadata();

1745 aclk_send_alarm_metadata();

1746

@@ -1774,7 +1792,7 @@ int aclk_send_single_chart(char *hostname, char *chart)

1792 rrdset2json(st, local_buffer, NULL, NULL, 1);

1793 buffer_sprintf(local_buffer, "\t\n}");

1794

1777 - aclk_send_message(ACLK_CHART_TOPIC, aclk_encode_response(local_buffer)->buffer, msg_id);

1795 + aclk_send_message(ACLK_CHART_TOPIC, local_buffer->buffer, msg_id);

1796

1797 freez(msg_id);

1798 buffer_free(local_buffer);

@@ -1833,7 +1851,7 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)

1851 netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);

1852

1853 buffer_sprintf(local_buffer, "\n}");

1836 - aclk_queue_query(ACLK_ALARMS_TOPIC, NULL, msg_id, aclk_encode_response(local_buffer)->buffer, 0, 1, ACLK_CMD_ALARM);

1854 + aclk_queue_query(ACLK_ALARMS_TOPIC, NULL, msg_id, local_buffer->buffer, 0, 1, ACLK_CMD_ALARM);

1855

1856 freez(msg_id);

1857 buffer_free(local_buffer);

aclk/mqtt.c

+2 -2

@@ -275,7 +275,7 @@ int _link_subscribe(char *topic, int qos)

275 *

276 */

277

278 -int _link_send_message(char *topic, char *message, int *mid)

278 +int _link_send_message(char *topic, unsigned char *message, int *mid)

279 {

280 int rc;

281

@@ -285,7 +285,7 @@ int _link_send_message(char *topic, char *message, int *mid)

285 return rc;

286

287 int msg_len = strlen(message);

288 -

288 + error("Sending MQTT len=%d starts %02x %02x %02x", msg_len, message[0], message[1], message[2]);

289 rc = mosquitto_publish(mosq, mid, topic, msg_len, message, ACLK_QOS, 0);

290

291 // TODO: Add better handling -- error will flood the logfile here

aclk/mqtt.h

+1 -1

@@ -14,7 +14,7 @@ int mqtt_attempt_connection(char *aclk_hostname, int aclk_port, char *username,

14 //int _link_lib_init();

15 int _mqtt_lib_init();

16 int _link_subscribe(char *topic, int qos);

17 -int _link_send_message(char *topic, char *message, int *mid);

17 +int _link_send_message(char *topic, unsigned char *message, int *mid);

18 const char *_link_strerror(int rc);

20 int aclk_handle_cloud_request(char *);

aclk/tests/launch-paho.sh new

@@ -0,0 +1,4 @@

1 +#!/usr/bin/env bash

2 +

3 +docker build -f paho.Dockerfile . --build-arg "HOST_HOSTNAME=$(ping -c1 "$(hostname).local" | head -n1 | grep -o '[0-9]*\.[0-9]*\.[0-9]*\.[0-9]*')" -t paho-client

4 +docker run -it paho-client

aclk/tests/paho-inspection.py new

+57

@@ -0,0 +1,57 @@

1 +import ssl

2 +import paho.mqtt.client as mqtt

3 +import json

4 +import time

5 +import sys

6 +

7 +def on_connect(mqttc, obj, flags, rc):

8 + if rc==0:

9 + print("Successful connection", flush=True)

10 + else :

11 + print(f"Connection error rc={rc}", flush=True)

12 + mqttc.subscribe("/agent/#",0)

13 +

14 +def on_disconnect(mqttc, obj, flags, rc):

15 + print("disconnected rc: "+str(rc), flush=True)

16 +

17 +def on_message(mqttc, obj, msg):

18 + print(f"{msg.topic} {len(msg.payload)}-bytes qos={msg.qos}", flush=True)

19 + try:

20 + print(f"Trying decode of {msg.payload[:60]}",flush=True)

21 + api_msg = json.loads(msg.payload)

22 + except Exception as e:

23 + print(e,flush=True)

24 + return

25 + ts = api_msg["timestamp"]

26 + mtype = api_msg["type"]

27 + print(f"Message {mtype} time={ts} size {len(api_msg)}", flush=True)

28 + now = time.time()

29 + print(f"Current {now} -> Delay {now-ts}", flush=True)

30 +

31 +def on_publish(mqttc, obj, mid):

32 + print("mid: "+str(mid), flush=True)

33 +

34 +def on_subscribe(mqttc, obj, mid, granted_qos):

35 + print("Subscribed: "+str(mid)+" "+str(granted_qos), flush=True)

36 +

37 +def on_log(mqttc, obj, level, string):

38 + print(string)

39 +

40 +print(f"Starting paho-inspection on {sys.argv[1]}", flush=True)

41 +mqttc = mqtt.Client(transport='websockets',client_id="paho")

42 +#mqttc.tls_set(certfile="server.crt", keyfile="server.key", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None)

43 +#mqttc.tls_set(ca_certs="server.crt", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS, ciphers=None)

44 +mqttc.tls_set(cert_reqs=ssl.CERT_NONE, tls_version=ssl.PROTOCOL_TLS, ciphers=None)

45 +mqttc.tls_insecure_set(True)

46 +mqttc.on_message = on_message

47 +mqttc.on_connect = on_connect

48 +mqttc.on_disconnect = on_disconnect

49 +mqttc.on_publish = on_publish

50 +mqttc.on_subscribe = on_subscribe

51 +mqttc.username_pw_set("paho","paho")

52 +mqttc.connect(sys.argv[1], 8443, 60)

53 +

54 +#mqttc.publish("/agent/mine","Test1")

55 +#mqttc.subscribe("$SYS/#", 0)

56 +print("Connected succesfully, monitoring /agent/#", flush=True)

57 +mqttc.loop_forever()

aclk/tests/paho.Dockerfile new

+14

@@ -0,0 +1,14 @@

1 +FROM archlinux/base:latest

2 +

3 +RUN pacman -Syyu --noconfirm

4 +RUN pacman --noconfirm --needed -S python-pip

5 +

6 +RUN pip install paho-mqtt

7 +

8 +RUN mkdir -p /opt/paho

9 +COPY paho-inspection.py /opt/paho/

10 +

11 +WORKDIR /opt/paho

12 +ARG HOST_HOSTNAME

13 +RUN echo $HOST_HOSTNAME >host

14 +CMD ["/bin/bash", "-c", "/usr/sbin/python paho-inspection.py $(cat host)"]