@cryptotaxi247 / netdata-1 / commits / 308ca33db

Use existing ACLK event loop for cloud queries (#18218)

* Use existing aclk sync event loop for ACLK queries * Remove puback check -- fix unit test * Remove more pub stats in internal checks * Fix compilation / Cleanup * Code cleanup Calculate aclk query threads based on available uv workers * Read max parallel query count from config, make use we keep below uv thread count / 2 * Fix compilation warning

Stelios Fragkakis committed Aug 26, 2024 at 13:52 UTC 308ca33db92dab1d7ca25e927c6061b1a94cac62
24 files changed +228 -1069
CMakeLists.txt
-2
@@ -1507,8 +1507,6 @@ set(ACLK_PROTO_DEFS
1507 )
1508
1509 set(ACLK_FILES
1510 - src/aclk/aclk_stats.c
1511 - src/aclk/aclk_stats.h
1510 src/aclk/aclk_query.c
1511 src/aclk/aclk_query.h
1512 src/aclk/aclk_query_queue.c
src/aclk/aclk.c
+12 -79
@@ -2,7 +2,6 @@
2
3 #include "aclk.h"
4
5 -#include "aclk_stats.h"
5 #include "mqtt_websockets/mqtt_wss_client.h"
6 #include "aclk_otp.h"
7 #include "aclk_tx_msgs.h"
@@ -21,8 +20,6 @@
20 #include <fcntl.h>
21 #endif
22
24 -#define ACLK_STABLE_TIMEOUT 3 // Minimum delay to mark AGENT as stable
25 -
23 int aclk_pubacks_per_conn = 0; // How many PubAcks we got since MQTT conn est.
24 int aclk_rcvd_cloud_msgs = 0;
25 int aclk_connection_counter = 0;
@@ -54,7 +51,6 @@ bool aclk_online_for_nodes(void) {
51
52 int aclk_ctx_based = 0;
53 int aclk_disable_runtime = 0;
57 -int aclk_stats_enabled;
54 int aclk_kill_link = 0;
55
56 usec_t aclk_session_us = 0;
@@ -70,10 +66,6 @@ time_t aclk_block_until = 0;
66
67 mqtt_wss_client mqttwss_client;
68
73 -//netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
74 -//#define ACLK_SHARED_STATE_LOCK netdata_mutex_lock(&aclk_shared_state_mutex)
75 -//#define ACLK_SHARED_STATE_UNLOCK netdata_mutex_unlock(&aclk_shared_state_mutex)
76 -
69 struct aclk_shared_state aclk_shared_state = {
70 .mqtt_shutdown_msg_id = -1,
71 .mqtt_shutdown_msg_rcvd = 0
@@ -303,9 +295,9 @@ static void puback_callback(uint16_t packet_id)
295 aclk_tbeb_reset();
296 }
297
306 -#ifdef NETDATA_INTERNAL_CHECKS
307 - aclk_stats_msg_puback(packet_id);
308 -#endif
298 +//#ifdef NETDATA_INTERNAL_CHECKS
299 +// aclk_stats_msg_puback(packet_id);
300 +//#endif
301
302 if (aclk_shared_state.mqtt_shutdown_msg_id == (int)packet_id) {
303 nd_log(NDLS_DAEMON, NDLP_DEBUG,
@@ -315,19 +307,6 @@ static void puback_callback(uint16_t packet_id)
307 }
308 }
309
318 -static int read_query_thread_count()
319 -{
320 - int threads = MIN(get_netdata_cpus()/2, 6);
321 - threads = MAX(threads, 2);
322 - threads = config_get_number(CONFIG_SECTION_CLOUD, "query thread count", threads);
323 - if(threads < 1) {
324 - netdata_log_error("You need at least one query thread. Overriding configured setting of \"%d\"", threads);
325 - threads = 1;
326 - config_set_number(CONFIG_SECTION_CLOUD, "query thread count", threads);
327 - }
328 - return threads;
329 -}
330 -
310 void aclk_graceful_disconnect(mqtt_wss_client client);
311
312 /* Keeps connection alive and handles all network communications.
@@ -338,7 +317,6 @@ void aclk_graceful_disconnect(mqtt_wss_client client);
317 */
318 static int handle_connection(mqtt_wss_client client)
319 {
341 - time_t last_periodic_query_wakeup = now_monotonic_sec();
320 while (service_running(SERVICE_ACLK)) {
321 // timeout 1000 to check at least once a second
322 // for netdata_exit
@@ -356,21 +334,10 @@ static int handle_connection(mqtt_wss_client client)
334 disconnect_req = 0;
335 aclk_kill_link = 0;
336 aclk_graceful_disconnect(client);
359 - aclk_queue_unlock();
337 aclk_shared_state.mqtt_shutdown_msg_id = -1;
338 aclk_shared_state.mqtt_shutdown_msg_rcvd = 0;
339 return 1;
340 }
364 -
365 - // mqtt_wss_service will return faster than in one second
366 - // if there is enough work to do
367 - time_t now = now_monotonic_sec();
368 - if (last_periodic_query_wakeup < now) {
369 - // wake up at least one Query Thread at least
370 - // once per second
371 - last_periodic_query_wakeup = now;
372 - QUERY_THREAD_WAKEUP;
373 - }
341 }
342 return 0;
343 }
@@ -390,7 +357,6 @@ static inline void mqtt_connected_actions(mqtt_wss_client client)
357 else
358 mqtt_wss_subscribe(client, topic, 1);
359
393 - aclk_stats_upd_online(1);
360 aclk_set_connected();
361 aclk_pubacks_per_conn = 0;
362 aclk_rcvd_cloud_msgs = 0;
@@ -408,9 +374,6 @@ void aclk_graceful_disconnect(mqtt_wss_client client)
374 nd_log(NDLS_DAEMON, NDLP_DEBUG,
375 "Preparing to gracefully shutdown ACLK connection");
376
411 - aclk_queue_lock();
412 - aclk_queue_flush();
413 -
377 aclk_shared_state.mqtt_shutdown_msg_id = aclk_send_agent_connection_update(client, 0);
378
379 time_t t = now_monotonic_sec();
@@ -429,7 +392,6 @@ void aclk_graceful_disconnect(mqtt_wss_client client)
392 nd_log(NDLS_DAEMON, NDLP_WARNING, "ACLK link is down");
393 nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK DISCONNECTED");
394
432 - aclk_stats_upd_online(0);
395 last_disconnect_time = now_realtime_sec();
396 aclk_set_disconnected();
397
@@ -806,11 +768,6 @@ void *aclk_main(void *ptr)
768 {
769 struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
770
809 - struct aclk_stats_thread *stats_thread = NULL;
810 -
811 - struct aclk_query_threads query_threads;
812 - query_threads.thread_list = NULL;
813 -
771 ACLK_PROXY_TYPE proxy_type;
772 aclk_get_proxy(&proxy_type);
773 if (proxy_type == PROXY_TYPE_SOCKS5) {
@@ -819,9 +776,7 @@ void *aclk_main(void *ptr)
776 return NULL;
777 }
778
822 - unsigned int proto_hdl_cnt = aclk_init_rx_msg_handlers();
823 -
824 - query_threads.count = read_query_thread_count();
779 + aclk_init_rx_msg_handlers();
780
781 if (wait_till_agent_claim_ready())
782 goto exit;
@@ -848,26 +803,15 @@ void *aclk_main(void *ptr)
803 // that send JSON payloads of 10 MB as single messages
804 mqtt_wss_set_max_buf_size(mqttwss_client, 25*1024*1024);
805
851 - aclk_stats_enabled = config_get_boolean(CONFIG_SECTION_CLOUD, "statistics", global_statistics_enabled);
852 - if (aclk_stats_enabled) {
853 - stats_thread = callocz(1, sizeof(struct aclk_stats_thread));
854 - stats_thread->query_thread_count = query_threads.count;
855 - stats_thread->client = mqttwss_client;
856 - aclk_stats_thread_prepare(query_threads.count, proto_hdl_cnt);
857 - stats_thread->thread = nd_thread_create("ACLK_STATS", NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread, stats_thread);
858 - }
859 -
806 // Keep reconnecting and talking until our time has come
807 // and the Grim Reaper (netdata_exit) calls
808 + netdata_log_info("Starting ACLK query event loop");
809 + aclk_query_init(mqttwss_client);
810 do {
811 if (aclk_attempt_to_connect(mqttwss_client))
812 goto exit_full;
813
866 - if (unlikely(!query_threads.thread_list))
867 - aclk_query_threads_start(&query_threads, mqttwss_client);
868 -
814 if (handle_connection(mqttwss_client)) {
870 - aclk_stats_upd_online(0);
815 last_disconnect_time = now_realtime_sec();
816 aclk_set_disconnected();
817 nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK DISCONNECTED");
@@ -882,16 +826,6 @@ void *aclk_main(void *ptr)
826 #endif
827
828 exit_full:
885 -// Tear Down
886 - QUERY_THREAD_WAKEUP_ALL;
887 -
888 - aclk_query_threads_cleanup(&query_threads);
889 -
890 - if (aclk_stats_enabled) {
891 - nd_thread_join(stats_thread->thread);
892 - aclk_stats_thread_cleanup();
893 - freez(stats_thread);
894 - }
829 free_topic_cache();
830 mqtt_wss_destroy(mqttwss_client);
831 exit:
@@ -906,7 +840,6 @@ exit:
840 void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
841 {
842 nd_uuid_t node_id;
909 - int ret = 0;
843
844 if (!aclk_online())
845 return;
@@ -915,7 +848,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
848 uuid_copy(node_id, host->node_id);
849 }
850 else {
918 - ret = get_node_id(&host->host_uuid, &node_id);
851 + int ret = get_node_id(&host->host_uuid, &node_id);
852 if (ret > 0) {
853 // this means we were not able to check if node_id already present
854 netdata_log_error("Unable to check for node_id. Ignoring the host state update.");
@@ -941,7 +874,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
874 nd_log(NDLS_DAEMON, NDLP_DEBUG,
875 "Registering host=%s, hops=%u", host->machine_guid, host->system_info->hops);
876
944 - aclk_queue_query(create_query);
877 + aclk_execute_query(create_query);
878 return;
879 }
880 }
@@ -968,7 +901,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
901 freez((void*)node_state_update.node_id);
902 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
903 query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
971 - aclk_queue_query(query);
904 + aclk_execute_query(query);
905 }
906
907 void aclk_send_node_instances()
@@ -1014,7 +947,7 @@ void aclk_send_node_instances()
947 freez((void*)node_state_update.node_id);
948 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
949 query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
1017 - aclk_queue_query(query);
950 + aclk_execute_query(query);
951 } else {
952 aclk_query_t create_query;
953 create_query = aclk_query_new(REGISTER_NODE);
@@ -1036,7 +969,7 @@ void aclk_send_node_instances()
969 (char*)node_instance_creation.machine_guid, list->hops);
970
971 freez((void *)node_instance_creation.machine_guid);
1039 - aclk_queue_query(create_query);
972 + aclk_execute_query(create_query);
973 }
974 freez(list->hostname);
975
@@ -1105,7 +1038,7 @@ char *aclk_state(void)
1038 strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tmptr);
1039 buffer_sprintf(wb, "Last Disconnect Time: %s\n", timebuf);
1040 }
1108 - if (!aclk_online() && next_connection_attempt && (tmptr = localtime_r(&next_connection_attempt, &tmbuf)) ) {
1041 + if (!aclk_connected && next_connection_attempt && (tmptr = localtime_r(&next_connection_attempt, &tmbuf)) ) {
1042 char timebuf[26];
1043 strftime(timebuf, 26, "%Y-%m-%d %H:%M:%S", tmptr);
1044 buffer_sprintf(wb, "Next Connection Attempt At: %s\nLast Backoff: %.3f", timebuf, last_backoff_value);
src/aclk/aclk.h
+1 -5
@@ -39,7 +39,7 @@ const char *aclk_status_to_string(void);
39
40 extern int aclk_ctx_based;
41 extern int aclk_disable_runtime;
42 -extern int aclk_stats_enabled;
42 +//extern int aclk_stats_enabled;
43 extern int aclk_kill_link;
44
45 bool aclk_online(void);
@@ -66,10 +66,6 @@ extern int disconnect_req;
66
67 void *aclk_main(void *ptr);
68
69 -extern netdata_mutex_t aclk_shared_state_mutex;
70 -#define ACLK_SHARED_STATE_LOCK netdata_mutex_lock(&aclk_shared_state_mutex)
71 -#define ACLK_SHARED_STATE_UNLOCK netdata_mutex_unlock(&aclk_shared_state_mutex)
72 -
69 extern struct aclk_shared_state {
70 // To wait for `disconnect` message PUBACK
71 // when shutting down
src/aclk/aclk_alarm_api.c
-9
@@ -8,15 +8,6 @@
8
9 #include "aclk.h"
10
11 -void aclk_send_provide_alarm_checkpoint(struct alarm_checkpoint *checkpoint)
12 -{
13 - aclk_query_t query = aclk_query_new(ALARM_PROVIDE_CHECKPOINT);
14 - query->data.bin_payload.payload = generate_alarm_checkpoint(&query->data.bin_payload.size, checkpoint);
15 - query->data.bin_payload.topic = ACLK_TOPICID_ALARM_CHECKPOINT;
16 - query->data.bin_payload.msg_name = "AlarmCheckpoint";
17 - QUEUE_IF_PAYLOAD_PRESENT(query);
18 -}
19 -
11 void aclk_send_alarm_log_entry(struct alarm_log_entry *log_entry)
12 {
13 size_t payload_size;
src/aclk/aclk_alarm_api.h
+1 -1
@@ -6,7 +6,7 @@
6 #include "../daemon/common.h"
7 #include "schema-wrappers/schema_wrappers.h"
8
9 -void aclk_send_provide_alarm_checkpoint(struct alarm_checkpoint *checkpoint);
9 +//void aclk_send_provide_alarm_checkpoint(struct alarm_checkpoint *checkpoint);
10 void aclk_send_alarm_log_entry(struct alarm_log_entry *log_entry);
11 void aclk_send_provide_alarm_cfg(struct provide_alarm_configuration *cfg);
12 void aclk_send_alarm_snapshot(alarm_snapshot_proto_ptr_t snapshot);
src/aclk/aclk_otp.c
-2
@@ -5,8 +5,6 @@
5 #include "aclk_util.h"
6 #include "aclk.h"
7
8 -#include "daemon/common.h"
9 -
8 #include "mqtt_websockets/c-rbuf/cringbuffer.h"
9
10 static int aclk_https_request(https_req_t *request, https_req_response_t *response) {
src/aclk/aclk_query.c
+30 -179
@@ -1,7 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_query.h"
4 -#include "aclk_stats.h"
4 #include "aclk_tx_msgs.h"
5 #include "../../web/server/web_client_cache.h"
6
@@ -24,7 +23,7 @@ struct pending_req_list {
23 };
24
25 static struct pending_req_list *pending_req_list_head = NULL;
27 -static pthread_mutex_t pending_req_list_lock = PTHREAD_MUTEX_INITIALIZER;
26 +static SPINLOCK pending_req_list_lock = NETDATA_SPINLOCK_INITIALIZER;
27
28 void aclk_config_get_query_scope(void) {
29 const char *s = config_get(CONFIG_SECTION_CLOUD, "scope", "full");
@@ -42,10 +41,10 @@ static struct pending_req_list *pending_req_list_add(const char *msg_id)
41 new->msg_id = msg_id;
42 new->hash = simple_hash(msg_id);
43
45 - pthread_mutex_lock(&pending_req_list_lock);
44 + spinlock_lock(&pending_req_list_lock);
45 new->next = pending_req_list_head;
46 pending_req_list_head = new;
48 - pthread_mutex_unlock(&pending_req_list_lock);
47 + spinlock_unlock(&pending_req_list_lock);
48 return new;
49 }
50
@@ -54,7 +53,7 @@ void pending_req_list_rm(const char *msg_id)
53 uint32_t hash = simple_hash(msg_id);
54 struct pending_req_list *prev = NULL;
55
57 - pthread_mutex_lock(&pending_req_list_lock);
56 + spinlock_lock(&pending_req_list_lock);
57 struct pending_req_list *curr = pending_req_list_head;
58
59 while (curr) {
@@ -71,26 +70,26 @@ void pending_req_list_rm(const char *msg_id)
70 prev = curr;
71 curr = curr->next;
72 }
74 - pthread_mutex_unlock(&pending_req_list_lock);
73 + spinlock_unlock(&pending_req_list_lock);
74 }
75
76 int mark_pending_req_cancelled(const char *msg_id)
77 {
78 uint32_t hash = simple_hash(msg_id);
79
81 - pthread_mutex_lock(&pending_req_list_lock);
80 + spinlock_lock(&pending_req_list_lock);
81 struct pending_req_list *curr = pending_req_list_head;
82
83 while (curr) {
84 if (curr->hash == hash && strcmp(curr->msg_id, msg_id) == 0) {
85 curr->canceled = 1;
87 - pthread_mutex_unlock(&pending_req_list_lock);
86 + spinlock_unlock(&pending_req_list_lock);
87 return 0;
88 }
89
90 curr = curr->next;
91 }
93 - pthread_mutex_unlock(&pending_req_list_lock);
92 + spinlock_unlock(&pending_req_list_lock);
93 return 1;
94 }
95
@@ -100,7 +99,8 @@ static bool aclk_web_client_interrupt_cb(struct web_client *w __maybe_unused, vo
99 return req->canceled;
100 }
101
103 -static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query) {
102 +int http_api_v2(mqtt_wss_client client, aclk_query_t query)
103 +{
104 ND_LOG_STACK lgs[] = {
105 ND_LOG_FIELD_TXT(NDF_SRC_TRANSPORT, "aclk"),
106 ND_LOG_FIELD_END(),
@@ -109,8 +109,6 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
109
110 int retval = 0;
111 BUFFER *local_buffer = NULL;
112 - size_t size = 0;
113 - size_t sent = 0;
112 usec_t dt_ut = 0;
113
114 int z_ret;
@@ -136,7 +134,7 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
134 nd_log(NDLS_ACCESS, NDLP_ERR, "ACLK received request is not valid, code %d", validation);
135 retval = 1;
136 w->response.code = HTTP_RESP_BAD_REQUEST;
139 - w->response.code = (short)aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id,
137 + w->response.code = (short)aclk_http_msg_v2(client, query->callback_topic, query->msg_id,
138 dt_ut, query->created, w->response.code,
139 NULL, 0);
140 goto cleanup;
@@ -149,39 +147,18 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
147 dt_ut / USEC_PER_MS, query->timeout);
148 retval = 1;
149 w->response.code = HTTP_RESP_SERVICE_UNAVAILABLE;
152 - aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0);
150 + aclk_http_msg_v2_err(client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0);
151 goto cleanup;
152 }
153
154 char *path = (char *)buffer_tostring(w->url_path_decoded);
155
158 - if (aclk_stats_enabled) {
159 - char *url_path_endpoint = strrchr(path, '/');
160 - ACLK_STATS_LOCK;
161 - int stat_idx = aclk_cloud_req_http_type_to_idx(url_path_endpoint ? url_path_endpoint + 1 : "other");
162 - aclk_metrics_per_sample.cloud_req_http_by_type[stat_idx]++;
163 - ACLK_STATS_UNLOCK;
164 - }
165 -
156 w->response.code = (short)web_client_api_request_with_node_selection(localhost, w, path);
157 web_client_timeout_checkpoint_response_ready(w, &dt_ut);
158
169 - if (aclk_stats_enabled) {
170 - ACLK_STATS_LOCK;
171 - aclk_metrics_per_sample.cloud_q_process_total += dt_ut;
172 - aclk_metrics_per_sample.cloud_q_process_count++;
173 - if (aclk_metrics_per_sample.cloud_q_process_max < dt_ut)
174 - aclk_metrics_per_sample.cloud_q_process_max = dt_ut;
175 - ACLK_STATS_UNLOCK;
176 - }
177 -
178 - size = w->response.data->len;
179 - sent = size;
180 -
159 if (w->response.data->len && w->response.zinitialized) {
160 w->response.zstream.next_in = (Bytef *)w->response.data->buffer;
161 w->response.zstream.avail_in = w->response.data->len;
184 -
162 do {
163 w->response.zstream.avail_out = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE;
164 w->response.zstream.next_out = w->response.zbuffer;
@@ -193,7 +170,7 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
170 netdata_log_error("Unknown error during zlib compression.");
171 retval = 1;
172 w->response.code = 500;
196 - aclk_http_msg_v2_err(query_thr->client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_ZLIB_ERROR, CLOUD_EMSG_ZLIB_ERROR, NULL, 0);
173 + aclk_http_msg_v2_err(client, query->callback_topic, query->msg_id, w->response.code, CLOUD_EC_ZLIB_ERROR, CLOUD_EMSG_ZLIB_ERROR, NULL, 0);
174 goto cleanup;
175 }
176 int bytes_to_cpy = NETDATA_WEB_RESPONSE_ZLIB_CHUNK_SIZE - w->response.zstream.avail_out;
@@ -220,16 +197,20 @@ static int http_api_v2(struct aclk_query_thread *query_thr, aclk_query_t query)
197 buffer_need_bytes(local_buffer, w->response.data->len);
198 memcpy(&local_buffer->buffer[local_buffer->len], w->response.data->buffer, w->response.data->len);
199 local_buffer->len += w->response.data->len;
223 - sent = sent - size + w->response.data->len;
224 - } else {
200 + } else
201 buffer_strcat(local_buffer, w->response.data->buffer);
226 - }
202 }
203
204 // send msg.
230 - w->response.code = (short)aclk_http_msg_v2(query_thr->client, query->callback_topic, query->msg_id,
231 - dt_ut, query->created, w->response.code,
232 - local_buffer->buffer, local_buffer->len);
205 + w->response.code = (short)aclk_http_msg_v2(
206 + client,
207 + query->callback_topic,
208 + query->msg_id,
209 + dt_ut,
210 + query->created,
211 + w->response.code,
212 + local_buffer->buffer,
213 + local_buffer->len);
214
215 cleanup:
216 web_client_log_completed_request(w, false);
@@ -242,144 +223,14 @@ cleanup:
223 return retval;
224 }
225
245 -static int send_bin_msg(struct aclk_query_thread *query_thr, aclk_query_t query)
226 +int send_bin_msg(mqtt_wss_client client, aclk_query_t query)
227 {
228 // this will be simplified when legacy support is removed
248 - aclk_send_bin_message_subtopic_pid(query_thr->client, query->data.bin_payload.payload, query->data.bin_payload.size, query->data.bin_payload.topic, query->data.bin_payload.msg_name);
249 - return 0;
250 -}
251 -
252 -const char *aclk_query_get_name(aclk_query_type_t qt, int unknown_ok)
253 -{
254 - switch (qt) {
255 - case HTTP_API_V2: return "http_api_request_v2";
256 - case REGISTER_NODE: return "register_node";
257 - case NODE_STATE_UPDATE: return "node_state_update";
258 - case CHART_DIMS_UPDATE: return "chart_and_dim_update";
259 - case CHART_CONFIG_UPDATED: return "chart_config_updated";
260 - case CHART_RESET: return "reset_chart_messages";
261 - case RETENTION_UPDATED: return "update_retention_info";
262 - case UPDATE_NODE_INFO: return "update_node_info";
263 - case ALARM_PROVIDE_CHECKPOINT: return "alarm_checkpoint";
264 - case ALARM_PROVIDE_CFG: return "provide_alarm_config";
265 - case ALARM_SNAPSHOT: return "alarm_snapshot";
266 - case UPDATE_NODE_COLLECTORS: return "update_node_collectors";
267 - case PROTO_BIN_MESSAGE: return "generic_binary_proto_message";
268 - default:
269 - if (!unknown_ok)
270 - error_report("Unknown query type used %d", (int) qt);
271 - return "unknown";
272 - }
273 -}
274 -
275 -static void aclk_query_process_msg(struct aclk_query_thread *query_thr, aclk_query_t query)
276 -{
277 - if (query->type == UNKNOWN || query->type >= ACLK_QUERY_TYPE_COUNT) {
278 - error_report("Unknown query in query queue. %u", query->type);
279 - aclk_query_free(query);
280 - return;
281 - }
282 -
283 - worker_is_busy(query->type);
284 - if (query->type == HTTP_API_V2) {
285 - netdata_log_debug(D_ACLK, "Processing Queued Message of type: \"http_api_request_v2\"");
286 - http_api_v2(query_thr, query);
287 - } else {
288 - netdata_log_debug(D_ACLK, "Processing Queued Message of type: \"%s\"", query->data.bin_payload.msg_name);
289 - send_bin_msg(query_thr, query);
290 - }
291 -
292 - if (aclk_stats_enabled) {
293 - ACLK_STATS_LOCK;
294 - aclk_metrics_per_sample.queries_dispatched++;
295 - aclk_queries_per_thread[query_thr->idx]++;
296 - aclk_metrics_per_sample.queries_per_type[query->type]++;
297 - ACLK_STATS_UNLOCK;
298 - }
299 -
300 - aclk_query_free(query);
301 -
302 - worker_is_idle();
303 -}
304 -
305 -/* Processes messages from queue. Compete for work with other threads
306 - */
307 -int aclk_query_process_msgs(struct aclk_query_thread *query_thr)
308 -{
309 - aclk_query_t query;
310 - while ((query = aclk_queue_pop()))
311 - aclk_query_process_msg(query_thr, query);
312 -
229 + aclk_send_bin_message_subtopic_pid(
230 + client,
231 + query->data.bin_payload.payload,
232 + query->data.bin_payload.size,
233 + query->data.bin_payload.topic,
234 + query->data.bin_payload.msg_name);
235 return 0;
236 }
315 -
316 -static void worker_aclk_register(void) {
317 - worker_register("ACLKQUERY");
318 - for (int i = 1; i < ACLK_QUERY_TYPE_COUNT; i++) {
319 - worker_register_job_name(i, aclk_query_get_name(i, 0));
320 - }
321 -}
322 -
323 -static void aclk_query_request_cancel(void *data)
324 -{
325 - pthread_cond_broadcast((pthread_cond_t *) data);
326 -}
327 -
328 -/**
329 - * Main query processing thread
330 - */
331 -void *aclk_query_main_thread(void *ptr)
332 -{
333 - worker_aclk_register();
334 -
335 - struct aclk_query_thread *query_thr = ptr;
336 -
337 - service_register(SERVICE_THREAD_TYPE_NETDATA, aclk_query_request_cancel, NULL, &query_cond_wait, false);
338 -
339 - while (service_running(SERVICE_ACLK | ABILITY_DATA_QUERIES)) {
340 - aclk_query_process_msgs(query_thr);
341 -
342 - worker_is_idle();
343 - QUERY_THREAD_LOCK;
344 - if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))
345 - sleep_usec(USEC_PER_SEC * 1);
346 - QUERY_THREAD_UNLOCK;
347 - }
348 -
349 - worker_unregister();
350 - return NULL;
351 -}
352 -
353 -#define TASK_LEN_MAX 22
354 -void aclk_query_threads_start(struct aclk_query_threads *query_threads, mqtt_wss_client client)
355 -{
356 - netdata_log_info("Starting %d query threads.", query_threads->count);
357 -
358 - char thread_name[TASK_LEN_MAX];
359 - query_threads->thread_list = callocz(query_threads->count, sizeof(struct aclk_query_thread));
360 - for (int i = 0; i < query_threads->count; i++) {
361 - query_threads->thread_list[i].idx = i; //thread needs to know its index for statistics
362 - query_threads->thread_list[i].client = client;
363 -
364 - if(unlikely(snprintfz(thread_name, TASK_LEN_MAX, "ACLK_QRY[%d]", i) < 0))
365 - netdata_log_error("snprintf encoding error");
366 -
367 - query_threads->thread_list[i].thread = nd_thread_create(
368 - thread_name,
369 - NETDATA_THREAD_OPTION_JOINABLE,
370 - aclk_query_main_thread,
371 - &query_threads->thread_list[i]);
372 - }
373 -}
374 -
375 -void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads)
376 -{
377 - if (query_threads && query_threads->thread_list) {
378 - for (int i = 0; i < query_threads->count; i++) {
379 - nd_thread_join(query_threads->thread_list[i].thread);
380 - }
381 - freez(query_threads->thread_list);
382 - }
383 - aclk_queue_lock();
384 - aclk_queue_flush();
385 -}
src/aclk/aclk_query.h
+5 -24
@@ -9,30 +9,11 @@
9
10 #include "aclk_query_queue.h"
11
12 -extern pthread_cond_t query_cond_wait;
13 -extern pthread_mutex_t query_lock_wait;
14 -#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
15 -#define QUERY_THREAD_WAKEUP_ALL pthread_cond_broadcast(&query_cond_wait)
16 -
17 -// TODO
18 -//extern volatile int aclk_connected;
19 -
20 -struct aclk_query_thread {
21 - ND_THREAD *thread;
22 - int idx;
23 - mqtt_wss_client client;
24 -};
25 -
26 -struct aclk_query_threads {
27 - struct aclk_query_thread *thread_list;
28 - int count;
29 -};
30 -
31 -void aclk_query_threads_start(struct aclk_query_threads *query_threads, mqtt_wss_client client);
32 -void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads);
33 -
34 -const char *aclk_query_get_name(aclk_query_type_t qt, int unknown_ok);
35 -
12 int mark_pending_req_cancelled(const char *msg_id);
13
14 +void aclk_execute_query(aclk_query_t query);
15 +void aclk_query_init(mqtt_wss_client client);
16 +int http_api_v2(mqtt_wss_client client, aclk_query_t query);
17 +int send_bin_msg(mqtt_wss_client client, aclk_query_t query);
18 +
19 #endif //NETDATA_AGENT_CLOUD_LINK_H
src/aclk/aclk_query_queue.c
+8 -103
@@ -1,87 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_query_queue.h"
4 -#include "aclk_query.h"
5 -#include "aclk_stats.h"
6 -
7 -static netdata_mutex_t aclk_query_queue_mutex = NETDATA_MUTEX_INITIALIZER;
8 -#define ACLK_QUEUE_LOCK netdata_mutex_lock(&aclk_query_queue_mutex)
9 -#define ACLK_QUEUE_UNLOCK netdata_mutex_unlock(&aclk_query_queue_mutex)
10 -
11 -static struct aclk_query_queue {
12 - aclk_query_t head;
13 - int block_push;
14 -} aclk_query_queue = {
15 - .head = NULL,
16 - .block_push = 0
17 -};
18 -
19 -static inline int _aclk_queue_query(aclk_query_t query)
20 -{
21 - now_monotonic_high_precision_timeval(&query->created_tv);
22 - query->created = now_realtime_usec();
23 -
24 - ACLK_QUEUE_LOCK;
25 - if (aclk_query_queue.block_push) {
26 - ACLK_QUEUE_UNLOCK;
27 - if(service_running(SERVICE_ACLK | ABILITY_DATA_QUERIES))
28 - netdata_log_error("Query Queue is blocked from accepting new requests. This is normally the case when ACLK prepares to shutdown.");
29 - aclk_query_free(query);
30 - return 1;
31 - }
32 - DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(aclk_query_queue.head, query, prev, next);
33 - ACLK_QUEUE_UNLOCK;
34 - return 0;
35 -
36 -}
37 -
38 -int aclk_queue_query(aclk_query_t query)
39 -{
40 - int ret = _aclk_queue_query(query);
41 - if (!ret) {
42 - QUERY_THREAD_WAKEUP;
43 - if (aclk_stats_enabled) {
44 - ACLK_STATS_LOCK;
45 - aclk_metrics_per_sample.queries_queued++;
46 - ACLK_STATS_UNLOCK;
47 - }
48 - }
49 - return ret;
50 -}
51 -
52 -aclk_query_t aclk_queue_pop(void)
53 -{
54 - aclk_query_t ret;
55 -
56 - ACLK_QUEUE_LOCK;
57 - if (aclk_query_queue.block_push) {
58 - ACLK_QUEUE_UNLOCK;
59 - if(service_running(SERVICE_ACLK | ABILITY_DATA_QUERIES))
60 - netdata_log_error("POP Query Queue is blocked from accepting new requests. This is normally the case when ACLK prepares to shutdown.");
61 - return NULL;
62 - }
63 -
64 - ret = aclk_query_queue.head;
65 - if (!ret) {
66 - ACLK_QUEUE_UNLOCK;
67 - return ret;
68 - }
69 -
70 - DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(aclk_query_queue.head, ret, prev, next);
71 - ACLK_QUEUE_UNLOCK;
72 -
73 - ret->next = NULL;
74 - return ret;
75 -}
76 -
77 -void aclk_queue_flush(void)
78 -{
79 - aclk_query_t query = aclk_queue_pop();
80 - while (query) {
81 - aclk_query_free(query);
82 - query = aclk_queue_pop();
83 - }
84 -}
4
5 aclk_query_t aclk_query_new(aclk_query_type_t type)
6 {
@@ -93,14 +12,14 @@ aclk_query_t aclk_query_new(aclk_query_type_t type)
12 void aclk_query_free(aclk_query_t query)
13 {
14 switch (query->type) {
96 - case HTTP_API_V2:
97 - freez(query->data.http_api_v2.payload);
98 - if (query->data.http_api_v2.query != query->dedup_id)
99 - freez(query->data.http_api_v2.query);
100 - break;
101 -
102 - default:
103 - break;
15 + case HTTP_API_V2:
16 + freez(query->data.http_api_v2.payload);
17 + if (query->data.http_api_v2.query != query->dedup_id)
18 + freez(query->data.http_api_v2.query);
19 + break;
20 +
21 + default:
22 + break;
23 }
24
25 freez(query->dedup_id);
@@ -108,17 +27,3 @@ void aclk_query_free(aclk_query_t query)
27 freez(query->msg_id);
28 freez(query);
29 }
111 -
112 -void aclk_queue_lock(void)
113 -{
114 - ACLK_QUEUE_LOCK;
115 - aclk_query_queue.block_push = 1;
116 - ACLK_QUEUE_UNLOCK;
117 -}
118 -
119 -void aclk_queue_unlock(void)
120 -{
121 - ACLK_QUEUE_LOCK;
122 - aclk_query_queue.block_push = 0;
123 - ACLK_QUEUE_UNLOCK;
124 -}
src/aclk/aclk_query_queue.h
+11 -21
@@ -14,12 +14,7 @@ typedef enum {
14 HTTP_API_V2,
15 REGISTER_NODE,
16 NODE_STATE_UPDATE,
17 - CHART_DIMS_UPDATE,
18 - CHART_CONFIG_UPDATED,
19 - CHART_RESET,
20 - RETENTION_UPDATED,
17 UPDATE_NODE_INFO,
22 - ALARM_PROVIDE_CHECKPOINT,
18 ALARM_PROVIDE_CFG,
19 ALARM_SNAPSHOT,
20 UPDATE_NODE_COLLECTORS,
@@ -32,7 +27,7 @@ struct aclk_query_http_api_v2 {
27 char *query;
28 };
29
35 -struct aclk_bin_payload {
30 +struct aclk_bin_payload {
31 char *payload;
32 size_t size;
33 enum aclk_topics topic;
@@ -55,7 +50,6 @@ struct aclk_query {
50 struct timeval created_tv;
51 usec_t created;
52 int timeout;
58 - aclk_query_t prev, next;
53
54 // TODO maybe remove?
55 int version;
@@ -68,20 +62,16 @@ struct aclk_query {
62 aclk_query_t aclk_query_new(aclk_query_type_t type);
63 void aclk_query_free(aclk_query_t query);
64
71 -int aclk_queue_query(aclk_query_t query);
72 -aclk_query_t aclk_queue_pop(void);
73 -void aclk_queue_flush(void);
65 +void aclk_execute_query(aclk_query_t query);
66
75 -void aclk_queue_lock(void);
76 -void aclk_queue_unlock(void);
77 -
78 -#define QUEUE_IF_PAYLOAD_PRESENT(query) do { \
79 - if (likely(query->data.bin_payload.payload)) { \
80 - aclk_queue_query(query); \
81 - } else { \
82 - nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to generate payload"); \
83 - aclk_query_free(query); \
84 - } \
85 -} while(0)
67 +#define QUEUE_IF_PAYLOAD_PRESENT(query) \
68 + do { \
69 + if (likely((query)->data.bin_payload.payload)) { \
70 + aclk_execute_query(query); \
71 + } else { \
72 + nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to generate payload"); \
73 + aclk_query_free(query); \
74 + } \
75 + } while (0)
76
77 #endif /* NETDATA_ACLK_QUERY_QUEUE_H */
src/aclk/aclk_rx_msgs.c
+3 -31
@@ -2,7 +2,6 @@
2
3 #include "aclk_rx_msgs.h"
4
5 -#include "aclk_stats.h"
5 #include "aclk_query_queue.h"
6 #include "aclk.h"
7 #include "aclk_capas.h"
@@ -165,7 +164,7 @@ static int aclk_handle_cloud_http_request_v2(struct aclk_request *cloud_to_agent
164 // it would be strange to get URL from `dedup_id`
165 query->data.http_api_v2.query = query->dedup_id;
166 query->msg_id = cloud_to_agent->msg_id;
168 - aclk_queue_query(query);
167 + aclk_execute_query(query);
168 return 0;
169
170 error:
@@ -301,7 +300,7 @@ int create_node_instance_result(const char *msg, size_t msg_len)
300 query->data.bin_payload.msg_name = "UpdateNodeInstanceConnection";
301 query->data.bin_payload.topic = ACLK_TOPICID_NODE_CONN;
302
304 - aclk_queue_query(query);
303 + aclk_execute_query(query);
304 freez(res.node_id);
305 freez(res.machine_guid);
306 return 0;
@@ -502,12 +501,7 @@ new_cloud_rx_msg_t *find_rx_handler_by_hash(simple_hash_t hash)
501 return NULL;
502 }
503
505 -const char *rx_handler_get_name(size_t i)
506 -{
507 - return rx_msgs[i].name;
508 -}
509 -
510 -unsigned int aclk_init_rx_msg_handlers(void)
504 +void aclk_init_rx_msg_handlers(void)
505 {
506 int i;
507 for (i = 0; rx_msgs[i].fnc; i++) {
@@ -520,29 +514,17 @@ unsigned int aclk_init_rx_msg_handlers(void)
514 }
515 rx_msgs[i].name_hash = hash;
516 }
523 - return i;
517 }
518
519 void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic __maybe_unused)
520 {
528 - if (aclk_stats_enabled) {
529 - ACLK_STATS_LOCK;
530 - aclk_metrics_per_sample.cloud_req_recvd++;
531 - ACLK_STATS_UNLOCK;
532 - }
521 new_cloud_rx_msg_t *msg_descriptor = find_rx_handler_by_hash(simple_hash(message_type));
522 netdata_log_debug(D_ACLK, "Got message named '%s' from cloud", message_type);
523 if (unlikely(!msg_descriptor)) {
524 netdata_log_error("Do not know how to handle message of type '%s'. Ignoring", message_type);
537 - if (aclk_stats_enabled) {
538 - ACLK_STATS_LOCK;
539 - aclk_metrics_per_sample.cloud_req_err++;
540 - ACLK_STATS_UNLOCK;
541 - }
525 return;
526 }
527
545 -
528 if (aclklog_enabled) {
529 if (!strncmp(message_type, "cmd", strlen("cmd"))) {
530 log_aclk_message_bin(msg, msg_len, 0, topic, msg_descriptor->name);
@@ -553,18 +535,8 @@ void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t
535 }
536 }
537
556 - if (aclk_stats_enabled) {
557 - ACLK_STATS_LOCK;
558 - aclk_proto_rx_msgs_sample[msg_descriptor-rx_msgs]++;
559 - ACLK_STATS_UNLOCK;
560 - }
538 if (msg_descriptor->fnc(msg, msg_len)) {
539 netdata_log_error("Error processing message of type '%s'", message_type);
563 - if (aclk_stats_enabled) {
564 - ACLK_STATS_LOCK;
565 - aclk_metrics_per_sample.cloud_req_err++;
566 - ACLK_STATS_UNLOCK;
567 - }
540 return;
541 }
542 }
src/aclk/aclk_rx_msgs.h
+1 -1
@@ -11,7 +11,7 @@
11 int aclk_handle_cloud_cmd_message(char *payload);
12
13 const char *rx_handler_get_name(size_t i);
14 -unsigned int aclk_init_rx_msg_handlers(void);
14 +void aclk_init_rx_msg_handlers(void);
15 void aclk_handle_new_cloud_msg(const char *message_type, const char *msg, size_t msg_len, const char *topic);
16
17 #endif /* ACLK_RX_MSGS_H */
src/aclk/aclk_stats.c deleted
-483
@@ -1,483 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -#ifndef MQTT_WSS_CPUSTATS
4 -#define MQTT_WSS_CPUSTATS
5 -#endif
6 -
7 -#include "aclk_stats.h"
8 -
9 -#include "aclk_query.h"
10 -
11 -netdata_mutex_t aclk_stats_mutex = NETDATA_MUTEX_INITIALIZER;
12 -
13 -struct {
14 - int query_thread_count;
15 - unsigned int proto_hdl_cnt;
16 - uint32_t *aclk_proto_rx_msgs_sample;
17 - RRDDIM **rx_msg_dims;
18 -} aclk_stats_cfg; // there is only 1 stats thread at a time
19 -
20 -// data ACLK stats need per query thread
21 -struct aclk_qt_data {
22 - RRDDIM *dim;
23 -} *aclk_qt_data = NULL;
24 -
25 -uint32_t *aclk_queries_per_thread = NULL;
26 -uint32_t *aclk_queries_per_thread_sample = NULL;
27 -uint32_t *aclk_proto_rx_msgs_sample = NULL;
28 -
29 -struct aclk_metrics aclk_metrics = {
30 - .online = 0,
31 -};
32 -
33 -struct aclk_metrics_per_sample aclk_metrics_per_sample;
34 -
35 -static void aclk_stats_collect(struct aclk_metrics_per_sample *per_sample, struct aclk_metrics *permanent)
36 -{
37 - static RRDSET *st_aclkstats = NULL;
38 - static RRDDIM *rd_online_status = NULL;
39 -
40 - if (unlikely(!st_aclkstats)) {
41 - st_aclkstats = rrdset_create_localhost(
42 - "netdata", "aclk_status", NULL, "aclk", NULL, "ACLK/Cloud connection status",
43 - "connected", "netdata", "stats", 200000, localhost->rrd_update_every, RRDSET_TYPE_LINE);
44 -
45 - rd_online_status = rrddim_add(st_aclkstats, "online", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
46 - }
47 -
48 - rrddim_set_by_pointer(st_aclkstats, rd_online_status, per_sample->offline_during_sample ? 0 : permanent->online);
49 -
50 - rrdset_done(st_aclkstats);
51 -}
52 -
53 -static void aclk_stats_query_queue(struct aclk_metrics_per_sample *per_sample)
54 -{
55 - static RRDSET *st_query_thread = NULL;
56 - static RRDDIM *rd_queued = NULL;
57 - static RRDDIM *rd_dispatched = NULL;
58 -
59 - if (unlikely(!st_query_thread)) {
60 - st_query_thread = rrdset_create_localhost(
61 - "netdata", "aclk_query_per_second", NULL, "aclk", NULL, "ACLK Queries per second", "queries/s",
62 - "netdata", "stats", 200001, localhost->rrd_update_every, RRDSET_TYPE_AREA);
63 -
64 - rd_queued = rrddim_add(st_query_thread, "added", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
65 - rd_dispatched = rrddim_add(st_query_thread, "dispatched", NULL, -1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
66 - }
67 -
68 - rrddim_set_by_pointer(st_query_thread, rd_queued, per_sample->queries_queued);
69 - rrddim_set_by_pointer(st_query_thread, rd_dispatched, per_sample->queries_dispatched);
70 -
71 - rrdset_done(st_query_thread);
72 -}
73 -
74 -#ifdef NETDATA_INTERNAL_CHECKS
75 -static void aclk_stats_latency(struct aclk_metrics_per_sample *per_sample)
76 -{
77 - static RRDSET *st = NULL;
78 - static RRDDIM *rd_avg = NULL;
79 - static RRDDIM *rd_max = NULL;
80 -
81 - if (unlikely(!st)) {
82 - st = rrdset_create_localhost(
83 - "netdata", "aclk_latency_mqtt", NULL, "aclk", NULL, "ACLK Message Publish Latency", "ms",
84 - "netdata", "stats", 200002, localhost->rrd_update_every, RRDSET_TYPE_LINE);
85 -
86 - rd_avg = rrddim_add(st, "avg", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
87 - rd_max = rrddim_add(st, "max", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
88 - }
89 -
90 - if(per_sample->latency_count)
91 - rrddim_set_by_pointer(st, rd_avg, roundf((float)per_sample->latency_total / per_sample->latency_count));
92 - else
93 - rrddim_set_by_pointer(st, rd_avg, 0);
94 -
95 - rrddim_set_by_pointer(st, rd_max, per_sample->latency_max);
96 -
97 - rrdset_done(st);
98 -}
99 -#endif
100 -
101 -static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample)
102 -{
103 - static RRDSET *st = NULL;
104 - static RRDDIM *rd_rq_rcvd = NULL;
105 - static RRDDIM *rd_rq_err = NULL;
106 -
107 - if (unlikely(!st)) {
108 - st = rrdset_create_localhost(
109 - "netdata", "aclk_cloud_req", NULL, "aclk", NULL, "Requests received from cloud", "req/s",
110 - "netdata", "stats", 200005, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
111 -
112 - rd_rq_rcvd = rrddim_add(st, "received", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
113 - rd_rq_err = rrddim_add(st, "malformed", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
114 - }
115 -
116 - rrddim_set_by_pointer(st, rd_rq_rcvd, per_sample->cloud_req_recvd - per_sample->cloud_req_err);
117 - rrddim_set_by_pointer(st, rd_rq_err, per_sample->cloud_req_err);
118 -
119 - rrdset_done(st);
120 -}
121 -
122 -static void aclk_stats_cloud_req_type(struct aclk_metrics_per_sample *per_sample)
123 -{
124 - static RRDSET *st = NULL;
125 - static RRDDIM *dims[ACLK_QUERY_TYPE_COUNT];
126 -
127 - if (unlikely(!st)) {
128 - st = rrdset_create_localhost(
129 - "netdata", "aclk_processed_query_type", NULL, "aclk", NULL, "Query thread commands processed by their type", "cmd/s",
130 - "netdata", "stats", 200006, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
131 -
132 - for (int i = 0; i < ACLK_QUERY_TYPE_COUNT; i++)
133 - dims[i] = rrddim_add(st, aclk_query_get_name(i, 1), NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
134 -
135 - }
136 -
137 - for (int i = 0; i < ACLK_QUERY_TYPE_COUNT; i++)
138 - rrddim_set_by_pointer(st, dims[i], per_sample->queries_per_type[i]);
139 -
140 - rrdset_done(st);
141 -}
142 -
143 -static char *cloud_req_http_type_names[ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT] = {
144 - "other",
145 - "info",
146 - "data",
147 - "alarms",
148 - "alarm_log",
149 - "chart",
150 - "charts",
151 - "function",
152 - "functions"
153 - // if you change then update `ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT`.
154 -};
155 -
156 -int aclk_cloud_req_http_type_to_idx(const char *name)
157 -{
158 - for (int i = 1; i < ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT; i++)
159 - if (!strcmp(cloud_req_http_type_names[i], name))
160 - return i;
161 - return 0;
162 -}
163 -
164 -static void aclk_stats_cloud_req_http_type(struct aclk_metrics_per_sample *per_sample)
165 -{
166 - static RRDSET *st = NULL;
167 - static RRDDIM *rd_rq_types[ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT];
168 -
169 - if (unlikely(!st)) {
170 - st = rrdset_create_localhost(
171 - "netdata", "aclk_cloud_req_http_type", NULL, "aclk", NULL, "Requests received from cloud via HTTP by their type", "req/s",
172 - "netdata", "stats", 200007, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
173 -
174 - for (int i = 0; i < ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT; i++)
175 - rd_rq_types[i] = rrddim_add(st, cloud_req_http_type_names[i], NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
176 - }
177 -
178 - for (int i = 0; i < ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT; i++)
179 - rrddim_set_by_pointer(st, rd_rq_types[i], per_sample->cloud_req_http_by_type[i]);
180 -
181 - rrdset_done(st);
182 -}
183 -
184 -#define MAX_DIM_NAME 22
185 -static void aclk_stats_query_threads(uint32_t *queries_per_thread)
186 -{
187 - static RRDSET *st = NULL;
188 -
189 - char dim_name[MAX_DIM_NAME];
190 -
191 - if (unlikely(!st)) {
192 - st = rrdset_create_localhost(
193 - "netdata", "aclk_query_threads", NULL, "aclk", NULL, "Queries Processed Per Thread", "req/s",
194 - "netdata", "stats", 200009, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
195 -
196 - for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) {
197 - if (snprintfz(dim_name, MAX_DIM_NAME, "Query %d", i) < 0)
198 - netdata_log_error("snprintf encoding error");
199 - aclk_qt_data[i].dim = rrddim_add(st, dim_name, NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
200 - }
201 - }
202 -
203 - for (int i = 0; i < aclk_stats_cfg.query_thread_count; i++) {
204 - rrddim_set_by_pointer(st, aclk_qt_data[i].dim, queries_per_thread[i]);
205 - }
206 -
207 - rrdset_done(st);
208 -}
209 -
210 -static void aclk_stats_query_time(struct aclk_metrics_per_sample *per_sample)
211 -{
212 - static RRDSET *st = NULL;
213 - static RRDDIM *rd_rq_avg = NULL;
214 - static RRDDIM *rd_rq_max = NULL;
215 - static RRDDIM *rd_rq_total = NULL;
216 -
217 - if (unlikely(!st)) {
218 - st = rrdset_create_localhost(
219 - "netdata", "aclk_query_time", NULL, "aclk", NULL, "Time it took to process cloud requested DB queries", "us",
220 - "netdata", "stats", 200008, localhost->rrd_update_every, RRDSET_TYPE_LINE);
221 -
222 - rd_rq_avg = rrddim_add(st, "avg", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
223 - rd_rq_max = rrddim_add(st, "max", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
224 - rd_rq_total = rrddim_add(st, "total", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
225 - }
226 -
227 - if(per_sample->cloud_q_process_count)
228 - rrddim_set_by_pointer(st, rd_rq_avg, roundf((float)per_sample->cloud_q_process_total / per_sample->cloud_q_process_count));
229 - else
230 - rrddim_set_by_pointer(st, rd_rq_avg, 0);
231 - rrddim_set_by_pointer(st, rd_rq_max, per_sample->cloud_q_process_max);
232 - rrddim_set_by_pointer(st, rd_rq_total, per_sample->cloud_q_process_total);
233 -
234 - rrdset_done(st);
235 -}
236 -
237 -const char *rx_handler_get_name(size_t i);
238 -static void aclk_stats_newproto_rx(uint32_t *rx_msgs_sample)
239 -{
240 - static RRDSET *st = NULL;
241 -
242 - if (unlikely(!st)) {
243 - st = rrdset_create_localhost(
244 - "netdata", "aclk_protobuf_rx_types", NULL, "aclk", NULL, "Received new cloud architecture messages by their type.", "msg/s",
245 - "netdata", "stats", 200010, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
246 -
247 - for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++) {
248 - aclk_stats_cfg.rx_msg_dims[i] = rrddim_add(st, rx_handler_get_name(i), NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
249 - }
250 - }
251 -
252 - for (unsigned int i = 0; i < aclk_stats_cfg.proto_hdl_cnt; i++)
253 - rrddim_set_by_pointer(st, aclk_stats_cfg.rx_msg_dims[i], rx_msgs_sample[i]);
254 -
255 - rrdset_done(st);
256 -}
257 -
258 -static void aclk_stats_mqtt_wss(struct mqtt_wss_stats *stats)
259 -{
260 - static RRDSET *st = NULL;
261 - static RRDDIM *rd_sent = NULL;
262 - static RRDDIM *rd_recvd = NULL;
263 - static uint64_t sent = 0;
264 - static uint64_t recvd = 0;
265 -
266 - static RRDSET *st_txbuf_perc = NULL;
267 - static RRDDIM *rd_txbuf_perc = NULL;
268 -
269 - static RRDSET *st_txbuf = NULL;
270 - static RRDDIM *rd_tx_buffer_usable = NULL;
271 - static RRDDIM *rd_tx_buffer_reclaimable = NULL;
272 - static RRDDIM *rd_tx_buffer_used = NULL;
273 - static RRDDIM *rd_tx_buffer_free = NULL;
274 - static RRDDIM *rd_tx_buffer_size = NULL;
275 -
276 - static RRDSET *st_timing = NULL;
277 - static RRDDIM *rd_keepalive = NULL;
278 - static RRDDIM *rd_read_socket = NULL;
279 - static RRDDIM *rd_write_socket = NULL;
280 - static RRDDIM *rd_process_websocket = NULL;
281 - static RRDDIM *rd_process_mqtt = NULL;
282 -
283 - sent += stats->bytes_tx;
284 - recvd += stats->bytes_rx;
285 -
286 - if (unlikely(!st)) {
287 - st = rrdset_create_localhost(
288 - "netdata", "aclk_openssl_bytes", NULL, "aclk", NULL, "Received and Sent bytes.", "B/s",
289 - "netdata", "stats", 200011, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
290 -
291 - rd_sent = rrddim_add(st, "sent", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
292 - rd_recvd = rrddim_add(st, "received", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
293 - }
294 -
295 - if (unlikely(!st_txbuf_perc)) {
296 - st_txbuf_perc = rrdset_create_localhost(
297 - "netdata", "aclk_mqtt_tx_perc", NULL, "aclk", NULL, "Actively used percentage of MQTT Tx Buffer,", "%",
298 - "netdata", "stats", 200012, localhost->rrd_update_every, RRDSET_TYPE_LINE);
299 -
300 - rd_txbuf_perc = rrddim_add(st_txbuf_perc, "used", NULL, 1, 100, RRD_ALGORITHM_ABSOLUTE);
301 - }
302 -
303 - if (unlikely(!st_txbuf)) {
304 - st_txbuf = rrdset_create_localhost(
305 - "netdata", "aclk_mqtt_tx_queue", NULL, "aclk", NULL, "State of transmit MQTT queue.", "B",
306 - "netdata", "stats", 200013, localhost->rrd_update_every, RRDSET_TYPE_LINE);
307 -
308 - rd_tx_buffer_usable = rrddim_add(st_txbuf, "usable", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
309 - rd_tx_buffer_reclaimable = rrddim_add(st_txbuf, "reclaimable", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
310 - rd_tx_buffer_used = rrddim_add(st_txbuf, "used", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
311 - rd_tx_buffer_free = rrddim_add(st_txbuf, "free", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
312 - rd_tx_buffer_size = rrddim_add(st_txbuf, "size", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
313 - }
314 -
315 - if (unlikely(!st_timing)) {
316 - st_timing = rrdset_create_localhost(
317 - "netdata", "aclk_mqtt_wss_time", NULL, "aclk", NULL, "Time spent handling MQTT, WSS, SSL and network communication.", "us",
318 - "netdata", "stats", 200014, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
319 -
320 - rd_keepalive = rrddim_add(st_timing, "keep-alive", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
321 - rd_read_socket = rrddim_add(st_timing, "socket_read_ssl", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
322 - rd_write_socket = rrddim_add(st_timing, "socket_write_ssl", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
323 - rd_process_websocket = rrddim_add(st_timing, "process_websocket", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
324 - rd_process_mqtt = rrddim_add(st_timing, "process_mqtt", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
325 - }
326 -
327 - rrddim_set_by_pointer(st, rd_sent, sent);
328 - rrddim_set_by_pointer(st, rd_recvd, recvd);
329 -
330 - float usage = ((float)stats->mqtt.tx_buffer_free + stats->mqtt.tx_buffer_reclaimable) / stats->mqtt.tx_buffer_size;
331 - usage = (1 - usage) * 10000;
332 - rrddim_set_by_pointer(st_txbuf_perc, rd_txbuf_perc, usage);
333 -
334 - rrddim_set_by_pointer(st_txbuf, rd_tx_buffer_usable, stats->mqtt.tx_buffer_reclaimable + stats->mqtt.tx_buffer_free);
335 - rrddim_set_by_pointer(st_txbuf, rd_tx_buffer_reclaimable, stats->mqtt.tx_buffer_reclaimable);
336 - rrddim_set_by_pointer(st_txbuf, rd_tx_buffer_used, stats->mqtt.tx_buffer_used);
337 - rrddim_set_by_pointer(st_txbuf, rd_tx_buffer_free, stats->mqtt.tx_buffer_free);
338 - rrddim_set_by_pointer(st_txbuf, rd_tx_buffer_size, stats->mqtt.tx_buffer_size);
339 -
340 - rrddim_set_by_pointer(st_timing, rd_keepalive, stats->time_keepalive);
341 - rrddim_set_by_pointer(st_timing, rd_read_socket, stats->time_read_socket);
342 - rrddim_set_by_pointer(st_timing, rd_write_socket, stats->time_write_socket);
343 - rrddim_set_by_pointer(st_timing, rd_process_websocket, stats->time_process_websocket);
344 - rrddim_set_by_pointer(st_timing, rd_process_mqtt, stats->time_process_mqtt);
345 -
346 - rrdset_done(st);
347 - rrdset_done(st_txbuf_perc);
348 - rrdset_done(st_txbuf);
349 - rrdset_done(st_timing);
350 -}
351 -
352 -void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt)
353 -{
354 - aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data));
355 - aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t));
356 - aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t));
357 -
358 - memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
359 -
360 - aclk_stats_cfg.proto_hdl_cnt = proto_hdl_cnt;
361 - aclk_stats_cfg.aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample));
362 - aclk_proto_rx_msgs_sample = callocz(proto_hdl_cnt, sizeof(*aclk_proto_rx_msgs_sample));
363 - aclk_stats_cfg.rx_msg_dims = callocz(proto_hdl_cnt, sizeof(RRDDIM*));
364 -}
365 -
366 -void aclk_stats_thread_cleanup()
367 -{
368 - freez(aclk_stats_cfg.rx_msg_dims);
369 - freez(aclk_proto_rx_msgs_sample);
370 - freez(aclk_stats_cfg.aclk_proto_rx_msgs_sample);
371 - freez(aclk_qt_data);
372 - freez(aclk_queries_per_thread);
373 - freez(aclk_queries_per_thread_sample);
374 -}
375 -
376 -void *aclk_stats_main_thread(void *ptr)
377 -{
378 - struct aclk_stats_thread *args = ptr;
379 -
380 - aclk_stats_cfg.query_thread_count = args->query_thread_count;
381 -
382 - heartbeat_t hb;
383 - heartbeat_init(&hb);
384 - usec_t step_ut = localhost->rrd_update_every * USEC_PER_SEC;
385 -
386 - struct aclk_metrics_per_sample per_sample;
387 - struct aclk_metrics permanent;
388 -
389 - while (service_running(SERVICE_ACLK | SERVICE_COLLECTORS)) {
390 -
391 - // ------------------------------------------------------------------------
392 - // Wait for the next iteration point.
393 -
394 - heartbeat_next(&hb, step_ut);
395 -
396 - if (!service_running(SERVICE_ACLK | SERVICE_COLLECTORS)) break;
397 -
398 - ACLK_STATS_LOCK;
399 - // to not hold lock longer than necessary, especially not to hold it
400 - // during database rrd* operations
401 - memcpy(&per_sample, &aclk_metrics_per_sample, sizeof(struct aclk_metrics_per_sample));
402 -
403 - memcpy(aclk_stats_cfg.aclk_proto_rx_msgs_sample, aclk_proto_rx_msgs_sample, sizeof(*aclk_proto_rx_msgs_sample) * aclk_stats_cfg.proto_hdl_cnt);
404 - memset(aclk_proto_rx_msgs_sample, 0, sizeof(*aclk_proto_rx_msgs_sample) * aclk_stats_cfg.proto_hdl_cnt);
405 -
406 - memcpy(&permanent, &aclk_metrics, sizeof(struct aclk_metrics));
407 - memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
408 -
409 - memcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count);
410 - memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * aclk_stats_cfg.query_thread_count);
411 - ACLK_STATS_UNLOCK;
412 -
413 - aclk_stats_collect(&per_sample, &permanent);
414 - aclk_stats_query_queue(&per_sample);
415 -#ifdef NETDATA_INTERNAL_CHECKS
416 - aclk_stats_latency(&per_sample);
417 -#endif
418 -
419 - aclk_stats_cloud_req(&per_sample);
420 - aclk_stats_cloud_req_type(&per_sample);
421 - aclk_stats_cloud_req_http_type(&per_sample);
422 -
423 - aclk_stats_query_threads(aclk_queries_per_thread_sample);
424 -
425 - aclk_stats_query_time(&per_sample);
426 -
427 - struct mqtt_wss_stats mqtt_wss_stats = mqtt_wss_get_stats(args->client);
428 - aclk_stats_mqtt_wss(&mqtt_wss_stats);
429 -
430 - aclk_stats_newproto_rx(aclk_stats_cfg.aclk_proto_rx_msgs_sample);
431 - }
432 -
433 - return 0;
434 -}
435 -
436 -void aclk_stats_upd_online(int online) {
437 - if(!aclk_stats_enabled)
438 - return;
439 -
440 - ACLK_STATS_LOCK;
441 - aclk_metrics.online = online;
442 -
443 - if(!online)
444 - aclk_metrics_per_sample.offline_during_sample = 1;
445 - ACLK_STATS_UNLOCK;
446 -}
447 -
448 -#ifdef NETDATA_INTERNAL_CHECKS
449 -static usec_t pub_time[UINT16_MAX + 1] = {0};
450 -void aclk_stats_msg_published(uint16_t id)
451 -{
452 - ACLK_STATS_LOCK;
453 - pub_time[id] = now_boottime_usec();
454 - ACLK_STATS_UNLOCK;
455 -}
456 -
457 -void aclk_stats_msg_puback(uint16_t id)
458 -{
459 - ACLK_STATS_LOCK;
460 - usec_t t;
461 -
462 - if (!aclk_stats_enabled) {
463 - ACLK_STATS_UNLOCK;
464 - return;
465 - }
466 -
467 - if (unlikely(!pub_time[id])) {
468 - ACLK_STATS_UNLOCK;
469 - netdata_log_error("Received PUBACK for unknown message?!");
470 - return;
471 - }
472 -
473 - t = now_boottime_usec() - pub_time[id];
474 - t /= USEC_PER_MS;
475 - pub_time[id] = 0;
476 - if (aclk_metrics_per_sample.latency_max < t)
477 - aclk_metrics_per_sample.latency_max = t;
478 -
479 - aclk_metrics_per_sample.latency_total += t;
480 - aclk_metrics_per_sample.latency_count++;
481 - ACLK_STATS_UNLOCK;
482 -}
483 -#endif /* NETDATA_INTERNAL_CHECKS */
src/aclk/aclk_stats.h deleted
-77
@@ -1,77 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -#ifndef NETDATA_ACLK_STATS_H
4 -#define NETDATA_ACLK_STATS_H
5 -
6 -#include "daemon/common.h"
7 -#include "libnetdata/libnetdata.h"
8 -#include "aclk_query_queue.h"
9 -#include "mqtt_websockets/mqtt_wss_client.h"
10 -
11 -extern netdata_mutex_t aclk_stats_mutex;
12 -
13 -#define ACLK_STATS_LOCK netdata_mutex_lock(&aclk_stats_mutex)
14 -#define ACLK_STATS_UNLOCK netdata_mutex_unlock(&aclk_stats_mutex)
15 -
16 -// if you change update `cloud_req_http_type_names`.
17 -#define ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT 9
18 -
19 -int aclk_cloud_req_http_type_to_idx(const char *name);
20 -
21 -struct aclk_stats_thread {
22 - ND_THREAD *thread;
23 - int query_thread_count;
24 - mqtt_wss_client client;
25 -};
26 -
27 -// preserve between samples
28 -struct aclk_metrics {
29 - volatile uint8_t online;
30 -};
31 -
32 -// reset to 0 on every sample
33 -extern struct aclk_metrics_per_sample {
34 - /* in the unlikely event of ACLK disconnecting
35 - and reconnecting under 1 sampling rate
36 - we want to make sure we record the disconnection
37 - despite it being then seemingly longer in graph */
38 - volatile uint8_t offline_during_sample;
39 -
40 - volatile uint32_t queries_queued;
41 - volatile uint32_t queries_dispatched;
42 -
43 -#ifdef NETDATA_INTERNAL_CHECKS
44 - volatile uint32_t latency_max;
45 - volatile uint32_t latency_total;
46 - volatile uint32_t latency_count;
47 -#endif
48 -
49 - volatile uint32_t cloud_req_recvd;
50 - volatile uint32_t cloud_req_err;
51 -
52 - // query types.
53 - volatile uint32_t queries_per_type[ACLK_QUERY_TYPE_COUNT];
54 -
55 - // HTTP-specific request types.
56 - volatile uint32_t cloud_req_http_by_type[ACLK_STATS_CLOUD_HTTP_REQ_TYPE_CNT];
57 -
58 - volatile uint32_t cloud_q_process_total;
59 - volatile uint32_t cloud_q_process_count;
60 - volatile uint32_t cloud_q_process_max;
61 -} aclk_metrics_per_sample;
62 -
63 -extern uint32_t *aclk_proto_rx_msgs_sample;
64 -
65 -extern uint32_t *aclk_queries_per_thread;
66 -
67 -void *aclk_stats_main_thread(void *ptr);
68 -void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt);
69 -void aclk_stats_thread_cleanup();
70 -void aclk_stats_upd_online(int online);
71 -
72 -#ifdef NETDATA_INTERNAL_CHECKS
73 -void aclk_stats_msg_published(uint16_t id);
74 -void aclk_stats_msg_puback(uint16_t id);
75 -#endif /* NETDATA_INTERNAL_CHECKS */
76 -
77 -#endif /* NETDATA_ACLK_STATS_H */
src/aclk/aclk_tx_msgs.c
+24 -33
@@ -1,9 +1,7 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_tx_msgs.h"
4 -#include "daemon/common.h"
4 #include "aclk_util.h"
6 -#include "aclk_stats.h"
5 #include "aclk.h"
6 #include "aclk_capas.h"
7
@@ -13,9 +11,6 @@
11 #pragma region aclk_tx_msgs helper functions
12 #endif
13
16 -// version for aclk legacy (old cloud arch)
17 -#define ACLK_VERSION 2
18 -
14 static void freez_aclk_publish5a(void *ptr) {
15 freez(ptr);
16 }
@@ -23,6 +18,8 @@ static void freez_aclk_publish5b(void *ptr) {
18 freez(ptr);
19 }
20
21 +#define ACLK_HEADER_VERSION (2)
22 +
23 uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname)
24 {
25 #ifndef ACLK_LOG_CONVERSATION_DIR
@@ -38,10 +35,6 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
35
36 mqtt_wss_publish5(client, (char*)topic, NULL, msg, &freez_aclk_publish5a, msg_len, MQTT_WSS_PUB_QOS1, &packet_id);
37
41 -#ifdef NETDATA_INTERNAL_CHECKS
42 - aclk_stats_msg_published(packet_id);
43 -#endif
44 -
38 if (aclklog_enabled) {
39 char *json = protomsg_to_json(msg, msg_len, msgname);
40 log_aclk_message_bin(json, strlen(json), 1, topic, msgname);
@@ -51,14 +44,13 @@ uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, s
44 return packet_id;
45 }
46
54 -#define TOPIC_MAX_LEN 512
47 #define V2_BIN_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A"
56 -static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_object *msg, const char *topic, const void *payload, size_t payload_len)
48 +static short aclk_send_message_with_bin_payload(mqtt_wss_client client, json_object *msg, const char *topic, const void *payload, size_t payload_len)
49 {
50 uint16_t packet_id;
51 const char *str;
52 char *full_msg = NULL;
61 - int len;
53 + size_t len;
54
55 if (unlikely(!topic || topic[0] != '/')) {
56 netdata_log_error("Full topic required!");
@@ -78,7 +70,7 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
70 json_object_put(msg);
71
72 if (payload_len) {
81 - memcpy(&full_msg[len], V2_BIN_PAYLOAD_SEPARATOR, strlen(V2_BIN_PAYLOAD_SEPARATOR));
73 + memcpy(&full_msg[len], V2_BIN_PAYLOAD_SEPARATOR, sizeof(V2_BIN_PAYLOAD_SEPARATOR) - 1);
74 len += strlen(V2_BIN_PAYLOAD_SEPARATOR);
75 memcpy(&full_msg[len], payload, payload_len);
76 }
@@ -88,10 +80,6 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
80 if (rc == MQTT_WSS_ERR_TOO_BIG_FOR_SERVER)
81 return HTTP_RESP_CONTENT_TOO_LONG;
82
91 -#ifdef NETDATA_INTERNAL_CHECKS
92 - aclk_stats_msg_published(packet_id);
93 -#endif
94 -
83 return 0;
84 }
85
@@ -99,12 +87,14 @@ static int aclk_send_message_with_bin_payload(mqtt_wss_client client, json_objec
87 * Creates universal header common for all ACLK messages. User gets ownership of json object created.
88 * Usually this is freed by send function after message has been sent.
89 */
102 -static struct json_object *create_hdr(const char *type, const char *msg_id, time_t ts_secs, usec_t ts_us, int version)
90 +static struct json_object *create_hdr(const char *type, const char *msg_id)
91 {
92 nd_uuid_t uuid;
105 - char uuid_str[36 + 1];
93 + char uuid_str[UUID_STR_LEN];
94 json_object *tmp;
95 json_object *obj = json_object_new_object();
96 + time_t ts_secs;
97 + usec_t ts_us;
98
99 tmp = json_object_new_string(type);
100 json_object_object_add(obj, "type", tmp);
@@ -115,11 +105,9 @@ static struct json_object *create_hdr(const char *type, const char *msg_id, time
105 msg_id = uuid_str;
106 }
107
118 - if (ts_secs == 0) {
119 - ts_us = now_realtime_usec();
120 - ts_secs = ts_us / USEC_PER_SEC;
121 - ts_us = ts_us % USEC_PER_SEC;
122 - }
108 + ts_us = now_realtime_usec();
109 + ts_secs = ts_us / USEC_PER_SEC;
110 + ts_us = ts_us % USEC_PER_SEC;
111
112 tmp = json_object_new_string(msg_id);
113 json_object_object_add(obj, "msg-id", tmp);
@@ -144,7 +132,7 @@ static struct json_object *create_hdr(const char *type, const char *msg_id, time
132 tmp = json_object_new_int64(aclk_session_us);
133 json_object_object_add(obj, "connect-offset-usec", tmp);
134
147 - tmp = json_object_new_int(version);
135 + tmp = json_object_new_int(ACLK_HEADER_VERSION);
136 json_object_object_add(obj, "version", tmp);
137
138 return obj;
@@ -161,7 +149,7 @@ static struct json_object *create_hdr(const char *type, const char *msg_id, time
149 void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char *msg_id, int http_code, int ec, const char* emsg, const char *payload, size_t payload_len)
150 {
151 json_object *tmp, *msg;
164 - msg = create_hdr("http", msg_id, 0, 0, 2);
152 + msg = create_hdr("http", msg_id);
153 tmp = json_object_new_int(http_code);
154 json_object_object_add(msg, "http-code", tmp);
155
@@ -176,11 +164,12 @@ void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char
164 }
165 }
166
179 -int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len)
167 +short aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created,
168 + short http_code, const char *payload, size_t payload_len)
169 {
170 json_object *tmp, *msg;
171
183 - msg = create_hdr("http", msg_id, 0, 0, 2);
172 + msg = create_hdr("http", msg_id);
173
174 tmp = json_object_new_int64(t_exec);
175 json_object_object_add(msg, "t-exec", tmp);
@@ -191,7 +180,7 @@ int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_
180 tmp = json_object_new_int(http_code);
181 json_object_object_add(msg, "http-code", tmp);
182
194 - int rc = aclk_send_message_with_bin_payload(client, msg, topic, payload, payload_len);
183 + short rc = aclk_send_message_with_bin_payload(client, msg, topic, payload, payload_len);
184
185 switch (rc) {
186 case HTTP_RESP_CONTENT_TOO_LONG:
@@ -200,12 +189,14 @@ int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_
189 case HTTP_RESP_INTERNAL_SERVER_ERROR:
190 aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_FAIL_TOPIC, CLOUD_EMSG_FAIL_TOPIC, payload, payload_len);
191 break;
203 - case HTTP_RESP_GATEWAY_TIMEOUT:
204 - case HTTP_RESP_SERVICE_UNAVAILABLE:
205 - aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, payload, payload_len);
192 +// case HTTP_RESP_SERVICE_UNAVAILABLE:
193 +// aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, payload, payload_len);
194 +// break;
195 + default:
196 + rc = http_code;
197 break;
198 }
208 - return rc ? rc : http_code;
199 + return rc;
200 }
201
202 uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable) {
src/aclk/aclk_tx_msgs.h
+2 -1
@@ -12,7 +12,8 @@
12 uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname);
13
14 void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char *msg_id, int http_code, int ec, const char* emsg, const char *payload, size_t payload_len);
15 -int aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, int http_code, const char *payload, size_t payload_len);
15 +short aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created,
16 + short http_code, const char *payload, size_t payload_len);
17
18 uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable);
19 char *aclk_generate_lwt(size_t *size);
src/aclk/https_client.c
+2 -1
@@ -105,7 +105,8 @@ static int parse_http_hdr(rbuf_t buf, http_parse_ctx *parse_ctx)
105 int idx, idx_end;
106 char buf_key[HTTP_HDR_BUFFER_SIZE];
107 char buf_val[HTTP_HDR_BUFFER_SIZE];
108 - char *ptr = buf_key;
108 + char *ptr;
109 +
110 if (!rbuf_find_bytes(buf, HTTP_LINE_TERM, strlen(HTTP_LINE_TERM), &idx_end)) {
111 netdata_log_error("CRLF expected");
112 return 1;
src/daemon/global_statistics.c
+1 -2
@@ -3502,8 +3502,7 @@ static struct worker_utilization all_workers_utilization[] = {
3502 { .name = "DBENGINE", .family = "workers dbengine instances", .priority = 1000000 },
3503 { .name = "LIBUV", .family = "workers libuv threadpool", .priority = 1000000 },
3504 { .name = "WEB", .family = "workers web server", .priority = 1000000 },
3505 - { .name = "ACLKQUERY", .family = "workers aclk query", .priority = 1000000 },
3506 - { .name = "ACLKSYNC", .family = "workers aclk host sync", .priority = 1000000 },
3505 + { .name = "ACLKSYNC", .family = "workers aclk sync", .priority = 1000000 },
3506 { .name = "METASYNC", .family = "workers metadata sync", .priority = 1000000 },
3507 { .name = "PLUGINSD", .family = "workers plugins.d", .priority = 1000000 },
3508 { .name = "STATSD", .family = "workers plugin statsd", .priority = 1000000 },
src/daemon/main.c
+1 -1
@@ -362,7 +362,7 @@ void netdata_cleanup_and_exit(int ret, const char *action, const char *action_re
362 watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);
363
364 service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS |
365 - ABILITY_STREAMING_CONNECTIONS | SERVICE_ACLK | SERVICE_ACLKSYNC);
365 + ABILITY_STREAMING_CONNECTIONS | SERVICE_ACLK);
366 watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK);
367
368 service_wait_exit(SERVICE_MAINTENANCE, 3 * USEC_PER_SEC);
src/daemon/main.h
+1 -2
@@ -23,8 +23,7 @@ typedef enum {
23 SERVICE_CONTEXT = (1 << 10),
24 SERVICE_ANALYTICS = (1 << 11),
25 SERVICE_EXPORTERS = (1 << 12),
26 - SERVICE_ACLKSYNC = (1 << 13),
27 - SERVICE_HTTPD = (1 << 14)
26 + SERVICE_HTTPD = (1 << 13)
27 } SERVICE_TYPE;
28
29 typedef enum {
src/database/contexts/query_target.c
+2 -2
@@ -1035,8 +1035,8 @@ void query_target_generate_name(QUERY_TARGET *qt) {
1035 }
1036
1037 QUERY_TARGET *query_target_create(QUERY_TARGET_REQUEST *qtr) {
1038 - if(!service_running(ABILITY_DATA_QUERIES))
1039 - return NULL;
1038 + //if(!service_running(ABILITY_DATA_QUERIES))
1039 + // return NULL;
1040
1041 QUERY_TARGET *qt = query_target_get();
1042
src/database/sqlite/sqlite_aclk.c
+118 -9
@@ -3,7 +3,14 @@
3 #include "sqlite_functions.h"
4 #include "sqlite_aclk.h"
5
6 +void sanity_check(void) {
7 + // make sure the compiler will stop on misconfigurations
8 + BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
9 +}
10 +
11 #include "sqlite_aclk_node.h"
12 +#include "../aclk_query_queue.h"
13 +#include "../aclk_query.h"
14
15 struct aclk_sync_config_s {
16 uv_thread_t thread;
@@ -11,15 +18,12 @@ struct aclk_sync_config_s {
18 uv_timer_t timer_req;
19 uv_async_t async;
20 bool initialized;
21 + mqtt_wss_client client;
22 + int aclk_queries_running;
23 SPINLOCK cmd_queue_lock;
24 struct aclk_database_cmd *cmd_base;
25 } aclk_sync_config = { 0 };
26
18 -void sanity_check(void) {
19 - // make sure the compiler will stop on misconfigurations
20 - BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
21 -}
22 -
27 static struct aclk_database_cmd aclk_database_deq_cmd(void)
28 {
29 struct aclk_database_cmd ret = { 0 };
@@ -267,6 +271,64 @@ static void timer_cb(uv_timer_t *handle)
271 }
272 }
273
274 +struct aclk_query_payload {
275 + uv_work_t request;
276 + void *data;
277 + struct aclk_sync_config_s *config;
278 +};
279 +
280 +static void after_aclk_run_query_job(uv_work_t *req, int status __maybe_unused)
281 +{
282 + worker_is_busy(ACLK_QUERY_EXECUTE);
283 + struct aclk_query_payload *payload = req->data;
284 + struct aclk_sync_config_s *config = payload->config;
285 + config->aclk_queries_running--;
286 + freez(payload);
287 +}
288 +
289 +static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t query)
290 +{
291 + if (query->type == UNKNOWN || query->type >= ACLK_QUERY_TYPE_COUNT) {
292 + error_report("Unknown query in query queue. %u", query->type);
293 + return;
294 + }
295 +
296 + if (query->type == HTTP_API_V2) {
297 + http_api_v2(config->client, query);
298 + } else {
299 + send_bin_msg(config->client, query);
300 + }
301 + aclk_query_free(query);
302 +}
303 +
304 +static void aclk_run_query_job(uv_work_t *req)
305 +{
306 + struct aclk_query_payload *payload = req->data;
307 + struct aclk_sync_config_s *config = payload->config;
308 + aclk_query_t query = (aclk_query_t) payload->data;
309 +
310 + aclk_run_query(config, query);
311 +}
312 +
313 +static int read_query_thread_count()
314 +{
315 + int threads = MIN(get_netdata_cpus()/2, 6);
316 + threads = MAX(threads, 2);
317 + threads = config_get_number(CONFIG_SECTION_CLOUD, "query thread count", threads);
318 + if(threads < 1) {
319 + netdata_log_error("You need at least one query thread. Overriding configured setting of \"%d\"", threads);
320 + threads = 1;
321 + config_set_number(CONFIG_SECTION_CLOUD, "query thread count", threads);
322 + }
323 + else {
324 + if (threads > libuv_worker_threads / 2) {
325 + threads = MAX(libuv_worker_threads / 2, 2);
326 + config_set_number(CONFIG_SECTION_CLOUD, "query thread count", threads);
327 + }
328 + }
329 + return threads;
330 +}
331 +
332 static void aclk_synchronization(void *arg)
333 {
334 struct aclk_sync_config_s *config = arg;
@@ -278,6 +340,8 @@ static void aclk_synchronization(void *arg)
340 worker_register_job_name(ACLK_DATABASE_NODE_STATE, "node state");
341 worker_register_job_name(ACLK_DATABASE_PUSH_ALERT, "alert push");
342 worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_CONFIG, "alert conf push");
343 + worker_register_job_name(ACLK_QUERY_EXECUTE, "query execute");
344 + worker_register_job_name(ACLK_QUERY_EXECUTE_SYNC, "query execute sync");
345 worker_register_job_name(ACLK_DATABASE_TIMER, "timer");
346
347 uv_loop_t *loop = &config->loop;
@@ -294,7 +358,10 @@ static void aclk_synchronization(void *arg)
358
359 sql_delete_aclk_table_list();
360
297 - while (likely(service_running(SERVICE_ACLKSYNC))) {
361 + int query_thread_count = read_query_thread_count();
362 + netdata_log_info("Starting ACLK synchronization thread with %d parallel query threads", query_thread_count);
363 +
364 + while (likely(service_running(SERVICE_ACLK))) {
365 enum aclk_database_opcode opcode;
366 worker_is_idle();
367 uv_run(loop, UV_RUN_DEFAULT);
@@ -303,16 +370,15 @@ static void aclk_synchronization(void *arg)
370 do {
371 struct aclk_database_cmd cmd = aclk_database_deq_cmd();
372
306 - if (unlikely(!service_running(SERVICE_ACLKSYNC)))
373 + if (unlikely(!service_running(SERVICE_ACLK)))
374 break;
375
376 opcode = cmd.opcode;
377
311 - if(likely(opcode != ACLK_DATABASE_NOOP))
378 + if(likely(opcode != ACLK_DATABASE_NOOP && opcode != ACLK_QUERY_EXECUTE))
379 worker_is_busy(opcode);
380
381 switch (opcode) {
315 - default:
382 case ACLK_DATABASE_NOOP:
383 /* the command queue was empty, do nothing */
384 break;
@@ -336,6 +402,35 @@ static void aclk_synchronization(void *arg)
402 case ACLK_DATABASE_PUSH_ALERT:
403 aclk_push_alert_events_for_all_hosts();
404 break;
405 +
406 + case ACLK_MQTT_WSS_CLIENT:
407 + config->client = (mqtt_wss_client) cmd.param[0];
408 + break;
409 +
410 + case ACLK_QUERY_EXECUTE:;
411 + aclk_query_t query = (aclk_query_t)cmd.param[0];
412 +
413 + struct aclk_query_payload *payload = NULL;
414 + config->aclk_queries_running++;
415 + bool execute_now = (config->aclk_queries_running > query_thread_count);
416 + if (!execute_now) {
417 + payload = mallocz(sizeof(*payload));
418 + payload->request.data = payload;
419 + payload->config = config;
420 + payload->data = query;
421 + execute_now = uv_queue_work(loop, &payload->request, aclk_run_query_job, after_aclk_run_query_job);
422 + }
423 +
424 + if (execute_now) {
425 + worker_is_busy(ACLK_QUERY_EXECUTE_SYNC);
426 + aclk_run_query(config, query);
427 + freez(payload);
428 + config->aclk_queries_running--;
429 + }
430 + break;
431 +
432 + default:
433 + break;
434 }
435 } while (opcode != ACLK_DATABASE_NOOP);
436 }
@@ -445,6 +540,20 @@ void aclk_push_alert_config(const char *node_id, const char *config_hash)
540 queue_aclk_sync_cmd(ACLK_DATABASE_PUSH_ALERT_CONFIG, strdupz(node_id), strdupz(config_hash));
541 }
542
543 +void aclk_execute_query(aclk_query_t query)
544 +{
545 + if (unlikely(!aclk_sync_config.initialized))
546 + return;
547 +
548 + queue_aclk_sync_cmd(ACLK_QUERY_EXECUTE, query, NULL);
549 +}
550 +
551 +void aclk_query_init(mqtt_wss_client client) {
552 +
553 + queue_aclk_sync_cmd(ACLK_MQTT_WSS_CLIENT, client, NULL);
554 +}
555 +
556 +
557 void schedule_node_info_update(RRDHOST *host __maybe_unused)
558 {
559 if (unlikely(!host))
src/database/sqlite/sqlite_aclk.h
+3
@@ -21,6 +21,9 @@ enum aclk_database_opcode {
21 ACLK_DATABASE_PUSH_ALERT,
22 ACLK_DATABASE_PUSH_ALERT_CONFIG,
23 ACLK_DATABASE_NODE_UNREGISTER,
24 + ACLK_MQTT_WSS_CLIENT,
25 + ACLK_QUERY_EXECUTE,
26 + ACLK_QUERY_EXECUTE_SYNC,
27 ACLK_DATABASE_TIMER,
28
29 // leave this last
src/web/api/v1/api_v1_ml_info.c
+2 -1
@@ -21,7 +21,8 @@ int api_v1_ml_info(RRDHOST *host, struct web_client *w, char *url) {
21
22 return HTTP_RESP_OK;
23 #else
24 + UNUSED(host);
25 + UNUSED(w);
26 return HTTP_RESP_SERVICE_UNAVAILABLE;
27 #endif // ENABLE_ML
28 }
27 -