@cryptotaxi247 / netdata-1 / commits / 6c41244db

add more aclk worker jobs (#19435)

* add more aclk worker jobs * garbage collect aclk buffer, when receiving ackowledgements * add msg callback worker job * added latency histogram for aclk pub-ack * fix compilation * Add a few workers, handle context checkpoint async * Proper register worker --------- Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 18, 2025 at 21:14 UTC 6c41244db9557c96b4813a28c21fcd4e2a044127

12 files changed +366 -29

src/aclk/aclk.c

+11

@@ -799,6 +799,17 @@ void *aclk_main(void *ptr)

799 worker_register_job_name(WORKER_ACLK_HANDLE_MQTT_INTERNAL, "mqtt internal");

800 worker_register_job_name(WORKER_ACLK_TX, "tx");

801 worker_register_job_name(WORKER_ACLK_TX_ERROR, "tx error");

802 + worker_register_job_name(WORKER_ACLK_TRY_SEND_ALL, "try send all");

803 + worker_register_job_name(WORKER_ACLK_HANDLE_INCOMING, "handle incoming");

804 + worker_register_job_name(WORKER_ACLK_CPT_CONNACK, "cpt connack");

805 + worker_register_job_name(WORKER_ACLK_CPT_PUBACK, "cpt puback");

806 + worker_register_job_name(WORKER_ACLK_CPT_PINGRESP, "cpt pingresp");

807 + worker_register_job_name(WORKER_ACLK_CPT_SUBACK, "cpt suback");

808 + worker_register_job_name(WORKER_ACLK_CPT_PUBLISH, "cpt publish");

809 + worker_register_job_name(WORKER_ACLK_CPT_DISCONNECT, "cpt disconnect");

810 + worker_register_job_name(WORKER_ACLK_CPT_UNKNOWN, "cpt unknown");

811 + worker_register_job_name(WORKER_ACLK_SEND_FRAGMENT, "send fragment");

812 + worker_register_job_name(WORKER_ACLK_MSG_CALLBACK, "msg callback");

813

814 ACLK_PROXY_TYPE proxy_type;

815 aclk_get_proxy(&proxy_type);

src/aclk/aclk_contexts_api.c

+2 -2

@@ -6,7 +6,7 @@

7 void aclk_send_contexts_snapshot(contexts_snapshot_t data)

8 {

9 - aclk_query_t query = aclk_query_new(PROTO_BIN_MESSAGE);

9 + aclk_query_t query = aclk_query_new(CTX_SEND_SNAPSHOT);

10 query->data.bin_payload.topic = ACLK_TOPICID_CTXS_SNAPSHOT;

11 query->data.bin_payload.payload = contexts_snapshot_2bin(data, &query->data.bin_payload.size);

12 query->data.bin_payload.msg_name = "ContextsSnapshot";

@@ -15,7 +15,7 @@ void aclk_send_contexts_snapshot(contexts_snapshot_t data)

16 void aclk_send_contexts_updated(contexts_updated_t data)

17 {

18 - aclk_query_t query = aclk_query_new(PROTO_BIN_MESSAGE);

18 + aclk_query_t query = aclk_query_new(CTX_SEND_SNAPSHOT_UPD);

19 query->data.bin_payload.topic = ACLK_TOPICID_CTXS_UPDATED;

20 query->data.bin_payload.payload = contexts_updated_2bin(data, &query->data.bin_payload.size);

21 query->data.bin_payload.msg_name = "ContextsUpdated";

src/aclk/aclk_query_queue.h

+7 -2

@@ -18,8 +18,12 @@ typedef enum {

18 ALARM_PROVIDE_CFG,

19 ALARM_SNAPSHOT,

20 UPDATE_NODE_COLLECTORS,

21 - PROTO_BIN_MESSAGE,

22 - ACLK_QUERY_TYPE_COUNT // always keep this as last

21 + CTX_SEND_SNAPSHOT, // Context snapshot to the cloud

22 + CTX_SEND_SNAPSHOT_UPD, // Context incremental update to the cloud

23 + CTX_CHECKPOINT, // Context checkpoint from the cloud

24 + CTX_STOP_STREAMING, // Context stop streaming

25 + CREATE_NODE_INSTANCE, // Create node instance on the agent

26 + ACLK_QUERY_TYPE_COUNT // always keep this as last

27 } aclk_query_type_t;

29 struct aclk_query_http_api_v2 {

@@ -56,6 +60,7 @@ struct aclk_query {

60 union {

61 struct aclk_query_http_api_v2 http_api_v2;

62 struct aclk_bin_payload bin_payload;

63 + void *payload;

64 } data;

65 };

src/aclk/aclk_rx_msgs.c

+7 -10

@@ -6,6 +6,7 @@

6 #include "aclk.h"

7 #include "aclk_capas.h"

8 #include "aclk_query.h"

9 +#include "mqtt_websockets/aclk_mqtt_workers.h"

11 #include "schema-wrappers/proto_2_json.h"

@@ -417,11 +418,9 @@ int contexts_checkpoint(const char *msg, size_t msg_len)

418 if (!cmd)

419 return 1;

420

420 - rrdcontext_hub_checkpoint_command(cmd);

421 -

422 - freez(cmd->claim_id);

423 - freez(cmd->node_id);

424 - freez(cmd);

421 + aclk_query_t query = aclk_query_new(CTX_CHECKPOINT);

422 + query->data.payload = cmd;

423 + aclk_execute_query(query);

424 return 0;

425 }

426

@@ -436,11 +435,9 @@ int stop_streaming_contexts(const char *msg, size_t msg_len)

435 if (!cmd)

436 return 1;

437

439 - rrdcontext_hub_stop_streaming_command(cmd);

440 -

441 - freez(cmd->claim_id);

442 - freez(cmd->node_id);

443 - freez(cmd);

438 + aclk_query_t query = aclk_query_new(CTX_STOP_STREAMING);

439 + query->data.payload = cmd;

440 + aclk_execute_query(query);

441 return 0;

442 }

443

src/aclk/mqtt_websockets/aclk_mqtt_workers.h

+11

@@ -26,5 +26,16 @@

26 #define WORKER_ACLK_HANDLE_MQTT_INTERNAL 20

27 #define WORKER_ACLK_TX 21

28 #define WORKER_ACLK_TX_ERROR 22

29 +#define WORKER_ACLK_TRY_SEND_ALL 23

30 +#define WORKER_ACLK_HANDLE_INCOMING 24

31 +#define WORKER_ACLK_CPT_CONNACK 25

32 +#define WORKER_ACLK_CPT_PUBACK 26

33 +#define WORKER_ACLK_CPT_PINGRESP 27

34 +#define WORKER_ACLK_CPT_SUBACK 28

35 +#define WORKER_ACLK_CPT_PUBLISH 29

36 +#define WORKER_ACLK_CPT_DISCONNECT 30

37 +#define WORKER_ACLK_CPT_UNKNOWN 31

38 +#define WORKER_ACLK_SEND_FRAGMENT 32

39 +#define WORKER_ACLK_MSG_CALLBACK 33

41 #endif //NETDATA_ACLK_MQTT_WORKERS_H

src/aclk/mqtt_websockets/mqtt_ng.c

+42 -6

@@ -5,10 +5,12 @@

5 #endif

7 #include "libnetdata/libnetdata.h"

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

10 #include "common_internal.h"

11 #include "mqtt_constants.h"

12 #include "mqtt_ng.h"

13 +#include "aclk_mqtt_workers.h"

15 #define SMALL_STRING_DONT_FRAGMENT_LIMIT 128

@@ -31,14 +33,13 @@

34 typedef uint16_t buffer_frag_flag_t;

35 struct buffer_fragment {

34 - size_t len;

35 - size_t sent;

36 + uint32_t len;

37 + uint32_t sent;

38 buffer_frag_flag_t flags;

39 + uint16_t packet_id;

40 void (*free_fnc)(void *ptr);

41 unsigned char *data;

39 -

40 - uint16_t packet_id;

41 -

42 + usec_t sent_monotonic_ut;

43 struct buffer_fragment *next;

44 };

@@ -1851,6 +1852,7 @@ static int mqtt_ng_next_to_send(struct mqtt_ng_client *client) {

1852 if ( client->ping_pending && (!frag || (frag->flags & BUFFER_FRAG_MQTT_PACKET_HEAD && frag->sent == 0)) ) {

1853 client->ping_pending = 0;

1854 ping_frag.sent = 0;

1855 + ping_frag.sent_monotonic_ut = 0;

1856 client->main_buffer.sending_frag = &ping_frag;

1857 return 0;

1858 }

@@ -1865,6 +1867,8 @@ static int mqtt_ng_next_to_send(struct mqtt_ng_client *client) {

1867 // nothing could be written anymore

1868 // return 1 if last fragment of a message was fully sent

1869 static int send_fragment(struct mqtt_ng_client *client) {

1870 + worker_is_busy(WORKER_ACLK_SEND_FRAGMENT);

1871 +

1872 struct buffer_fragment *frag = client->main_buffer.sending_frag;

1873

1874 // for readability

@@ -1878,6 +1882,7 @@ static int send_fragment(struct mqtt_ng_client *client) {

1882 else

1883 nd_log(NDLS_DAEMON, NDLP_WARNING, "This fragment was fully sent already. This should not happen!");

1884

1885 + frag->sent_monotonic_ut = now_monotonic_usec();

1886 frag->sent += processed;

1887 if (frag->sent != frag->len)

1888 return -1;

@@ -1925,6 +1930,7 @@ static void mark_message_for_gc(struct buffer_fragment *frag)

1930

1931 static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)

1932 {

1933 + size_t reclaimable = 0;

1934 LOCK_HDR_BUFFER(&client->main_buffer);

1935 struct buffer_fragment *frag = BUFFER_FIRST_FRAG(&client->main_buffer.hdr_buffer);

1936 while (frag) {

@@ -1934,10 +1940,20 @@ static int mark_packet_acked(struct mqtt_ng_client *client, uint16_t packet_id)

1940 UNLOCK_HDR_BUFFER(&client->main_buffer);

1941 return 1;

1942 }

1943 + pulse_aclk_sent_message_acked(frag->sent_monotonic_ut, frag->len);

1944 mark_message_for_gc(frag);

1945 +

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

1947 + if (reclaimable >= (used / 4))

1948 + transaction_buffer_garbage_collect(&client->main_buffer);

1949 +

1950 UNLOCK_HDR_BUFFER(&client->main_buffer);

1951 return 0;

1952 }

1953 +

1954 + if(frag_is_marked_for_gc(frag))

1955 + reclaimable += FRAG_SIZE_IN_BUFFER(frag);

1956 +

1957 frag = frag->next;

1958 }

1959 nd_log(NDLS_DAEMON, NDLP_ERR, "Received packet_id (%" PRIu16 ") is unknown!", packet_id);

@@ -1963,6 +1979,8 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

1979 uint8_t ctrl_packet_type = get_control_packet_type(client->parser.mqtt_control_packet_type);

1980 switch (ctrl_packet_type) {

1981 case MQTT_CPT_CONNACK:

1982 + worker_is_busy(WORKER_ACLK_CPT_CONNACK);

1983 +

1984 LOCK_HDR_BUFFER(&client->main_buffer);

1985 mark_message_for_gc(client->connect_msg);

1986 UNLOCK_HDR_BUFFER(&client->main_buffer);

@@ -1991,6 +2009,8 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2009 return MQTT_NG_CLIENT_SERVER_RETURNED_ERROR;

2010

2011 case MQTT_CPT_PUBACK:

2012 + worker_is_busy(WORKER_ACLK_CPT_PUBACK);

2013 +

2014 if (mark_packet_acked(client, client->parser.mqtt_packet.puback.packet_id))

2015 return MQTT_NG_CLIENT_PROTOCOL_ERROR;

2016 if (client->puback_callback)

@@ -1998,14 +2018,18 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2018 break;

2019

2020 case MQTT_CPT_PINGRESP:

2021 + worker_is_busy(WORKER_ACLK_CPT_PINGRESP);

2022 + pulse_aclk_sent_message_acked(ping_frag.sent_monotonic_ut, ping_frag.len);

2023 break;

2024

2025 case MQTT_CPT_SUBACK:

2026 + worker_is_busy(WORKER_ACLK_CPT_SUBACK);

2027 if (mark_packet_acked(client, client->parser.mqtt_packet.suback.packet_id))

2028 return MQTT_NG_CLIENT_PROTOCOL_ERROR;

2029 break;

2030

2031 case MQTT_CPT_PUBLISH:

2032 + worker_is_busy(WORKER_ACLK_CPT_PUBLISH);

2033 pub = &client->parser.mqtt_packet.publish;

2034

2035 if (pub->qos > 1) {

@@ -2038,8 +2062,11 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2062 }

2063 }

2064

2041 - if (client->msg_callback)

2065 + if (client->msg_callback) {

2066 + worker_is_busy(WORKER_ACLK_MSG_CALLBACK);

2067 client->msg_callback(pub->topic, pub->data, pub->data_len, pub->qos);

2068 + }

2069 +

2070 // in case we have property topic alias and we have topic we take over the string

2071 // and add pointer to it into topic alias list

2072 if (prop == NULL)

@@ -2048,11 +2075,13 @@ int handle_incoming_traffic(struct mqtt_ng_client *client)

2075 return MQTT_NG_CLIENT_WANT_WRITE;

2076

2077 case MQTT_CPT_DISCONNECT:

2078 + worker_is_busy(WORKER_ACLK_CPT_DISCONNECT);

2079 nd_log(NDLS_DAEMON, NDLP_INFO, "Got MQTT DISCONNECT control packet from server. Reason code: %d", (int)client->parser.mqtt_packet.disconnect.reason_code);

2080 client->client_state = MQTT_STATE_DISCONNECTED;

2081 break;

2082

2083 default:

2084 + worker_is_busy(WORKER_ACLK_CPT_UNKNOWN);

2085 nd_log(NDLS_DAEMON, NDLP_INFO, "Got unknown control packet %u from server", ctrl_packet_type);

2086 break;

2087 }

@@ -2068,19 +2097,26 @@ int mqtt_ng_sync(struct mqtt_ng_client *client)

2097 if (client->client_state == MQTT_STATE_ERROR)

2098 return 1;

2099

2100 + worker_is_busy(WORKER_ACLK_TRY_SEND_ALL);

2101 +

2102 LOCK_HDR_BUFFER(&client->main_buffer);

2103 try_send_all(client);

2104 UNLOCK_HDR_BUFFER(&client->main_buffer);

2105

2106 int rc;

2107

2108 + worker_is_busy(WORKER_ACLK_HANDLE_INCOMING);

2109 while ((rc = handle_incoming_traffic(client)) != MQTT_NG_CLIENT_NEED_MORE_BYTES) {

2110 if (rc < 0)

2111 break;

2112 if (rc == MQTT_NG_CLIENT_WANT_WRITE) {

2113 + worker_is_busy(WORKER_ACLK_TRY_SEND_ALL);

2114 +

2115 LOCK_HDR_BUFFER(&client->main_buffer);

2116 try_send_all(client);

2117 UNLOCK_HDR_BUFFER(&client->main_buffer);

2118 +

2119 + worker_is_busy(WORKER_ACLK_HANDLE_INCOMING);

2120 }

2121 }

2122

src/daemon/libuv_workers.c

+11

@@ -60,6 +60,17 @@ void register_libuv_worker_jobs() {

60 worker_register_job_name(UV_EVENT_ACLK_NODE_INFO, "aclk host node info");

61 worker_register_job_name(UV_EVENT_ACLK_ALERT_PUSH, "aclk alert push");

62 worker_register_job_name(UV_EVENT_ACLK_QUERY_EXECUTE, "aclk query execute");

63 + // aclk

64 + worker_register_job_name(UV_EVENT_CTX_STOP_STREAMING, "ctx stop streaming");

65 + worker_register_job_name(UV_EVENT_CTX_CHECKPOINT, "ctx version check");

66 + worker_register_job_name(UV_EVENT_ALARM_PROVIDE_CFG, "send alarm config");

67 + worker_register_job_name(UV_EVENT_ALARM_SNAPSHOT, "alert snapshot");

68 + worker_register_job_name(UV_EVENT_REGISTER_NODE, "register node");

69 + worker_register_job_name(UV_EVENT_UPDATE_NODE_COLLECTORS, "update collectors");

70 + worker_register_job_name(UV_EVENT_UPDATE_NODE_INFO, "send node info");

71 + worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT, "ctx send snapshot");

72 + worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT_UPD, "ctx send update");

73 + worker_register_job_name(UV_EVENT_NODE_STATE_UPDATE, "node state update");

75 // netdatacli

76 worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");

src/daemon/libuv_workers.h

+12

@@ -53,6 +53,18 @@ enum event_loop_job {

53 UV_EVENT_ACLK_ALERT_PUSH,

54 UV_EVENT_ACLK_QUERY_EXECUTE,

56 + //

57 + UV_EVENT_CTX_STOP_STREAMING,

58 + UV_EVENT_CTX_CHECKPOINT,

59 + UV_EVENT_ALARM_PROVIDE_CFG,

60 + UV_EVENT_ALARM_SNAPSHOT,

61 + UV_EVENT_REGISTER_NODE,

62 + UV_EVENT_UPDATE_NODE_COLLECTORS,

63 + UV_EVENT_UPDATE_NODE_INFO,

64 + UV_EVENT_CTX_SEND_SNAPSHOT,

65 + UV_EVENT_CTX_SEND_SNAPSHOT_UPD,

66 + UV_EVENT_NODE_STATE_UPDATE,

67 +

68 // netdatacli

69 UV_EVENT_SCHEDULE_CMD,

70 };

src/daemon/pulse/pulse-network.c

+184

@@ -19,6 +19,130 @@ static struct network_statistics {

19 PAD64(uint64_t) stream_bytes_sent;

20 } live_stats = { 0 };

22 +// --------------------------------------------------------------------------------------------------------------------

23 +// aclk time heatmap

24 +// a similar history exists in dbengine cache.c

25 +

26 +struct aclk_histogram_entry {

27 + usec_t upto;

28 + size_t count;

29 +};

30 +

31 +#define ACLK_TIME_HISTOGRAM_ENTRIES 19

32 +

33 +static struct aclk_time_histogram {

34 + struct aclk_histogram_entry array[ACLK_TIME_HISTOGRAM_ENTRIES];

35 +} aclk_time_heatmap;

36 +

37 +void aclk_time_histogram_init(void) {

38 + struct aclk_time_histogram *h = &aclk_time_heatmap;

39 +

40 + // the histogram MUST be all-inclusive for the possible sizes,

41 + // so we start from 0, and the last value is UINT64_MAX.

42 +

43 + usec_t values[ACLK_TIME_HISTOGRAM_ENTRIES] = {

44 + // minimum

45 + 0,

46 +

47 + // ms

48 + 10 * USEC_PER_MS,

49 + 50 * USEC_PER_MS, 100 * USEC_PER_MS, 200 * USEC_PER_MS, 350 * USEC_PER_MS,

50 + 500 * USEC_PER_MS, 750 * USEC_PER_MS,

51 +

52 + // seconds

53 + 1 * USEC_PER_SEC, 2 * USEC_PER_SEC, 4 * USEC_PER_SEC, 8 * USEC_PER_SEC,

54 + 15 * USEC_PER_SEC, 30 * USEC_PER_SEC, 45 * USEC_PER_SEC,

55 +

56 + // minutes

57 + 60 * USEC_PER_SEC, 120 * USEC_PER_SEC, 180 * USEC_PER_SEC,

58 +

59 + // maximum

60 + UINT64_MAX

61 + };

62 +

63 + usec_t last_value = 0;

64 + for(size_t i = 0; i < ACLK_TIME_HISTOGRAM_ENTRIES; i++) {

65 + if(i > 0 && values[i] == 0)

66 + fatal("only the first value in the array can be zero");

67 +

68 + if(i > 0 && values[i] <= last_value)

69 + fatal("the values need to be sorted");

70 +

71 + h->array[i].upto = values[i];

72 + last_value = values[i];

73 + }

74 +}

75 +

76 +static inline size_t aclk_time_histogram_slot(struct aclk_time_histogram *h, usec_t dt_ut) {

77 + if(dt_ut <= h->array[0].upto)

78 + return 0;

79 +

80 + if(dt_ut >= h->array[_countof(h->array) - 1].upto)

81 + return _countof(h->array) - 1;

82 +

83 + // binary search for the right size

84 + size_t low = 0, high = _countof(h->array) - 1;

85 + while (low < high) {

86 + size_t mid = low + (high - low) / 2;

87 + if (dt_ut < h->array[mid].upto)

88 + high = mid;

89 + else

90 + low = mid + 1;

91 + }

92 + return low - 1;

93 +}

94 +

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

96 + if(!sent_ut) return;

97 +

98 + usec_t usec = now_monotonic_usec() - sent_ut;

99 +

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

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

102 +

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

104 +}

105 +

106 +static void pulse_aclk_time_heatmap(void) {

107 + static RRDSET *st;

108 + static RRDDIM *rds[ACLK_TIME_HISTOGRAM_ENTRIES];

109 +

110 + if(!st) {

111 + st = rrdset_create_localhost(

112 + "netdata",

113 + "aclk_puback_latency",

114 + NULL,

115 + PULSE_NETWORK_CHART_FAMILY,

116 + "netdata.aclk_puback_latency",

117 + "Netdata ACLK PubACK Latency In Seconds",

118 + "messages",

119 + "netdata",

120 + "pulse",

121 + PULSE_NETWORK_CHART_PRIORITY + 1,

122 + localhost->rrd_update_every,

123 + RRDSET_TYPE_HEATMAP);

124 +

125 + rds[0] = rrddim_add(st, "instant", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);

126 + for(size_t i = 1; i < _countof(rds) - 1 ;i++) {

127 + char buf[128];

128 + snprintf(buf, sizeof(buf), "%.2fs", (double)aclk_time_heatmap.array[i].upto / (double)USEC_PER_SEC);

129 + // duration_snprintf(buf, sizeof(buf), aclk_time_heatmap.array[i].upto, "us", false);

130 + rds[i] = rrddim_add(st, buf, NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);

131 + }

132 + rds[_countof(rds) - 1] = rrddim_add(st, "+inf", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);

133 + }

134 +

135 + for(size_t i = 0; i < _countof(rds) - 1 ;i++) {

136 + size_t old_value = 0, new_value = 0;

137 + __atomic_exchange(&aclk_time_heatmap.array[i].count, &new_value, &old_value, __ATOMIC_RELAXED);

138 + rrddim_set_by_pointer(st, rds[i], (collected_number)old_value);

139 + }

140 +

141 + rrdset_done(st);

142 +}

143 +

144 +// --------------------------------------------------------------------------------------------------------------------

145 +

146 void pulse_web_server_received_bytes(size_t bytes) {

147 __atomic_add_fetch(&live_stats.api_bytes_received, bytes, __ATOMIC_RELAXED);

148 }

@@ -185,5 +309,65 @@ void pulse_network_do(bool extended __maybe_unused) {

309 rrddim_set_by_pointer(st_bytes, rd_out, (collected_number)t.bytes_tx);

310 rrdset_done(st_bytes);

311 }

312 +

313 + pulse_aclk_time_heatmap();

314 +

315 + if(extended) {

316 + static RRDSET *st_aclk_queue_size = NULL;

317 + static RRDDIM *rd_messages = NULL;

318 +

319 + if (unlikely(!st_aclk_queue_size)) {

320 + st_aclk_queue_size = rrdset_create_localhost(

321 + "netdata",

322 + "network_aclk_send_queue",

323 + NULL,

324 + PULSE_NETWORK_CHART_FAMILY,

325 + "netdata.network_aclk_send_queue",

326 + "Netdata ACLK Send Queue Size",

327 + "messages",

328 + "netdata",

329 + "pulse",

330 + PULSE_NETWORK_CHART_PRIORITY + 2,

331 + localhost->rrd_update_every,

332 + RRDSET_TYPE_AREA);

333 +

334 + rrdlabels_add(st_aclk_queue_size->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);

335 +

336 + rd_messages = rrddim_add(st_aclk_queue_size, "messages", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);

337 + }

338 +

339 + rrddim_set_by_pointer(st_aclk_queue_size, rd_messages, (collected_number)t.mqtt.tx_messages_queued);

340 + rrdset_done(st_aclk_queue_size);

341 + }

342 +

343 + if(extended) {

344 + static RRDSET *st_aclk_messages = NULL;

345 + static RRDDIM *rd_in = NULL, *rd_out = NULL;

346 +

347 + if (unlikely(!st_aclk_messages)) {

348 + st_aclk_messages = rrdset_create_localhost(

349 + "netdata",

350 + "network_aclk_messages",

351 + NULL,

352 + PULSE_NETWORK_CHART_FAMILY,

353 + "netdata.network_aclk_messages",

354 + "Netdata ACLK Messages",

355 + "messages/s",

356 + "netdata",

357 + "pulse",

358 + PULSE_NETWORK_CHART_PRIORITY + 3,

359 + localhost->rrd_update_every,

360 + RRDSET_TYPE_AREA);

361 +

362 + rrdlabels_add(st_aclk_messages->rrdlabels, "endpoint", "aclk", RRDLABEL_SRC_AUTO);

363 +

364 + rd_in = rrddim_add(st_aclk_messages, "received", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);

365 + rd_out = rrddim_add(st_aclk_messages, "queued", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);

366 + }

367 +

368 + rrddim_set_by_pointer(st_aclk_messages, rd_in, (collected_number)t.mqtt.rx_messages_rcvd);

369 + rrddim_set_by_pointer(st_aclk_messages, rd_out, (collected_number)t.mqtt.tx_messages_sent);

370 + rrdset_done(st_aclk_messages);

371 + }

372 }

