adds support for multiple ACLK query processing threads (#9355)
Timotej S committed
Jul 10, 2020 at 15:31 UTC
42aa54eaf6a64348f23cc0c9ed0a071b79bc9cad
11 files changed
+896
-663
CMakeLists.txt
+2
@@ -665,6 +665,8 @@ set(ACLK_PLUGIN_FILES
665
aclk/aclk_common.h
666
aclk/agent_cloud_link.c
667
aclk/agent_cloud_link.h
668
+ aclk/aclk_query.c
669
+ aclk/aclk_query.h
670
aclk/aclk_lws_wss_client.c
671
aclk/aclk_lws_wss_client.h
672
aclk/aclk_lws_https_client.c
Makefile.am
+2
@@ -518,6 +518,8 @@ if ENABLE_ACLK
518
ACLK_FILES += \
519
aclk/agent_cloud_link.c \
520
aclk/agent_cloud_link.h \
521
+ aclk/aclk_query.c \
522
+ aclk/aclk_query.h \
523
aclk/mqtt.c \
524
aclk/mqtt.h \
525
aclk/aclk_lws_wss_client.c \
aclk/README.md
+10
@@ -38,6 +38,16 @@ configuration uses two settings:
38
If your Agent needs to use a proxy to access the internet, you must [set up a proxy for
39
claiming](/claim/README.md#claim-through-a-proxy).
40
41
+You can configure following keys in the `netdata.conf` section `[cloud]`:
42
+```
43
+[cloud]
44
+ statistics = yes
45
+ query thread count = 2
46
+```
47
+
48
+- `statistics` enables/disables ACLK related statistics and their charts. You can disable this to save some space in the database and slightly reduce memory usage of Netdata Agent.
49
+- `query thread count` specifies the number of threads to process cloud queries. Increasing this setting is useful for nodes with many children (streaming), which can expect to handle more queries (and/or more complicated queries).
50
+
51
## Disable the ACLK
52
53
You have two options if you prefer to disable the ACLK and not use Netdata Cloud.
aclk/aclk_common.c
+8
@@ -2,6 +2,14 @@
2
3
#include "../daemon/common.h"
4
5
+netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
6
+
7
+struct aclk_shared_state aclk_shared_state = {
8
+ .metadata_submitted = ACLK_METADATA_REQUIRED,
9
+ .agent_state = AGENT_INITIALIZING,
10
+ .last_popcorn_interrupt = 0
11
+};
12
+
13
struct {
14
ACLK_PROXY_TYPE type;
15
const char *url_str;
aclk/aclk_common.h
+30
@@ -3,6 +3,36 @@
3
4
#include "libnetdata/libnetdata.h"
5
6
+extern netdata_mutex_t aclk_shared_state_mutex;
7
+#define ACLK_SHARED_STATE_LOCK netdata_mutex_lock(&aclk_shared_state_mutex)
8
+#define ACLK_SHARED_STATE_UNLOCK netdata_mutex_unlock(&aclk_shared_state_mutex)
9
+
10
+typedef enum aclk_cmd {
11
+ ACLK_CMD_CLOUD,
12
+ ACLK_CMD_ONCONNECT,
13
+ ACLK_CMD_INFO,
14
+ ACLK_CMD_CHART,
15
+ ACLK_CMD_CHARTDEL,
16
+ ACLK_CMD_ALARM,
17
+ ACLK_CMD_MAX
18
+} ACLK_CMD;
19
+
20
+typedef enum aclk_metadata_state {
21
+ ACLK_METADATA_REQUIRED,
22
+ ACLK_METADATA_CMD_QUEUED,
23
+ ACLK_METADATA_SENT
24
+} ACLK_METADATA_STATE;
25
+
26
+typedef enum aclk_agent_state {
27
+ AGENT_INITIALIZING,
28
+ AGENT_STABLE
29
+} ACLK_AGENT_STATE;
30
+extern struct aclk_shared_state {
31
+ ACLK_METADATA_STATE metadata_submitted;
32
+ ACLK_AGENT_STATE agent_state;
33
+ time_t last_popcorn_interrupt;
34
+} aclk_shared_state;
35
+
36
typedef enum aclk_proxy_type {
37
PROXY_TYPE_UNKNOWN = 0,
38
PROXY_TYPE_SOCKS5,
aclk/aclk_query.c
new
+571
@@ -0,0 +1,571 @@
1
+#include "aclk_common.h"
2
+#include "aclk_query.h"
3
+#include "aclk_stats.h"
4
+
5
+pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
6
+pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
7
+#define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait)
8
+#define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
9
+
10
+volatile int aclk_connected = 0;
11
+
12
+#ifndef __GNUC__
13
+#pragma region ACLK_QUEUE
14
+#endif
15
+
16
+static netdata_mutex_t queue_mutex = NETDATA_MUTEX_INITIALIZER;
17
+#define ACLK_QUEUE_LOCK netdata_mutex_lock(&queue_mutex)
18
+#define ACLK_QUEUE_UNLOCK netdata_mutex_unlock(&queue_mutex)
19
+
20
+struct aclk_query {
21
+ time_t created;
22
+ time_t run_after; // Delay run until after this time
23
+ ACLK_CMD cmd; // What command is this
24
+ char *topic; // Topic to respond to
25
+ char *data; // Internal data (NULL if request from the cloud)
26
+ char *msg_id; // msg_id generated by the cloud (NULL if internal)
27
+ char *query; // The actual query
28
+ u_char deleted; // Mark deleted for garbage collect
29
+ struct aclk_query *next;
30
+};
31
+
32
+struct aclk_query_queue {
33
+ struct aclk_query *aclk_query_head;
34
+ struct aclk_query *aclk_query_tail;
35
+ unsigned int count;
36
+} aclk_queue = { .aclk_query_head = NULL, .aclk_query_tail = NULL, .count = 0 };
37
+
38
+
39
+unsigned int aclk_query_size()
40
+{
41
+ int r;
42
+ ACLK_QUEUE_LOCK;
43
+ r = aclk_queue.count;
44
+ ACLK_QUEUE_UNLOCK;
45
+ return r;
46
+}
47
+
48
+/*
49
+ * Free a query structure when done
50
+ */
51
+static void aclk_query_free(struct aclk_query *this_query)
52
+{
53
+ if (unlikely(!this_query))
54
+ return;
55
+
56
+ freez(this_query->topic);
57
+ if (likely(this_query->query))
58
+ freez(this_query->query);
59
+ if (likely(this_query->data))
60
+ freez(this_query->data);
61
+ if (likely(this_query->msg_id))
62
+ freez(this_query->msg_id);
63
+ freez(this_query);
64
+}
65
+
66
+/*
67
+ * Get the next query to process - NULL if nothing there
68
+ * The caller needs to free memory by calling aclk_query_free()
69
+ *
70
+ * topic
71
+ * query
72
+ * The structure itself
73
+ *
74
+ */
75
+static struct aclk_query *aclk_queue_pop()
76
+{
77
+ struct aclk_query *this_query;
78
+
79
+ ACLK_QUEUE_LOCK;
80
+
81
+ if (likely(!aclk_queue.aclk_query_head)) {
82
+ ACLK_QUEUE_UNLOCK;
83
+ return NULL;
84
+ }
85
+
86
+ this_query = aclk_queue.aclk_query_head;
87
+
88
+ // Get rid of the deleted entries
89
+ while (this_query && this_query->deleted) {
90
+ aclk_queue.count--;
91
+
92
+ aclk_queue.aclk_query_head = aclk_queue.aclk_query_head->next;
93
+
94
+ if (likely(!aclk_queue.aclk_query_head)) {
95
+ aclk_queue.aclk_query_tail = NULL;
96
+ }
97
+
98
+ aclk_query_free(this_query);
99
+
100
+ this_query = aclk_queue.aclk_query_head;
101
+ }
102
+
103
+ if (likely(!this_query)) {
104
+ ACLK_QUEUE_UNLOCK;
105
+ return NULL;
106
+ }
107
+
108
+ if (!this_query->deleted && this_query->run_after > now_realtime_sec()) {
109
+ info("Query %s will run in %ld seconds", this_query->query, this_query->run_after - now_realtime_sec());
110
+ ACLK_QUEUE_UNLOCK;
111
+ return NULL;
112
+ }
113
+
114
+ aclk_queue.count--;
115
+ aclk_queue.aclk_query_head = aclk_queue.aclk_query_head->next;
116
+
117
+ if (likely(!aclk_queue.aclk_query_head)) {
118
+ aclk_queue.aclk_query_tail = NULL;
119
+ }
120
+
121
+ ACLK_QUEUE_UNLOCK;
122
+ return this_query;
123
+}
124
+
125
+// Returns the entry after which we need to create a new entry to run at the specified time
126
+// If NULL is returned we need to add to HEAD
127
+// Need to have a QUERY lock before calling this
128
+
129
+static struct aclk_query *aclk_query_find_position(time_t time_to_run)
130
+{
131
+ struct aclk_query *tmp_query, *last_query;
132
+
133
+ // Quick check if we will add to the end
134
+ if (likely(aclk_queue.aclk_query_tail)) {
135
+ if (aclk_queue.aclk_query_tail->run_after <= time_to_run)
136
+ return aclk_queue.aclk_query_tail;
137
+ }
138
+
139
+ last_query = NULL;
140
+ tmp_query = aclk_queue.aclk_query_head;
141
+
142
+ while (tmp_query) {
143
+ if (tmp_query->run_after > time_to_run)
144
+ return last_query;
145
+ last_query = tmp_query;
146
+ tmp_query = tmp_query->next;
147
+ }
148
+ return last_query;
149
+}
150
+
151
+// Need to have a QUERY lock before calling this
152
+static struct aclk_query *
153
+aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
154
+{
155
+ struct aclk_query *tmp_query, *prev_query;
156
+ UNUSED(cmd);
157
+
158
+ tmp_query = aclk_queue.aclk_query_head;
159
+ prev_query = NULL;
160
+ while (tmp_query) {
161
+ if (likely(!tmp_query->deleted)) {
162
+ if (strcmp(tmp_query->topic, topic) == 0 && (!query || strcmp(tmp_query->query, query) == 0)) {
163
+ if ((!data || (data && strcmp(data, tmp_query->data) == 0)) &&
164
+ (!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0))) {
165
+ if (likely(last_query))
166
+ *last_query = prev_query;
167
+ return tmp_query;
168
+ }
169
+ }
170
+ }
171
+ prev_query = tmp_query;
172
+ tmp_query = tmp_query->next;
173
+ }
174
+ return NULL;
175
+}
176
+
177
+/*
178
+ * Add a query to execute, the result will be send to the specified topic
179
+ */
180
+
181
+int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run_after, int internal, ACLK_CMD aclk_cmd)
182
+{
183
+ struct aclk_query *new_query, *tmp_query;
184
+
185
+ // Ignore all commands while we wait for the agent to initialize
186
+ if (unlikely(!aclk_connected))
187
+ return 1;
188
+
189
+ run_after = now_realtime_sec() + run_after;
190
+
191
+ ACLK_QUEUE_LOCK;
192
+ struct aclk_query *last_query = NULL;
193
+
194
+ tmp_query = aclk_query_find(topic, data, msg_id, query, aclk_cmd, &last_query);
195
+ if (unlikely(tmp_query)) {
196
+ if (tmp_query->run_after == run_after) {
197
+ ACLK_QUEUE_UNLOCK;
198
+ QUERY_THREAD_WAKEUP;
199
+ return 0;
200
+ }
201
+
202
+ if (last_query)
203
+ last_query->next = tmp_query->next;
204
+ else
205
+ aclk_queue.aclk_query_head = tmp_query->next;
206
+
207
+ debug(D_ACLK, "Removing double entry");
208
+ aclk_query_free(tmp_query);
209
+ aclk_queue.count--;
210
+ }
211
+
212
+ if (aclk_stats_enabled) {
213
+ ACLK_STATS_LOCK;
214
+ aclk_metrics_per_sample.queries_queued++;
215
+ ACLK_STATS_UNLOCK;
216
+ }
217
+
218
+ new_query = callocz(1, sizeof(struct aclk_query));
219
+ new_query->cmd = aclk_cmd;
220
+ if (internal) {
221
+ new_query->topic = strdupz(topic);
222
+ if (likely(query))
223
+ new_query->query = strdupz(query);
224
+ } else {
225
+ new_query->topic = topic;
226
+ new_query->query = query;
227
+ new_query->msg_id = msg_id;
228
+ }
229
+
230
+ if (data)
231
+ new_query->data = strdupz(data);
232
+
233
+ new_query->next = NULL;
234
+ new_query->created = now_realtime_sec();
235
+ new_query->run_after = run_after;
236
+
237
+ debug(D_ACLK, "Added query (%s) (%s)", topic, query ? query : "");
238
+
239
+ tmp_query = aclk_query_find_position(run_after);
240
+
241
+ if (tmp_query) {
242
+ new_query->next = tmp_query->next;
243
+ tmp_query->next = new_query;
244
+ if (tmp_query == aclk_queue.aclk_query_tail)
245
+ aclk_queue.aclk_query_tail = new_query;
246
+ aclk_queue.count++;
247
+ ACLK_QUEUE_UNLOCK;
248
+ QUERY_THREAD_WAKEUP;
249
+ return 0;
250
+ }
251
+
252
+ new_query->next = aclk_queue.aclk_query_head;
253
+ aclk_queue.aclk_query_head = new_query;
254
+ aclk_queue.count++;
255
+
256
+ ACLK_QUEUE_UNLOCK;
257
+ QUERY_THREAD_WAKEUP;
258
+ return 0;
259
+}
260
+
261
+#ifndef __GNUC__
262
+#pragma endregion
263
+#endif
264
+
265
+#ifndef __GNUC__
266
+#pragma region Helper Functions
267
+#endif
268
+
269
+/*
270
+ * Take a buffer, encode it and rewrite it
271
+ *
272
+ */
273
+
274
+static char *aclk_encode_response(char *src, size_t content_size, int keep_newlines)
275
+{
276
+ char *tmp_buffer = mallocz(content_size * 2);
277
+ char *dst = tmp_buffer;
278
+ while (content_size > 0) {
279
+ switch (*src) {
280
+ case '\n':
281
+ if (keep_newlines)
282
+ {
283
+ *dst++ = '\\';
284
+ *dst++ = 'n';
285
+ }
286
+ break;
287
+ case '\t':
288
+ break;
289
+ case 0x01 ... 0x08:
290
+ case 0x0b ... 0x1F:
291
+ *dst++ = '\\';
292
+ *dst++ = 'u';
293
+ *dst++ = '0';
294
+ *dst++ = '0';
295
+ *dst++ = (*src < 0x0F) ? '0' : '1';
296
+ *dst++ = to_hex(*src);
297
+ break;
298
+ case '\"':
299
+ *dst++ = '\\';
300
+ *dst++ = *src;
301
+ break;
302
+ default:
303
+ *dst++ = *src;
304
+ }
305
+ src++;
306
+ content_size--;
307
+ }
308
+ *dst = '\0';
309
+
310
+ return tmp_buffer;
311
+}
312
+
313
+#ifndef __GNUC__
314
+#pragma endregion
315
+#endif
316
+
317
+#ifndef __GNUC__
318
+#pragma region ACLK_QUERY
319
+#endif
320
+
321
+static int aclk_execute_query(struct aclk_query *this_query)
322
+{
323
+ if (strncmp(this_query->query, "/api/v1/", 8) == 0) {
324
+ struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client));
325
+ w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
326
+ w->response.header = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
327
+ w->response.header_output = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
328
+ strcpy(w->origin, "*"); // Simulate web_client_create_on_fd()
329
+ w->cookie1[0] = 0; // Simulate web_client_create_on_fd()
330
+ w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
331
+ w->acl = 0x1f;
332
+
333
+ char *mysep = strchr(this_query->query, '?');
334
+ if (mysep) {
335
+ strncpyz(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE);
336
+ *mysep = '\0';
337
+ } else
338
+ strncpyz(w->decoded_query_string, this_query->query, NETDATA_WEB_REQUEST_URL_SIZE);
339
+
340
+ mysep = strrchr(this_query->query, '/');
341
+
342
+ // TODO: handle bad response perhaps in a different way. For now it does to the payload
343
+ w->response.code = web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
344
+ now_realtime_timeval(&w->tv_ready);
345
+ w->response.data->date = w->tv_ready.tv_sec;
346
+ web_client_build_http_header(w); // TODO: this function should offset from date, not tv_ready
347
+ BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
348
+ buffer_flush(local_buffer);
349
+ local_buffer->contenttype = CT_APPLICATION_JSON;
350
+
351
+ aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0);
352
+ buffer_strcat(local_buffer, ",\n\t\"payload\": ");
353
+ char *encoded_response = aclk_encode_response(w->response.data->buffer, w->response.data->len, 0);
354
+ char *encoded_header = aclk_encode_response(w->response.header_output->buffer, w->response.header_output->len, 1);
355
+
356
+ buffer_sprintf(
357
+ local_buffer, "{\n\"code\": %d,\n\"body\": \"%s\",\n\"headers\": \"%s\"\n}",
358
+ w->response.code, encoded_response, encoded_header);
359
+
360
+ buffer_sprintf(local_buffer, "\n}");
361
+
362
+ debug(D_ACLK, "Response:%s", encoded_header);
363
+
364
+ aclk_send_message(this_query->topic, local_buffer->buffer, this_query->msg_id);
365
+
366
+ buffer_free(w->response.data);
367
+ buffer_free(w->response.header);
368
+ buffer_free(w->response.header_output);
369
+ freez(w);
370
+ buffer_free(local_buffer);
371
+ freez(encoded_response);
372
+ freez(encoded_header);
373
+ return 0;
374
+ }
375
+ return 1;
376
+}
377
+
378
+/*
379
+ * This function will fetch the next pending command and process it
380
+ *
381
+ */
382
+static int aclk_process_query(int t_idx)
383
+{
384
+ struct aclk_query *this_query;
385
+ static long int query_count = 0;
386
+ ACLK_METADATA_STATE meta_state;
387
+ usec_t t = 0;
388
+
389
+ if (!aclk_connected)
390
+ return 0;
391
+
392
+ this_query = aclk_queue_pop();
393
+ if (likely(!this_query)) {
394
+ return 0;
395
+ }
396
+
397
+ if (unlikely(this_query->deleted)) {
398
+ debug(D_ACLK, "Garbage collect query %s:%s", this_query->topic, this_query->query);
399
+ aclk_query_free(this_query);
400
+ return 1;
401
+ }
402
+ query_count++;
403
+
404
+ debug(
405
+ D_ACLK, "Query #%ld (%s) size=%zu in queue %d seconds", query_count, this_query->topic,
406
+ this_query->query ? strlen(this_query->query) : 0, (int)(now_realtime_sec() - this_query->created));
407
+
408
+ switch (this_query->cmd) {
409
+ case ACLK_CMD_ONCONNECT:
410
+ debug(D_ACLK, "EXECUTING on connect metadata command");
411
+ ACLK_SHARED_STATE_LOCK;
412
+ meta_state = aclk_shared_state.metadata_submitted;
413
+ aclk_shared_state.metadata_submitted = ACLK_METADATA_SENT;
414
+ ACLK_SHARED_STATE_UNLOCK;
415
+ aclk_send_metadata(meta_state);
416
+ break;
417
+
418
+ case ACLK_CMD_CHART:
419
+ debug(D_ACLK, "EXECUTING a chart update command");
420
+ aclk_send_single_chart(this_query->data, this_query->query);
421
+ break;
422
+
423
+ case ACLK_CMD_CHARTDEL:
424
+ debug(D_ACLK, "EXECUTING a chart delete command");
425
+ //TODO: This send the info metadata for now
426
+ aclk_send_info_metadata(ACLK_METADATA_SENT);
427
+ break;
428
+
429
+ case ACLK_CMD_ALARM:
430
+ debug(D_ACLK, "EXECUTING an alarm update command");
431
+ aclk_send_message(this_query->topic, this_query->query, this_query->msg_id);
432
+ break;
433
+
434
+ case ACLK_CMD_CLOUD:
435
+ t = now_monotonic_high_precision_usec();
436
+ debug(D_ACLK, "EXECUTING a cloud command");
437
+ aclk_execute_query(this_query);
438
+ t = now_monotonic_high_precision_usec() - t;
439
+ break;
440
+
441
+ default:
442
+ break;
443
+ }
444
+ debug(D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
445
+
446
+ if (aclk_stats_enabled) {
447
+ ACLK_STATS_LOCK;
448
+ aclk_metrics_per_sample.queries_dispatched++;
449
+ aclk_queries_per_thread[t_idx]++;
450
+ if(this_query->cmd == ACLK_CMD_CLOUD) {
451
+ aclk_metrics_per_sample.cloud_q_process_total += t;
452
+ aclk_metrics_per_sample.cloud_q_process_count++;
453
+ if(aclk_metrics_per_sample.cloud_q_process_max < t)
454
+ aclk_metrics_per_sample.cloud_q_process_max = t;
455
+ }
456
+ ACLK_STATS_UNLOCK;
457
+ }
458
+
459
+ aclk_query_free(this_query);
460
+
461
+ return 1;
462
+}
463
+
464
+void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads)
465
+{
466
+ if (query_threads && query_threads->thread_list) {
467
+ for (int i = 0; i < query_threads->count; i++) {
468
+ netdata_thread_join(query_threads->thread_list[i].thread, NULL);
469
+ }
470
+ freez(query_threads->thread_list);
471
+ }
472
+
473
+ struct aclk_query *this_query;
474
+
475
+ do {
476
+ this_query = aclk_queue_pop();
477
+ aclk_query_free(this_query);
478
+ } while (this_query);
479
+}
480
+
481
+#define TASK_LEN_MAX 16
482
+void aclk_query_threads_start(struct aclk_query_threads *query_threads)
483
+{
484
+ info("Starting %d query threads.", query_threads->count);
485
+
486
+ char thread_name[TASK_LEN_MAX];
487
+ query_threads->thread_list = callocz(query_threads->count, sizeof(struct aclk_query_thread));
488
+ for (int i = 0; i < query_threads->count; i++) {
489
+ query_threads->thread_list[i].idx = i; //thread needs to know its index for statistics
490
+
491
+ snprintf(thread_name, TASK_LEN_MAX, "%s_%d", ACLK_THREAD_NAME, i);
492
+ netdata_thread_create(
493
+ &query_threads->thread_list[i].thread, thread_name, NETDATA_THREAD_OPTION_JOINABLE, aclk_query_main_thread,
494
+ &query_threads->thread_list[i]);
495
+ }
496
+}
497
+
498
+/**
499
+ * Main query processing thread
500
+ *
501
+ * On startup wait for the agent collectors to initialize
502
+ * Expect at least a time of ACLK_STABLE_TIMEOUT seconds
503
+ * of no new collectors coming in in order to mark the agent
504
+ * as stable (set agent_state = AGENT_STABLE)
505
+ */
506
+void *aclk_query_main_thread(void *ptr)
507
+{
508
+ struct aclk_query_thread *info = ptr;
509
+ time_t previous_popcorn_interrupt = 0;
510
+
511
+ while (!netdata_exit) {
512
+ ACLK_SHARED_STATE_LOCK;
513
+ if (aclk_shared_state.agent_state != AGENT_INITIALIZING) {
514
+ ACLK_SHARED_STATE_UNLOCK;
515
+ break;
516
+ }
517
+
518
+ time_t checkpoint = now_realtime_sec() - aclk_shared_state.last_popcorn_interrupt;
519
+
520
+ if (checkpoint > ACLK_STABLE_TIMEOUT) {
521
+ aclk_shared_state.agent_state = AGENT_STABLE;
522
+ ACLK_SHARED_STATE_UNLOCK;
523
+ info("AGENT stable, last collector initialization activity was %ld seconds ago", checkpoint);
524
+#ifdef ACLK_DEBUG
525
+ _dump_collector_list();
526
+#endif
527
+ break;
528
+ }
529
+
530
+ if (previous_popcorn_interrupt != aclk_shared_state.last_popcorn_interrupt) {
531
+ info("Waiting %ds from this moment for agent collectors to initialize." , ACLK_STABLE_TIMEOUT);
532
+ previous_popcorn_interrupt = aclk_shared_state.last_popcorn_interrupt;
533
+ }
534
+ ACLK_SHARED_STATE_UNLOCK;
535
+ sleep_usec(USEC_PER_SEC * 1);
536
+ }
537
+
538
+ while (!netdata_exit) {
539
+ ACLK_SHARED_STATE_LOCK;
540
+ if (unlikely(!aclk_shared_state.metadata_submitted)) {
541
+ ACLK_SHARED_STATE_UNLOCK;
542
+ if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
543
+ errno = 0;
544
+ error("ACLK failed to queue on_connect command");
545
+ sleep(1);
546
+ continue;
547
+ }
548
+ ACLK_SHARED_STATE_LOCK;
549
+ aclk_shared_state.metadata_submitted = ACLK_METADATA_CMD_QUEUED;
550
+ }
551
+ ACLK_SHARED_STATE_UNLOCK;
552
+
553
+ while (aclk_process_query(info->idx)) {
554
+ // Process all commands
555
+ };
556
+
557
+ QUERY_THREAD_LOCK;
558
+
559
+ // TODO: Need to check if there are queries awaiting already
560
+ if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))
561
+ sleep_usec(USEC_PER_SEC * 1);
562
+
563
+ QUERY_THREAD_UNLOCK;
564
+ }
565
+
566
+ return NULL;
567
+}
568
+
569
+#ifndef __GNUC__
570
+#pragma endregion
571
+#endif
aclk/aclk_query.h
new
+34
@@ -0,0 +1,34 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_ACLK_QUERY_H
4
+#define NETDATA_ACLK_QUERY_H
5
+
6
+#include "libnetdata/libnetdata.h"
7
+
8
+#define ACLK_STABLE_TIMEOUT 3 // Minimum delay to mark AGENT as stable
9
+
10
+extern pthread_cond_t query_cond_wait;
11
+extern pthread_mutex_t query_lock_wait;
12
+#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
13
+#define QUERY_THREAD_WAKEUP_ALL pthread_cond_broadcast(&query_cond_wait)
14
+
15
+extern volatile int aclk_connected;
16
+
17
+struct aclk_query_thread {
18
+ netdata_thread_t thread;
19
+ int idx;
20
+};
21
+
22
+struct aclk_query_threads {
23
+ struct aclk_query_thread *thread_list;
24
+ int count;
25
+};
26
+
27
+void *aclk_query_main_thread(void *ptr);
28
+int aclk_queue_query(char *token, char *data, char *msg_type, char *query, int run_after, int internal, ACLK_CMD cmd);
29
+
30
+void aclk_query_threads_start(struct aclk_query_threads *query_threads);
31
+void aclk_query_threads_cleanup(struct aclk_query_threads *query_threads);
32
+unsigned int aclk_query_size();
33
+
34
+#endif //NETDATA_AGENT_CLOUD_LINK_H
aclk/aclk_stats.c
+97
-11
@@ -4,6 +4,16 @@ netdata_mutex_t aclk_stats_mutex = NETDATA_MUTEX_INITIALIZER;
4
5
int aclk_stats_enabled;
6
7
+int query_thread_count;
8
+
9
+// data ACLK stats need per query thread
10
+struct aclk_qt_data {
11
+ RRDDIM *dim;
12
+} *aclk_qt_data = NULL;
13
+
14
+uint32_t *aclk_queries_per_thread = NULL;
15
+uint32_t *aclk_queries_per_thread_sample = NULL;
16
+
17
struct aclk_metrics aclk_metrics = {
18
.online = 0,
19
};
@@ -17,7 +27,7 @@ static void aclk_stats_collect(struct aclk_metrics_per_sample *per_sample, struc
27
28
if (unlikely(!st_aclkstats)) {
29
st_aclkstats = rrdset_create_localhost(
20
- "netdata", "aclk_status", NULL, "aclk_stats", NULL, "ACLK/Cloud connection status",
30
+ "netdata", "aclk_status", NULL, "aclk", NULL, "ACLK/Cloud connection status",
31
"connected", "netdata", "stats", 200000, localhost->rrd_update_every, RRDSET_TYPE_LINE);
32
33
rd_online_status = rrddim_add(st_aclkstats, "online", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
@@ -29,7 +39,7 @@ static void aclk_stats_collect(struct aclk_metrics_per_sample *per_sample, struc
39
rrdset_done(st_aclkstats);
40
}
41
32
-static void aclk_stats_query_thread(struct aclk_metrics_per_sample *per_sample)
42
+static void aclk_stats_query_queue(struct aclk_metrics_per_sample *per_sample)
43
{
44
static RRDSET *st_query_thread = NULL;
45
static RRDDIM *rd_queued = NULL;
@@ -37,16 +47,16 @@ static void aclk_stats_query_thread(struct aclk_metrics_per_sample *per_sample)
47
48
if (unlikely(!st_query_thread)) {
49
st_query_thread = rrdset_create_localhost(
40
- "netdata", "aclk_query_per_second", NULL, "aclk_stats", NULL, "ACLK Queries per second", "queries/s",
50
+ "netdata", "aclk_query_per_second", NULL, "aclk", NULL, "ACLK Queries per second", "queries/s",
51
"netdata", "stats", 200001, localhost->rrd_update_every, RRDSET_TYPE_AREA);
52
53
rd_queued = rrddim_add(st_query_thread, "added", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
44
- rd_dispatched = rrddim_add(st_query_thread, "dispatched", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
54
+ rd_dispatched = rrddim_add(st_query_thread, "dispatched", NULL, -1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
55
} else
56
rrdset_next(st_query_thread);
57
58
rrddim_set_by_pointer(st_query_thread, rd_queued, per_sample->queries_queued);
49
- rrddim_set_by_pointer(st_query_thread, rd_dispatched, -per_sample->queries_dispatched);
59
+ rrddim_set_by_pointer(st_query_thread, rd_dispatched, per_sample->queries_dispatched);
60
61
rrdset_done(st_query_thread);
62
}
@@ -60,7 +70,7 @@ static void aclk_stats_latency(struct aclk_metrics_per_sample *per_sample)
70
71
if (unlikely(!st)) {
72
st = rrdset_create_localhost(
63
- "netdata", "aclk_latency_mqtt", NULL, "aclk_stats", NULL, "ACLK Message Publish Latency", "ms",
73
+ "netdata", "aclk_latency_mqtt", NULL, "aclk", NULL, "ACLK Message Publish Latency", "ms",
74
"netdata", "stats", 200002, localhost->rrd_update_every, RRDSET_TYPE_LINE);
75
76
rd_avg = rrddim_add(st, "avg", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
@@ -86,7 +96,7 @@ static void aclk_stats_write_q(struct aclk_metrics_per_sample *per_sample)
96
97
if (unlikely(!st)) {
98
st = rrdset_create_localhost(
89
- "netdata", "aclk_write_q", NULL, "aclk_stats", NULL, "Write Queue Mosq->Libwebsockets", "kB/s",
99
+ "netdata", "aclk_write_q", NULL, "aclk", NULL, "Write Queue Mosq->Libwebsockets", "kB/s",
100
"netdata", "stats", 200003, localhost->rrd_update_every, RRDSET_TYPE_AREA);
101
102
rd_wq_add = rrddim_add(st, "added", NULL, 1, 1024 * localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
@@ -108,7 +118,7 @@ static void aclk_stats_read_q(struct aclk_metrics_per_sample *per_sample)
118
119
if (unlikely(!st)) {
120
st = rrdset_create_localhost(
111
- "netdata", "aclk_read_q", NULL, "aclk_stats", NULL, "Read Queue Libwebsockets->Mosq", "kB/s",
121
+ "netdata", "aclk_read_q", NULL, "aclk", NULL, "Read Queue Libwebsockets->Mosq", "kB/s",
122
"netdata", "stats", 200004, localhost->rrd_update_every, RRDSET_TYPE_AREA);
123
124
rd_rq_add = rrddim_add(st, "added", NULL, 1, 1024 * localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
@@ -130,7 +140,7 @@ static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample)
140
141
if (unlikely(!st)) {
142
st = rrdset_create_localhost(
133
- "netdata", "aclk_cloud_req", NULL, "aclk_stats", NULL, "Requests received from cloud", "req/s",
143
+ "netdata", "aclk_cloud_req", NULL, "aclk", NULL, "Requests received from cloud", "req/s",
144
"netdata", "stats", 200005, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
145
146
rd_rq_rcvd = rrddim_add(st, "received", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
@@ -144,13 +154,82 @@ static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample)
154
rrdset_done(st);
155
}
156
157
+#define MAX_DIM_NAME 16
158
+static void aclk_stats_query_threads(uint32_t *queries_per_thread)
159
+{
160
+ static RRDSET *st = NULL;
161
+
162
+ char dim_name[MAX_DIM_NAME];
163
+
164
+ if (unlikely(!st)) {
165
+ st = rrdset_create_localhost(
166
+ "netdata", "aclk_query_threads", NULL, "aclk", NULL, "Queries Processed Per Thread", "req/s",
167
+ "netdata", "stats", 200007, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
168
+
169
+ for (int i = 0; i < query_thread_count; i++) {
170
+ snprintf(dim_name, MAX_DIM_NAME, "Query %d", i);
171
+ aclk_qt_data[i].dim = rrddim_add(st, dim_name, NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
172
+ }
173
+ } else
174
+ rrdset_next(st);
175
+
176
+ for (int i = 0; i < query_thread_count; i++) {
177
+ rrddim_set_by_pointer(st, aclk_qt_data[i].dim, queries_per_thread[i]);
178
+ }
179
+
180
+ rrdset_done(st);
181
+}
182
+
183
+static void aclk_stats_query_time(struct aclk_metrics_per_sample *per_sample)
184
+{
185
+ static RRDSET *st = NULL;
186
+ static RRDDIM *rd_rq_avg = NULL;
187
+ static RRDDIM *rd_rq_max = NULL;
188
+ static RRDDIM *rd_rq_total = NULL;
189
+
190
+ if (unlikely(!st)) {
191
+ st = rrdset_create_localhost(
192
+ "netdata", "aclk_query_time", NULL, "aclk", NULL, "Time it took to process cloud requested DB queries", "us",
193
+ "netdata", "stats", 200006, localhost->rrd_update_every, RRDSET_TYPE_LINE);
194
+
195
+ rd_rq_avg = rrddim_add(st, "avg", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
196
+ rd_rq_max = rrddim_add(st, "max", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
197
+ rd_rq_total = rrddim_add(st, "total", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
198
+ } else
199
+ rrdset_next(st);
200
+
201
+ if(per_sample->cloud_q_process_count)
202
+ rrddim_set_by_pointer(st, rd_rq_avg, roundf((float)per_sample->cloud_q_process_total / per_sample->cloud_q_process_count));
203
+ else
204
+ rrddim_set_by_pointer(st, rd_rq_avg, 0);
205
+ rrddim_set_by_pointer(st, rd_rq_max, per_sample->cloud_q_process_max);
206
+ rrddim_set_by_pointer(st, rd_rq_total, per_sample->cloud_q_process_total);
207
+
208
+ rrdset_done(st);
209
+}
210
+
211
+void aclk_stats_thread_cleanup()
212
+{
213
+ freez(aclk_qt_data);
214
+ freez(aclk_queries_per_thread);
215
+ freez(aclk_queries_per_thread_sample);
216
+}
217
+
218
void *aclk_stats_main_thread(void *ptr)
219
{
149
- UNUSED(ptr);
220
+ struct aclk_stats_thread *args = ptr;
221
+
222
+ query_thread_count = args->query_thread_count;
223
+ aclk_qt_data = callocz(query_thread_count, sizeof(struct aclk_qt_data));
224
+ aclk_queries_per_thread = callocz(query_thread_count, sizeof(uint32_t));
225
+ aclk_queries_per_thread_sample = callocz(query_thread_count, sizeof(uint32_t));
226
+
227
heartbeat_t hb;
228
heartbeat_init(&hb);
229
usec_t step_ut = localhost->rrd_update_every * USEC_PER_SEC;
230
+
231
memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
232
+
233
struct aclk_metrics_per_sample per_sample;
234
struct aclk_metrics permanent;
235
@@ -168,10 +247,13 @@ void *aclk_stats_main_thread(void *ptr)
247
memcpy(&per_sample, &aclk_metrics_per_sample, sizeof(struct aclk_metrics_per_sample));
248
memcpy(&permanent, &aclk_metrics, sizeof(struct aclk_metrics));
249
memset(&aclk_metrics_per_sample, 0, sizeof(struct aclk_metrics_per_sample));
250
+
251
+ mempcpy(aclk_queries_per_thread_sample, aclk_queries_per_thread, sizeof(uint32_t) * query_thread_count);
252
+ memset(aclk_queries_per_thread, 0, sizeof(uint32_t) * query_thread_count);
253
ACLK_STATS_UNLOCK;
254
255
aclk_stats_collect(&per_sample, &permanent);
174
- aclk_stats_query_thread(&per_sample);
256
+ aclk_stats_query_queue(&per_sample);
257
#ifdef NETDATA_INTERNAL_CHECKS
258
aclk_stats_latency(&per_sample);
259
#endif
@@ -179,7 +261,11 @@ void *aclk_stats_main_thread(void *ptr)
261
aclk_stats_read_q(&per_sample);
262
263
aclk_stats_cloud_req(&per_sample);
264
+ aclk_stats_query_threads(aclk_queries_per_thread_sample);
265
+
266
+ aclk_stats_query_time(&per_sample);
267
}
268
+
269
return 0;
270
}
271
aclk/aclk_stats.h
+17
-2
@@ -5,6 +5,7 @@
5
6
#include "../daemon/common.h"
7
#include "libnetdata/libnetdata.h"
8
+#include "aclk_common.h"
9
10
#define ACLK_STATS_THREAD_NAME "ACLK_Stats"
11
@@ -15,6 +16,11 @@ extern netdata_mutex_t aclk_stats_mutex;
16
17
extern int aclk_stats_enabled;
18
19
+struct aclk_stats_thread {
20
+ netdata_thread_t *thread;
21
+ int query_thread_count;
22
+};
23
+
24
// preserve between samples
25
struct aclk_metrics {
26
volatile uint8_t online;
@@ -28,13 +34,15 @@ extern struct aclk_metrics_per_sample {
34
despite it being then seemingly longer in graph */
35
volatile uint8_t offline_during_sample;
36
31
- volatile uint8_t queries_queued;
32
- volatile uint8_t queries_dispatched;
37
+ volatile uint32_t queries_queued;
38
+ volatile uint32_t queries_dispatched;
39
+
40
#ifdef NETDATA_INTERNAL_CHECKS
41
volatile uint32_t latency_max;
42
volatile uint32_t latency_total;
43
volatile uint32_t latency_count;
44
#endif
45
+
46
volatile uint32_t write_q_added;
47
volatile uint32_t write_q_consumed;
48
@@ -43,9 +51,16 @@ extern struct aclk_metrics_per_sample {
51
52
volatile uint32_t cloud_req_recvd;
53
volatile uint32_t cloud_req_err;
54
+
55
+ volatile uint32_t cloud_q_process_total;
56
+ volatile uint32_t cloud_q_process_count;
57
+ volatile uint32_t cloud_q_process_max;
58
} aclk_metrics_per_sample;
59
60
+extern uint32_t *aclk_queries_per_thread;
61
+
62
void *aclk_stats_main_thread(void *ptr);
63
+void aclk_stats_thread_cleanup();
64
void aclk_stats_upd_online(int online);
65
66
#endif /* NETDATA_ACLK_STATS_H */
aclk/agent_cloud_link.c
+119
-620
@@ -3,34 +3,26 @@
3
#include "libnetdata/libnetdata.h"
4
#include "agent_cloud_link.h"
5
#include "aclk_lws_https_client.h"
6
+#include "aclk_query.h"
7
#include "aclk_common.h"
8
#include "aclk_stats.h"
9
10
int aclk_shutting_down = 0;
10
-// State-machine for the on-connect metadata transmission.
11
-// TODO: The AGENT_STATE should be centralized as it would be useful to control error-logging during the initial
12
-// agent startup phase.
13
-static ACLK_METADATA_STATE aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
14
-static AGENT_STATE agent_state = AGENT_INITIALIZING;
11
12
// Other global state
13
static int aclk_subscribed = 0;
14
static int aclk_disable_single_updates = 0;
19
-static time_t last_init_sequence = 0;
20
-static int waiting_init = 1;
15
static char *aclk_username = NULL;
16
static char *aclk_password = NULL;
17
18
static char *global_base_topic = NULL;
19
static int aclk_connecting = 0;
26
-int aclk_connected = 0; // Exposed in the web-api
20
int aclk_force_reconnect = 0; // Indication from lower layers
21
int aclk_kill_link = 0; // Tell the agent to tear down the link
22
usec_t aclk_session_us = 0; // Used by the mqtt layer
23
time_t aclk_session_sec = 0; // Used by the mqtt layer
24
25
static netdata_mutex_t aclk_mutex = NETDATA_MUTEX_INITIALIZER;
33
-static netdata_mutex_t query_mutex = NETDATA_MUTEX_INITIALIZER;
26
static netdata_mutex_t collector_mutex = NETDATA_MUTEX_INITIALIZER;
27
28
#define ACLK_LOCK netdata_mutex_lock(&aclk_mutex)
@@ -39,16 +31,6 @@ static netdata_mutex_t collector_mutex = NETDATA_MUTEX_INITIALIZER;
31
#define COLLECTOR_LOCK netdata_mutex_lock(&collector_mutex)
32
#define COLLECTOR_UNLOCK netdata_mutex_unlock(&collector_mutex)
33
42
-#define QUERY_LOCK netdata_mutex_lock(&query_mutex)
43
-#define QUERY_UNLOCK netdata_mutex_unlock(&query_mutex)
44
-
45
-pthread_cond_t query_cond_wait = PTHREAD_COND_INITIALIZER;
46
-pthread_mutex_t query_lock_wait = PTHREAD_MUTEX_INITIALIZER;
47
-
48
-#define QUERY_THREAD_LOCK pthread_mutex_lock(&query_lock_wait);
49
-#define QUERY_THREAD_UNLOCK pthread_mutex_unlock(&query_lock_wait)
50
-#define QUERY_THREAD_WAKEUP pthread_cond_signal(&query_cond_wait)
51
-
34
void lws_wss_check_queues(size_t *write_len, size_t *write_len_bytes, size_t *read_len);
35
void aclk_lws_wss_destroy_context();
36
/*
@@ -71,24 +53,6 @@ struct _collector {
53
54
struct _collector *collector_list = NULL;
55
74
-struct aclk_query {
75
- time_t created;
76
- time_t run_after; // Delay run until after this time
77
- ACLK_CMD cmd; // What command is this
78
- char *topic; // Topic to respond to
79
- char *data; // Internal data (NULL if request from the cloud)
80
- char *msg_id; // msg_id generated by the cloud (NULL if internal)
81
- char *query; // The actual query
82
- u_char deleted; // Mark deleted for garbage collect
83
- struct aclk_query *next;
84
-};
85
-
86
-struct aclk_query_queue {
87
- struct aclk_query *aclk_query_head;
88
- struct aclk_query *aclk_query_tail;
89
- uint64_t count;
90
-} aclk_queue = { .aclk_query_head = NULL, .aclk_query_tail = NULL, .count = 0 };
91
-
56
char *create_uuid()
57
{
58
uuid_t uuid;
@@ -219,225 +183,6 @@ unsigned long int aclk_reconnect_delay(int mode)
183
return delay;
184
}
185
222
-/*
223
- * Free a query structure when done
224
- */
225
-
226
-void aclk_query_free(struct aclk_query *this_query)
227
-{
228
- if (unlikely(!this_query))
229
- return;
230
-
231
- freez(this_query->topic);
232
- if (likely(this_query->query))
233
- freez(this_query->query);
234
- if (likely(this_query->data))
235
- freez(this_query->data);
236
- if (likely(this_query->msg_id))
237
- freez(this_query->msg_id);
238
- freez(this_query);
239
-}
240
-
241
-// Returns the entry after which we need to create a new entry to run at the specified time
242
-// If NULL is returned we need to add to HEAD
243
-// Need to have a QUERY lock before calling this
244
-
245
-struct aclk_query *aclk_query_find_position(time_t time_to_run)
246
-{
247
- struct aclk_query *tmp_query, *last_query;
248
-
249
- // Quick check if we will add to the end
250
- if (likely(aclk_queue.aclk_query_tail)) {
251
- if (aclk_queue.aclk_query_tail->run_after <= time_to_run)
252
- return aclk_queue.aclk_query_tail;
253
- }
254
-
255
- last_query = NULL;
256
- tmp_query = aclk_queue.aclk_query_head;
257
-
258
- while (tmp_query) {
259
- if (tmp_query->run_after > time_to_run)
260
- return last_query;
261
- last_query = tmp_query;
262
- tmp_query = tmp_query->next;
263
- }
264
- return last_query;
265
-}
266
-
267
-// Need to have a QUERY lock before calling this
268
-struct aclk_query *
269
-aclk_query_find(char *topic, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query)
270
-{
271
- struct aclk_query *tmp_query, *prev_query;
272
- UNUSED(cmd);
273
-
274
- tmp_query = aclk_queue.aclk_query_head;
275
- prev_query = NULL;
276
- while (tmp_query) {
277
- if (likely(!tmp_query->deleted)) {
278
- if (strcmp(tmp_query->topic, topic) == 0 && (!query || strcmp(tmp_query->query, query) == 0)) {
279
- if ((!data || (data && strcmp(data, tmp_query->data) == 0)) &&
280
- (!msg_id || (msg_id && strcmp(msg_id, tmp_query->msg_id) == 0))) {
281
- if (likely(last_query))
282
- *last_query = prev_query;
283
- return tmp_query;
284
- }
285
- }
286
- }
287
- prev_query = tmp_query;
288
- tmp_query = tmp_query->next;
289
- }
290
- return NULL;
291
-}
292
-
293
-/*
294
- * Add a query to execute, the result will be send to the specified topic
295
- */
296
-
297
-int aclk_queue_query(char *topic, char *data, char *msg_id, char *query, int run_after, int internal, ACLK_CMD aclk_cmd)
298
-{
299
- struct aclk_query *new_query, *tmp_query;
300
-
301
- // Ignore all commands while we wait for the agent to initialize
302
- if (unlikely(waiting_init))
303
- return 1;
304
-
305
- run_after = now_realtime_sec() + run_after;
306
-
307
- QUERY_LOCK;
308
- struct aclk_query *last_query = NULL;
309
-
310
- tmp_query = aclk_query_find(topic, data, msg_id, query, aclk_cmd, &last_query);
311
- if (unlikely(tmp_query)) {
312
- if (tmp_query->run_after == run_after) {
313
- QUERY_UNLOCK;
314
- QUERY_THREAD_WAKEUP;
315
- return 0;
316
- }
317
-
318
- if (last_query)
319
- last_query->next = tmp_query->next;
320
- else
321
- aclk_queue.aclk_query_head = tmp_query->next;
322
-
323
- debug(D_ACLK, "Removing double entry");
324
- aclk_query_free(tmp_query);
325
- aclk_queue.count--;
326
- }
327
-
328
- if (aclk_stats_enabled) {
329
- ACLK_STATS_LOCK;
330
- aclk_metrics_per_sample.queries_queued++;
331
- ACLK_STATS_UNLOCK;
332
- }
333
-
334
- new_query = callocz(1, sizeof(struct aclk_query));
335
- new_query->cmd = aclk_cmd;
336
- if (internal) {
337
- new_query->topic = strdupz(topic);
338
- if (likely(query))
339
- new_query->query = strdupz(query);
340
- } else {
341
- new_query->topic = topic;
342
- new_query->query = query;
343
- new_query->msg_id = msg_id;
344
- }
345
-
346
- if (data)
347
- new_query->data = strdupz(data);
348
-
349
- new_query->next = NULL;
350
- new_query->created = now_realtime_sec();
351
- new_query->run_after = run_after;
352
-
353
- debug(D_ACLK, "Added query (%s) (%s)", topic, query ? query : "");
354
-
355
- tmp_query = aclk_query_find_position(run_after);
356
-
357
- if (tmp_query) {
358
- new_query->next = tmp_query->next;
359
- tmp_query->next = new_query;
360
- if (tmp_query == aclk_queue.aclk_query_tail)
361
- aclk_queue.aclk_query_tail = new_query;
362
- aclk_queue.count++;
363
- QUERY_UNLOCK;
364
- QUERY_THREAD_WAKEUP;
365
- return 0;
366
- }
367
-
368
- new_query->next = aclk_queue.aclk_query_head;
369
- aclk_queue.aclk_query_head = new_query;
370
- aclk_queue.count++;
371
-
372
- QUERY_UNLOCK;
373
- QUERY_THREAD_WAKEUP;
374
- return 0;
375
-}
376
-
377
-inline int aclk_submit_request(struct aclk_request *request)
378
-{
379
- return aclk_queue_query(request->callback_topic, NULL, request->msg_id, request->payload, 0, 0, ACLK_CMD_CLOUD);
380
-}
381
-
382
-/*
383
- * Get the next query to process - NULL if nothing there
384
- * The caller needs to free memory by calling aclk_query_free()
385
- *
386
- * topic
387
- * query
388
- * The structure itself
389
- *
390
- */
391
-struct aclk_query *aclk_queue_pop()
392
-{
393
- struct aclk_query *this_query;
394
-
395
- QUERY_LOCK;
396
-
397
- if (likely(!aclk_queue.aclk_query_head)) {
398
- QUERY_UNLOCK;
399
- return NULL;
400
- }
401
-
402
- this_query = aclk_queue.aclk_query_head;
403
-
404
- // Get rid of the deleted entries
405
- while (this_query && this_query->deleted) {
406
- aclk_queue.count--;
407
-
408
- aclk_queue.aclk_query_head = aclk_queue.aclk_query_head->next;
409
-
410
- if (likely(!aclk_queue.aclk_query_head)) {
411
- aclk_queue.aclk_query_tail = NULL;
412
- }
413
-
414
- aclk_query_free(this_query);
415
-
416
- this_query = aclk_queue.aclk_query_head;
417
- }
418
-
419
- if (likely(!this_query)) {
420
- QUERY_UNLOCK;
421
- return NULL;
422
- }
423
-
424
- if (!this_query->deleted && this_query->run_after > now_realtime_sec()) {
425
- info("Query %s will run in %ld seconds", this_query->query, this_query->run_after - now_realtime_sec());
426
- QUERY_UNLOCK;
427
- return NULL;
428
- }
429
-
430
- aclk_queue.count--;
431
- aclk_queue.aclk_query_head = aclk_queue.aclk_query_head->next;
432
-
433
- if (likely(!aclk_queue.aclk_query_head)) {
434
- aclk_queue.aclk_query_tail = NULL;
435
- }
436
-
437
- QUERY_UNLOCK;
438
- return this_query;
439
-}
440
-
186
// This will give the base topic that the agent will publish messages.
187
// subtopics will be sent under the base topic e.g. base_topic/subtopic
188
// This is called during the connection, we delete any previous topic
@@ -489,6 +234,10 @@ char *get_topic(char *sub_topic, char *final_topic, int max_size)
234
return final_topic;
235
}
236
237
+#ifndef __GNUC__
238
+#pragma region ACLK Internal Collector Tracking
239
+#endif
240
+
241
/*
242
* Free a collector structure
243
*/
@@ -673,6 +422,22 @@ static struct _collector *_add_collector(const char *hostname, const char *plugi
422
return tmp_collector;
423
}
424
425
+#ifndef __GNUC__
426
+#pragma endregion
427
+#endif
428
+
429
+inline static int aclk_popcorn_check_bump()
430
+{
431
+ ACLK_SHARED_STATE_LOCK;
432
+ if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
433
+ aclk_shared_state.last_popcorn_interrupt = now_realtime_sec();
434
+ ACLK_SHARED_STATE_UNLOCK;
435
+ return 1;
436
+ }
437
+ ACLK_SHARED_STATE_UNLOCK;
438
+ return 0;
439
+}
440
+
441
/*
442
* Add a new collector to the list
443
* If it exists, update the chart count
@@ -693,14 +458,13 @@ void aclk_add_collector(const char *hostname, const char *plugin_name, const cha
458
return;
459
}
460
696
- if (unlikely(agent_state == AGENT_INITIALIZING))
697
- last_init_sequence = now_realtime_sec();
698
- else {
699
- if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
700
- debug(D_ACLK, "ACLK failed to queue on_connect command on collector addition");
701
- }
702
-
461
COLLECTOR_UNLOCK;
462
+
463
+ if(aclk_popcorn_check_bump())
464
+ return;
465
+
466
+ if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
467
+ debug(D_ACLK, "ACLK failed to queue on_connect command on collector addition");
468
}
469
470
/*
@@ -733,286 +497,13 @@ void aclk_del_collector(const char *hostname, const char *plugin_name, const cha
497
498
COLLECTOR_UNLOCK;
499
736
- if (unlikely(agent_state == AGENT_INITIALIZING))
737
- last_init_sequence = now_realtime_sec();
738
- else {
739
- if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
740
- debug(D_ACLK, "ACLK failed to queue on_connect command on collector deletion");
741
- }
742
-
500
_free_collector(tmp_collector);
501
745
-}
746
-/*
747
- * Take a buffer, encode it and rewrite it
748
- *
749
- */
750
-
751
-static char *aclk_encode_response(char *src, size_t content_size, int keep_newlines)
752
-{
753
- char *tmp_buffer = mallocz(content_size * 2);
754
- char *dst = tmp_buffer;
755
- while (content_size > 0) {
756
- switch (*src) {
757
- case '\n':
758
- if (keep_newlines)
759
- {
760
- *dst++ = '\\';
761
- *dst++ = 'n';
762
- }
763
- break;
764
- case '\t':
765
- break;
766
- case 0x01 ... 0x08:
767
- case 0x0b ... 0x1F:
768
- *dst++ = '\\';
769
- *dst++ = 'u';
770
- *dst++ = '0';
771
- *dst++ = '0';
772
- *dst++ = (*src < 0x0F) ? '0' : '1';
773
- *dst++ = to_hex(*src);
774
- break;
775
- case '\"':
776
- *dst++ = '\\';
777
- *dst++ = *src;
778
- break;
779
- default:
780
- *dst++ = *src;
781
- }
782
- src++;
783
- content_size--;
784
- }
785
- *dst = '\0';
786
-
787
- return tmp_buffer;
788
-}
789
-
790
-int aclk_execute_query(struct aclk_query *this_query)
791
-{
792
- if (strncmp(this_query->query, "/api/v1/", 8) == 0) {
793
- struct web_client *w = (struct web_client *)callocz(1, sizeof(struct web_client));
794
- w->response.data = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
795
- w->response.header = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
796
- w->response.header_output = buffer_create(NETDATA_WEB_RESPONSE_HEADER_SIZE);
797
- strcpy(w->origin, "*"); // Simulate web_client_create_on_fd()
798
- w->cookie1[0] = 0; // Simulate web_client_create_on_fd()
799
- w->cookie2[0] = 0; // Simulate web_client_create_on_fd()
800
- w->acl = 0x1f;
801
-
802
- char *mysep = strchr(this_query->query, '?');
803
- if (mysep) {
804
- strncpyz(w->decoded_query_string, mysep, NETDATA_WEB_REQUEST_URL_SIZE);
805
- *mysep = '\0';
806
- } else
807
- strncpyz(w->decoded_query_string, this_query->query, NETDATA_WEB_REQUEST_URL_SIZE);
808
-
809
- mysep = strrchr(this_query->query, '/');
810
-
811
- // TODO: handle bad response perhaps in a different way. For now it does to the payload
812
- w->response.code = web_client_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
813
- now_realtime_timeval(&w->tv_ready);
814
- w->response.data->date = w->tv_ready.tv_sec;
815
- web_client_build_http_header(w); // TODO: this function should offset from date, not tv_ready
816
- BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
817
- buffer_flush(local_buffer);
818
- local_buffer->contenttype = CT_APPLICATION_JSON;
819
-
820
- aclk_create_header(local_buffer, "http", this_query->msg_id, 0, 0);
821
- buffer_strcat(local_buffer, ",\n\t\"payload\": ");
822
- char *encoded_response = aclk_encode_response(w->response.data->buffer, w->response.data->len, 0);
823
- char *encoded_header = aclk_encode_response(w->response.header_output->buffer, w->response.header_output->len, 1);
824
-
825
- buffer_sprintf(
826
- local_buffer, "{\n\"code\": %d,\n\"body\": \"%s\",\n\"headers\": \"%s\"\n}",
827
- w->response.code, encoded_response, encoded_header);
828
-
829
- buffer_sprintf(local_buffer, "\n}");
830
-
831
- debug(D_ACLK, "Response:%s", encoded_header);
832
-
833
- aclk_send_message(this_query->topic, local_buffer->buffer, this_query->msg_id);
834
-
835
- buffer_free(w->response.data);
836
- buffer_free(w->response.header);
837
- buffer_free(w->response.header_output);
838
- freez(w);
839
- buffer_free(local_buffer);
840
- freez(encoded_response);
841
- freez(encoded_header);
842
- return 0;
843
- }
844
- return 1;
845
-}
846
-
847
-/*
848
- * This function will fetch the next pending command and process it
849
- *
850
- */
851
-int aclk_process_query()
852
-{
853
- struct aclk_query *this_query;
854
- static long int query_count = 0;
855
-
856
- if (!aclk_connected)
857
- return 0;
858
-
859
- this_query = aclk_queue_pop();
860
- if (likely(!this_query)) {
861
- return 0;
862
- }
863
-
864
- if (unlikely(this_query->deleted)) {
865
- debug(D_ACLK, "Garbage collect query %s:%s", this_query->topic, this_query->query);
866
- aclk_query_free(this_query);
867
- return 1;
868
- }
869
- query_count++;
870
-
871
- debug(
872
- D_ACLK, "Query #%ld (%s) size=%zu in queue %d seconds", query_count, this_query->topic,
873
- this_query->query ? strlen(this_query->query) : 0, (int)(now_realtime_sec() - this_query->created));
874
-
875
- switch (this_query->cmd) {
876
- case ACLK_CMD_ONCONNECT:
877
- debug(D_ACLK, "EXECUTING on connect metadata command");
878
- aclk_send_metadata();
879
- aclk_metadata_submitted = ACLK_METADATA_SENT;
880
- break;
881
-
882
- case ACLK_CMD_CHART:
883
- debug(D_ACLK, "EXECUTING a chart update command");
884
- aclk_send_single_chart(this_query->data, this_query->query);
885
- break;
886
-
887
- case ACLK_CMD_CHARTDEL:
888
- debug(D_ACLK, "EXECUTING a chart delete command");
889
- //TODO: This send the info metadata for now
890
- aclk_send_info_metadata();
891
- break;
892
-
893
- case ACLK_CMD_ALARM:
894
- debug(D_ACLK, "EXECUTING an alarm update command");
895
- aclk_send_message(this_query->topic, this_query->query, this_query->msg_id);
896
- break;
897
-
898
- case ACLK_CMD_CLOUD:
899
- debug(D_ACLK, "EXECUTING a cloud command");
900
- aclk_execute_query(this_query);
901
- break;
902
-
903
- default:
904
- break;
905
- }
906
- debug(D_ACLK, "Query #%ld (%s) done", query_count, this_query->topic);
907
-
908
- aclk_query_free(this_query);
909
-
910
- if (aclk_stats_enabled) {
911
- ACLK_STATS_LOCK;
912
- aclk_metrics_per_sample.queries_dispatched++;
913
- ACLK_STATS_UNLOCK;
914
- }
915
-
916
- return 1;
917
-}
918
-
919
-/*
920
- * Process all pending queries
921
- * Return 0 if no queries were processed, 1 otherwise
922
- *
923
- */
924
-
925
-int aclk_process_queries()
926
-{
927
- if (unlikely(netdata_exit || !aclk_connected))
928
- return 0;
929
-
930
- if (likely(!aclk_queue.count))
931
- return 0;
932
-
933
- debug(D_ACLK, "Processing %d queries", (int)aclk_queue.count);
934
-
935
- //TODO: may consider possible throttling here
936
- while (aclk_process_query()) {
937
- // Process all commands
938
- };
939
-
940
- return 1;
941
-}
942
-
943
-static void aclk_query_thread_cleanup(void *ptr)
944
-{
945
- struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
946
-
947
- info("cleaning up...");
948
-
949
- _reset_collector_list();
950
- freez(collector_list);
951
-
952
- // Clean memory for pending queries if any
953
- struct aclk_query *this_query;
954
-
955
- do {
956
- this_query = aclk_queue_pop();
957
- aclk_query_free(this_query);
958
- } while (this_query);
959
-
960
- freez(static_thread->thread);
961
- freez(static_thread);
962
-}
963
-
964
-/**
965
- * Main query processing thread
966
- *
967
- * On startup wait for the agent collectors to initialize
968
- * Expect at least a time of ACLK_STABLE_TIMEOUT seconds
969
- * of no new collectors coming in in order to mark the agent
970
- * as stable (set agent_state = AGENT_STABLE)
971
- */
972
-void *aclk_query_main_thread(void *ptr)
973
-{
974
- netdata_thread_cleanup_push(aclk_query_thread_cleanup, ptr);
975
-
976
- while (agent_state == AGENT_INITIALIZING && !netdata_exit) {
977
- time_t checkpoint;
978
-
979
- checkpoint = now_realtime_sec() - last_init_sequence;
980
- if (checkpoint > ACLK_STABLE_TIMEOUT) {
981
- agent_state = AGENT_STABLE;
982
- info("AGENT stable, last collector initialization activity was %ld seconds ago", checkpoint);
983
-#ifdef ACLK_DEBUG
984
- _dump_collector_list();
985
-#endif
986
- break;
987
- }
988
- info("Waiting for agent collectors to initialize. Last activity was %ld seconds ago" , checkpoint);
989
- sleep_usec(USEC_PER_SEC * 1);
990
- }
991
-
992
- while (!netdata_exit) {
993
- if (unlikely(!aclk_metadata_submitted)) {
994
- aclk_metadata_submitted = ACLK_METADATA_CMD_QUEUED;
995
- if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
996
- errno = 0;
997
- error("ACLK failed to queue on_connect command");
998
- aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
999
- }
1000
- }
1001
-
1002
- aclk_process_queries();
1003
-
1004
- QUERY_THREAD_LOCK;
1005
-
1006
- // TODO: Need to check if there are queries awaiting already
1007
- if (unlikely(pthread_cond_wait(&query_cond_wait, &query_lock_wait)))
1008
- sleep_usec(USEC_PER_SEC * 1);
1009
-
1010
- QUERY_THREAD_UNLOCK;
502
+ if (aclk_popcorn_check_bump())
503
+ return;
504
1012
- } // forever
1013
- info("Shutting down query processing thread");
1014
- netdata_thread_cleanup_pop(1);
1015
- return NULL;
505
+ if (unlikely(aclk_queue_query("collector", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT)))
506
+ debug(D_ACLK, "ACLK failed to queue on_connect command on collector deletion");
507
}
508
509
static void aclk_graceful_disconnect()
@@ -1047,23 +538,9 @@ static void aclk_graceful_disconnect()
538
aclk_shutting_down = 0;
539
}
540
1050
-
1051
-// Thread cleanup
1052
-static void aclk_main_cleanup(void *ptr)
1053
-{
1054
- struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
1055
- static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
1056
-
1057
- info("cleaning up...");
1058
-
1059
- char *agent_id = is_agent_claimed();
1060
- if (agent_id && aclk_connected) {
1061
- freez(agent_id);
1062
- // Wakeup thread to cleanup
1063
- QUERY_THREAD_WAKEUP;
1064
- aclk_graceful_disconnect();
1065
- }
1066
-}
541
+#ifndef __GNUC__
542
+#pragma region Incoming Msg Parsing
543
+#endif
544
545
struct dictionary_singleton {
546
char *key;
@@ -1092,6 +569,15 @@ int json_extract_singleton(JSON_ENTRY *e)
569
return 0;
570
}
571
572
+#ifndef __GNUC__
573
+#pragma endregion
574
+#endif
575
+
576
+
577
+#ifndef __GNUC__
578
+#pragma region Challenge Response
579
+#endif
580
+
581
// Base-64 decoder.
582
// Note: This is non-validating, invalid input will be decoded without an error.
583
// Challenges are packed into json strings so we don't skip newlines.
@@ -1225,29 +711,6 @@ int private_decrypt(unsigned char * enc_data, int data_len, unsigned char *decry
711
return result;
712
}
713
1228
-char *extract_payload(BUFFER *b)
1229
-{
1230
-char *s = b->buffer;
1231
-unsigned int line_len=0;
1232
- for (size_t i=0; i<b->len; i++)
1233
- {
1234
- if (*s == 0 )
1235
- return NULL;
1236
- if (*s == '\n' ) {
1237
- if (line_len==0)
1238
- return s+1;
1239
- line_len = 0;
1240
- }
1241
- else if (*s == '\r') {
1242
- /* don't count */
1243
- }
1244
- else
1245
- line_len ++;
1246
- s++;
1247
- }
1248
- return NULL;
1249
-}
1250
-
714
void aclk_get_challenge(char *aclk_hostname, char *aclk_port)
715
{
716
char *data_buffer = mallocz(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
@@ -1340,6 +803,10 @@ CLEANUP:
803
return;
804
}
805
806
+#ifndef __GNUC__
807
+#pragma endregion
808
+#endif
809
+
810
static void aclk_try_to_connect(char *hostname, char *port, int port_num)
811
{
812
if (!aclk_private_key) {
@@ -1359,7 +826,6 @@ static void aclk_try_to_connect(char *hostname, char *port, int port_num)
826
}
827
}
828
1362
-
829
/**
830
* Main agent cloud link thread
831
*
@@ -1373,8 +839,10 @@ static void aclk_try_to_connect(char *hostname, char *port, int port_num)
839
void *aclk_main(void *ptr)
840
{
841
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
1376
- struct netdata_static_thread *query_thread;
1377
- struct netdata_static_thread *stats_thread = NULL;
842
+ struct aclk_query_threads query_threads;
843
+ struct aclk_stats_thread *stats_thread = NULL;
844
+
845
+ query_threads.thread_list = NULL;
846
847
// This thread is unusual in that it cannot be cancelled by cancel_main_threads()
848
// as it must notify the far end that it shutdown gracefully and avoid the LWT.
@@ -1400,18 +868,25 @@ void *aclk_main(void *ptr)
868
}
869
}
870
1403
- aclk_stats_enabled = appconfig_get_boolean(&cloud_config, CONFIG_SECTION_GLOBAL, "statistics", CONFIG_BOOLEAN_YES);
871
+ query_threads.count = config_get_number(CONFIG_SECTION_CLOUD, "query thread count", 2);
872
+ if(query_threads.count < 1) {
873
+ error("You need at least one query thread. Overriding configured setting of \"%d\"", query_threads.count);
874
+ query_threads.count = 1;
875
+ config_set_number(CONFIG_SECTION_CLOUD, "query thread count", query_threads.count);
876
+ }
877
+
878
+ aclk_shared_state.last_popcorn_interrupt = now_realtime_sec(); // without mutex here because threads are not yet started
879
+
880
+ aclk_stats_enabled = config_get_boolean(CONFIG_SECTION_CLOUD, "statistics", CONFIG_BOOLEAN_YES);
881
if (aclk_stats_enabled) {
1405
- stats_thread = callocz(1, sizeof(struct netdata_static_thread));
882
+ stats_thread = callocz(1, sizeof(struct aclk_stats_thread));
883
stats_thread->thread = mallocz(sizeof(netdata_thread_t));
884
+ stats_thread->query_thread_count = query_threads.count;
885
netdata_thread_create(
886
stats_thread->thread, ACLK_STATS_THREAD_NAME, NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread,
887
stats_thread);
888
}
889
1412
- last_init_sequence = now_realtime_sec();
1413
- query_thread = NULL;
1414
-
890
char *aclk_hostname = NULL; // Initializers are over-written but prevent gcc complaining about clobbering.
891
char *aclk_port = NULL;
892
uint32_t port_num = 0;
@@ -1508,17 +983,13 @@ void *aclk_main(void *ptr)
983
aclk_subscribed = !aclk_subscribe(ACLK_COMMAND_TOPIC, 1);
984
}
985
1511
- if (unlikely(!query_thread)) {
1512
- query_thread = callocz(1, sizeof(struct netdata_static_thread));
1513
- query_thread->thread = mallocz(sizeof(netdata_thread_t));
1514
- netdata_thread_create(
1515
- query_thread->thread, ACLK_THREAD_NAME, NETDATA_THREAD_OPTION_DEFAULT, aclk_query_main_thread,
1516
- query_thread);
986
+ if (unlikely(!query_threads.thread_list)) {
987
+ aclk_query_threads_start(&query_threads);
988
}
989
} // forever
990
exited:
991
// Wakeup query thread to cleanup
1521
- QUERY_THREAD_WAKEUP;
992
+ QUERY_THREAD_WAKEUP_ALL;
993
994
freez(aclk_username);
995
freez(aclk_password);
@@ -1527,10 +998,24 @@ exited:
998
if (aclk_private_key != NULL)
999
RSA_free(aclk_private_key);
1000
1530
- aclk_main_cleanup(ptr);
1001
+ static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
1002
+
1003
+ char *agent_id = is_agent_claimed();
1004
+ if (agent_id && aclk_connected) {
1005
+ freez(agent_id);
1006
+ // Wakeup thread to cleanup
1007
+ QUERY_THREAD_WAKEUP;
1008
+ aclk_graceful_disconnect();
1009
+ }
1010
+
1011
+ aclk_query_threads_cleanup(&query_threads);
1012
+
1013
+ _reset_collector_list();
1014
+ freez(collector_list);
1015
1016
if(aclk_stats_enabled) {
1017
netdata_thread_join(*stats_thread->thread, NULL);
1018
+ aclk_stats_thread_cleanup();
1019
freez(stats_thread->thread);
1020
freez(stats_thread);
1021
}
@@ -1626,12 +1111,11 @@ int aclk_subscribe(char *sub_topic, int qos)
1111
// This is called from a callback when the link goes up
1112
void aclk_connect()
1113
{
1629
- info("Connection detected (%"PRIu64" queued queries)", aclk_queue.count);
1114
+ info("Connection detected (%u queued queries)", aclk_query_size());
1115
1116
aclk_stats_upd_online(1);
1117
1118
aclk_connected = 1;
1634
- waiting_init = 0;
1119
aclk_reconnect_delay(0);
1120
QUERY_THREAD_WAKEUP;
1121
return;
@@ -1641,13 +1125,14 @@ void aclk_connect()
1125
void aclk_disconnect()
1126
{
1127
if (likely(aclk_connected))
1644
- info("Disconnect detected (%"PRIu64" queued queries)", aclk_queue.count);
1128
+ info("Disconnect detected (%u queued queries)", aclk_query_size());
1129
1130
aclk_stats_upd_online(0);
1131
1132
aclk_subscribed = 0;
1649
- aclk_metadata_submitted = ACLK_METADATA_REQUIRED;
1650
- waiting_init = 1;
1133
+ ACLK_SHARED_STATE_LOCK;
1134
+ aclk_shared_state.metadata_submitted = ACLK_METADATA_REQUIRED;
1135
+ ACLK_SHARED_STATE_UNLOCK;
1136
aclk_connected = 0;
1137
aclk_connecting = 0;
1138
aclk_force_reconnect = 1;
@@ -1692,7 +1177,8 @@ inline void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts
1177
* active alarms
1178
*/
1179
void health_active_log_alarms_2json(RRDHOST *host, BUFFER *wb);
1695
-void aclk_send_alarm_metadata()
1180
+
1181
+void aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted)
1182
{
1183
BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1184
@@ -1706,7 +1192,8 @@ void aclk_send_alarm_metadata()
1192
// use the session time as the fake timestamp to indicate that it starts the session. If it is
1193
// a fake on_connect message then use the real timestamp to indicate it is within the existing
1194
// session.
1709
- if (aclk_metadata_submitted == ACLK_METADATA_SENT)
1195
+
1196
+ if (metadata_submitted == ACLK_METADATA_SENT)
1197
aclk_create_header(local_buffer, "connect_alarms", msg_id, 0, 0);
1198
else
1199
aclk_create_header(local_buffer, "connect_alarms", msg_id, aclk_session_sec, aclk_session_us);
@@ -1737,7 +1224,7 @@ void aclk_send_alarm_metadata()
1224
* /api/v1/info
1225
* charts
1226
*/
1740
-int aclk_send_info_metadata()
1227
+int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted)
1228
{
1229
BUFFER *local_buffer = buffer_create(NETDATA_WEB_RESPONSE_INITIAL_SIZE);
1230
@@ -1751,7 +1238,7 @@ int aclk_send_info_metadata()
1238
// use the session time as the fake timestamp to indicate that it starts the session. If it is
1239
// a fake on_connect message then use the real timestamp to indicate it is within the existing
1240
// session.
1754
- if (aclk_metadata_submitted == ACLK_METADATA_SENT)
1241
+ if (metadata_submitted == ACLK_METADATA_SENT)
1242
aclk_create_header(local_buffer, "update", msg_id, 0, 0);
1243
else
1244
aclk_create_header(local_buffer, "connect", msg_id, aclk_session_sec, aclk_session_us);
@@ -1794,11 +1281,11 @@ void aclk_send_stress_test(size_t size)
1281
1282
// Send info metadata message to the cloud if the link is established
1283
// or on request
1797
-int aclk_send_metadata()
1284
+int aclk_send_metadata(ACLK_METADATA_STATE state)
1285
{
1286
1800
- aclk_send_info_metadata();
1801
- aclk_send_alarm_metadata();
1287
+ aclk_send_info_metadata(state);
1288
+ aclk_send_alarm_metadata(state);
1289
1290
return 0;
1291
}
@@ -1816,8 +1303,13 @@ void aclk_single_update_enable()
1303
// Trigged by a health reload, sends the alarm metadata
1304
void aclk_alarm_reload()
1305
{
1819
- if (unlikely(agent_state == AGENT_INITIALIZING))
1306
+
1307
+ ACLK_SHARED_STATE_LOCK;
1308
+ if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1309
+ ACLK_SHARED_STATE_UNLOCK;
1310
return;
1311
+ }
1312
+ ACLK_SHARED_STATE_UNLOCK;
1313
1314
if (unlikely(aclk_queue_query("on_connect", NULL, NULL, NULL, 0, 1, ACLK_CMD_ONCONNECT))) {
1315
if (likely(aclk_connected)) {
@@ -1881,16 +1373,16 @@ int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd)
1373
if (unlikely(aclk_disable_single_updates))
1374
return 0;
1375
1884
- if (unlikely(agent_state == AGENT_INITIALIZING))
1885
- last_init_sequence = now_realtime_sec();
1886
- else {
1887
- if (unlikely(aclk_queue_query("_chart", host->hostname, NULL, chart_name, 0, 1, aclk_cmd))) {
1888
- if (likely(aclk_connected)) {
1889
- errno = 0;
1890
- error("ACLK failed to queue chart_update command");
1891
- }
1376
+ if (aclk_popcorn_check_bump())
1377
+ return 0;
1378
+
1379
+ if (unlikely(aclk_queue_query("_chart", host->hostname, NULL, chart_name, 0, 1, aclk_cmd))) {
1380
+ if (likely(aclk_connected)) {
1381
+ errno = 0;
1382
+ error("ACLK failed to queue chart_update command");
1383
}
1384
}
1385
+
1386
return 0;
1387
#endif
1388
}
@@ -1905,8 +1397,12 @@ int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae)
1397
if (host != localhost)
1398
return 0;
1399
1908
- if (unlikely(agent_state == AGENT_INITIALIZING))
1400
+ ACLK_SHARED_STATE_LOCK;
1401
+ if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1402
+ ACLK_SHARED_STATE_UNLOCK;
1403
return 0;
1404
+ }
1405
+ ACLK_SHARED_STATE_UNLOCK;
1406
1407
/*
1408
* Check if individual updates have been disabled
@@ -1959,10 +1455,13 @@ int aclk_handle_cloud_request(char *payload)
1455
ACLK_STATS_UNLOCK;
1456
}
1457
1962
- if (unlikely(agent_state == AGENT_INITIALIZING)) {
1458
+ ACLK_SHARED_STATE_LOCK;
1459
+ if (unlikely(aclk_shared_state.agent_state == AGENT_INITIALIZING)) {
1460
debug(D_ACLK, "Ignoring cloud request; agent not in stable state");
1461
+ ACLK_SHARED_STATE_UNLOCK;
1462
return 0;
1463
}
1464
+ ACLK_SHARED_STATE_UNLOCK;
1465
1466
if (unlikely(!payload)) {
1467
debug(D_ACLK, "ACLK incoming message is empty");
@@ -2010,7 +1509,7 @@ int aclk_handle_cloud_request(char *payload)
1509
cloud_to_agent.type_id = NULL;
1510
}
1511
2013
- if (unlikely(aclk_submit_request(&cloud_to_agent)))
1512
+ if (unlikely(aclk_queue_query(cloud_to_agent.callback_topic, NULL, cloud_to_agent.msg_id, cloud_to_agent.payload, 0, 0, ACLK_CMD_CLOUD)))
1513
debug(D_ACLK, "ACLK failed to queue incoming message (%s)", payload);
1514
1515
// Note: the payload comes from the callback and it will be automatically freed
aclk/agent_cloud_link.h
+6
-30
@@ -5,6 +5,7 @@
5
6
#include "../daemon/common.h"
7
#include "mqtt.h"
8
+#include "aclk_common.h"
9
10
#define ACLK_VERSION 1
11
#define ACLK_THREAD_NAME "ACLK_Query"
@@ -25,7 +26,6 @@
26
#define ACLK_MAX_TOPIC 255
27
28
#define ACLK_RECONNECT_DELAY 1 // reconnect delay -- with backoff stragegy fow now
28
-#define ACLK_STABLE_TIMEOUT 3 // Minimum delay to mark AGENT as stable
29
#define ACLK_DEFAULT_PORT 9002
30
#define ACLK_DEFAULT_HOST "localhost"
31
@@ -37,27 +37,6 @@ struct aclk_request {
37
int version;
38
};
39
40
-typedef enum aclk_cmd {
41
- ACLK_CMD_CLOUD,
42
- ACLK_CMD_ONCONNECT,
43
- ACLK_CMD_INFO,
44
- ACLK_CMD_CHART,
45
- ACLK_CMD_CHARTDEL,
46
- ACLK_CMD_ALARM,
47
- ACLK_CMD_MAX
48
-} ACLK_CMD;
49
-
50
-typedef enum aclk_metadata_state {
51
- ACLK_METADATA_REQUIRED,
52
- ACLK_METADATA_CMD_QUEUED,
53
- ACLK_METADATA_SENT
54
-} ACLK_METADATA_STATE;
55
-
56
-typedef enum agent_state {
57
- AGENT_INITIALIZING,
58
- AGENT_STABLE
59
-} AGENT_STATE;
60
-
40
typedef enum aclk_init_action { ACLK_INIT, ACLK_REINIT } ACLK_INIT_ACTION;
41
42
void *aclk_main(void *ptr);
@@ -82,25 +61,22 @@ int aclk_subscribe(char *topic, int qos);
61
int cloud_to_agent_parse(JSON_ENTRY *e);
62
void aclk_disconnect();
63
void aclk_connect();
85
-int aclk_send_metadata();
86
-int aclk_send_info_metadata();
64
+
65
+int aclk_send_metadata(ACLK_METADATA_STATE state);
66
+int aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted);
67
+void aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted);
68
+
69
int aclk_wait_for_initialization();
70
char *create_publish_base_topic();
71
72
int aclk_send_single_chart(char *host, char *chart);
91
-int aclk_queue_query(char *token, char *data, char *msg_type, char *query, int run_after, int internal, ACLK_CMD cmd);
92
-struct aclk_query *
93
-aclk_query_find(char *token, char *data, char *msg_id, char *query, ACLK_CMD cmd, struct aclk_query **last_query);
73
int aclk_update_chart(RRDHOST *host, char *chart_name, ACLK_CMD aclk_cmd);
74
int aclk_update_alarm(RRDHOST *host, ALARM_ENTRY *ae);
75
void aclk_create_header(BUFFER *dest, char *type, char *msg_id, time_t ts_secs, usec_t ts_us);
76
int aclk_handle_cloud_request(char *payload);
98
-int aclk_submit_request(struct aclk_request *);
77
void aclk_add_collector(const char *hostname, const char *plugin_name, const char *module_name);
78
void aclk_del_collector(const char *hostname, const char *plugin_name, const char *module_name);
79
void aclk_alarm_reload();
102
-void aclk_send_alarm_metadata();
103
-int aclk_execute_query(struct aclk_query *query);
80
unsigned long int aclk_reconnect_delay(int mode);
81
extern void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST *host);
82
void aclk_single_update_enable();