373 }

src/daemon/pulse/pulse-network.h

@@ -16,4 +16,10 @@ 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);

20 +

21 +#ifdef PULSE_INTERNALS

22 +void aclk_time_histogram_init(void);

23 +#endif

24 +

25 #endif //NETDATA_PULSE_NETWORK_H

src/daemon/pulse/pulse.c

@@ -73,6 +73,7 @@ void *pulse_thread_main(void *ptr) {

73 }

75 pulse_aral_init();

76 + aclk_time_histogram_init();

78 usec_t step = update_every * USEC_PER_SEC;

79 heartbeat_t hb;

src/database/sqlite/sqlite_aclk.c

+72 -9

@@ -303,18 +303,83 @@ static void after_aclk_run_query_job(uv_work_t *req, int status __maybe_unused)

303 freez(payload);

304 }

305

306 -static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query)

306 +static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query, bool is_worker)

307 {

308 if (query->type == UNKNOWN || query->type >= ACLK_QUERY_TYPE_COUNT) {

309 error_report("Unknown query in query queue. %u", query->type);

310 return;

311 }

312

313 - if (query->type == HTTP_API_V2) {

314 - http_api_v2(config->client, query);

315 - } else {

316 - send_bin_msg(config->client, query);

313 + struct ctxs_checkpoint *cmd;

314 +

315 + bool ok_to_send = true;

316 +

317 + switch (query->type) {

318 + case HTTP_API_V2:

319 + if (is_worker)

320 + worker_is_busy(UV_EVENT_ACLK_QUERY_EXECUTE);

321 + http_api_v2(config->client, query);

322 + ok_to_send = false;

323 + break;

324 + case CTX_CHECKPOINT:;

325 + if (is_worker)

326 + worker_is_busy(UV_EVENT_CTX_CHECKPOINT);

327 + cmd = query->data.payload;

328 + rrdcontext_hub_checkpoint_command(cmd);

329 + freez(cmd->claim_id);

330 + freez(cmd->node_id);

331 + freez(cmd);

332 + ok_to_send = false;

333 + break;

334 + case CTX_STOP_STREAMING:

335 + if (is_worker)

336 + worker_is_busy(UV_EVENT_CTX_STOP_STREAMING);

337 + cmd = query->data.payload;

338 + rrdcontext_hub_stop_streaming_command(cmd);

339 + freez(cmd->claim_id);

340 + freez(cmd->node_id);

341 + freez(cmd);

342 + ok_to_send = false;

343 + break;

344 + case ALARM_PROVIDE_CFG:

345 + if (is_worker)

346 + worker_is_busy(UV_EVENT_ALARM_PROVIDE_CFG);

347 + break;

348 + case ALARM_SNAPSHOT:

349 + if (is_worker)

350 + worker_is_busy(UV_EVENT_ALARM_SNAPSHOT);

351 + break;

352 + case REGISTER_NODE:

353 + if (is_worker)

354 + worker_is_busy(UV_EVENT_REGISTER_NODE);

355 + break;

356 + case UPDATE_NODE_COLLECTORS:

357 + if (is_worker)

358 + worker_is_busy(UV_EVENT_UPDATE_NODE_COLLECTORS);

359 + break;

360 + case UPDATE_NODE_INFO:

361 + if (is_worker)

362 + worker_is_busy(UV_EVENT_UPDATE_NODE_INFO);

363 + break;

364 + case CTX_SEND_SNAPSHOT:

365 + if (is_worker)

366 + worker_is_busy(UV_EVENT_CTX_SEND_SNAPSHOT);

367 + break;

368 + case CTX_SEND_SNAPSHOT_UPD:

369 + if (is_worker)

370 + worker_is_busy(UV_EVENT_CTX_SEND_SNAPSHOT_UPD);

371 + break;

372 + case NODE_STATE_UPDATE:

373 + if (is_worker)

374 + worker_is_busy(UV_EVENT_NODE_STATE_UPDATE);

375 + break;

376 + default:

377 + nd_log_daemon(NDLP_ERR, "Unknown msg type %u; ignoring", query->type);

378 + ok_to_send = false;

379 + break;

380 }

381 + if (ok_to_send)

382 + send_bin_msg(config->client, query);

383 aclk_query_free(query);

384 }

385

@@ -322,13 +387,11 @@ static void aclk_run_query_job(uv_work_t *req)

387 {

388 register_libuv_worker_jobs();

389

325 - worker_is_busy(UV_EVENT_ACLK_QUERY_EXECUTE);

326 -

390 struct aclk_query_payload *payload = req->data;

391 struct aclk_sync_config_s *config = payload->config;

392 aclk_query_t query = (aclk_query_t) payload->data;

393

331 - aclk_run_query(config, query);

394 + aclk_run_query(config, query, true);

395 worker_is_idle();

396 }

397

@@ -508,7 +571,7 @@ static void aclk_synchronization(void *arg)

571

572 if (execute_now) {

573 worker_is_busy(ACLK_QUERY_EXECUTE_SYNC);

511 - aclk_run_query(config, query);

574 + aclk_run_query(config, query, false);

575 freez(payload);

576 config->aclk_queries_running--;

577 }