@cryptotaxi247 / netdata-1 / commits / 6f3b2d8a2

Add ACLK synchronization event loop (#11396)

Stelios Fragkakis committed Aug 11, 2021 at 17:13 UTC 6f3b2d8a2a9e145ee2bd098ea2f4b653ef6bd8ed
10 files changed +1112 -10
CMakeLists.txt
+2
@@ -621,6 +621,8 @@ set(RRD_PLUGIN_FILES
621 database/rrdvar.h
622 database/sqlite/sqlite_functions.c
623 database/sqlite/sqlite_functions.h
624 + database/sqlite/sqlite_aclk.c
625 + database/sqlite/sqlite_aclk.h
626 database/sqlite/sqlite3.c
627 database/sqlite/sqlite3.h
628 database/engine/rrdengine.c
Makefile.am
+2
@@ -400,6 +400,8 @@ RRD_PLUGIN_FILES = \
400 database/rrdvar.h \
401 database/sqlite/sqlite_functions.c \
402 database/sqlite/sqlite_functions.h \
403 + database/sqlite/sqlite_aclk.c \
404 + database/sqlite/sqlite_aclk.h \
405 database/sqlite/sqlite3.c \
406 database/sqlite/sqlite3.h \
407 $(NULL)
aclk/legacy/agent_cloud_link.h
+2
@@ -55,6 +55,7 @@ int legacy_cloud_to_agent_parse(JSON_ENTRY *e);
55 void aclk_disconnect();
56 void aclk_connect();
57
58 +#ifdef ENABLE_ACLK
59 int aclk_send_metadata(ACLK_METADATA_STATE state, RRDHOST *host);
60 int legacy_aclk_send_info_metadata(ACLK_METADATA_STATE metadata_submitted, RRDHOST *host);
61 void legacy_aclk_send_alarm_metadata(ACLK_METADATA_STATE metadata_submitted);
@@ -76,5 +77,6 @@ extern void health_alarm_entry2json_nolock(BUFFER *wb, ALARM_ENTRY *ae, RRDHOST
77 void legacy_aclk_host_state_update(RRDHOST *host, int connect);
78 int aclk_send_info_child_connection(RRDHOST *host, ACLK_CMD cmd);
79 void aclk_update_next_child_to_popcorn(void);
80 +#endif
81
82 #endif //NETDATA_AGENT_CLOUD_LINK_H
database/rrd.h
+2 -1
@@ -795,6 +795,7 @@ struct rrdhost {
795 struct sender_state *sender;
796 volatile unsigned int rrdpush_sender_spawn:1; // 1 when the sender thread has been spawn
797 netdata_thread_t rrdpush_sender_thread; // the sender thread
798 + void *dbsync_worker;
799
800 volatile unsigned int rrdpush_sender_connected:1; // 1 when the sender is ready to push metrics
801 int rrdpush_sender_socket; // the fd of the socket to the remote host, or -1
@@ -1350,5 +1351,5 @@ extern void set_host_properties(
1351 #include "database/engine/rrdengineapi.h"
1352 #endif
1353 #include "sqlite/sqlite_functions.h"
1353 -
1354 +#include "sqlite/sqlite_aclk.h"
1355 #endif /* NETDATA_RRD_H */
database/rrdhost.c
+1
@@ -719,6 +719,7 @@ int rrd_init(char *hostname, struct rrdhost_system_info *system_info) {
719 fatal("Failed to initialize dbengine");
720 }
721 #endif
722 + sql_aclk_sync_init();
723 rrd_unlock();
724
725 web_client_api_v1_management_init();
database/sqlite/sqlite_aclk.c new
+858
@@ -0,0 +1,858 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "sqlite_functions.h"
4 +#include "sqlite_aclk.h"
5 +
6 +// TODO: To be added
7 +//#include "sqlite_aclk_chart.h"
8 +//#include "sqlite_aclk_alert.h"
9 +//#include "sqlite_aclk_node.h"
10 +
11 +const char *aclk_sync_config[] = {
12 + NULL,
13 + "CREATE TABLE IF NOT EXISTS delete_dimension(host_id, chart_id, dim_id, chart_name, dimension_id, dimension_name);"
14 + "CREATE TRIGGER IF NOT EXISTS tr_del_dim AFTER DELETE ON dimension " \
15 + "BEGIN INSERT INTO delete_dimension (host_id, chart_id, dim_id, chart_name, dimension_id, dimension_name) " \
16 + "SELECT c.host_id, c.chart_id, old.dim_id, c.type||'.'||c.id, old.id, old.name from chart c where c.chart_id = old.chart_id; end;",
17 + NULL
18 +};
19 +
20 +int aclk_architecture = 0;
21 +
22 +uv_mutex_t aclk_async_lock;
23 +struct aclk_database_worker_config *aclk_thread_head = NULL;
24 +
25 +static inline int claimed()
26 +{
27 + int rc;
28 + rrdhost_aclk_state_lock(localhost);
29 + rc = (localhost->aclk_state.claimed_id != NULL);
30 + rrdhost_aclk_state_unlock(localhost);
31 + return rc;
32 +}
33 +
34 +void aclk_add_worker_thread(struct aclk_database_worker_config *wc)
35 +{
36 + if (unlikely(!wc))
37 + return;
38 +
39 + uv_mutex_lock(&aclk_async_lock);
40 + if (unlikely(!wc->host)) {
41 + wc->next = aclk_thread_head;
42 + aclk_thread_head = wc;
43 + }
44 + uv_mutex_unlock(&aclk_async_lock);
45 + return;
46 +}
47 +
48 +void aclk_del_worker_thread(struct aclk_database_worker_config *wc)
49 +{
50 + if (unlikely(!wc))
51 + return;
52 +
53 + uv_mutex_lock(&aclk_async_lock);
54 + struct aclk_database_worker_config **tmp = &aclk_thread_head;
55 + while ((*tmp) != wc)
56 + tmp = &(*tmp)->next;
57 + *tmp = wc->next;
58 + uv_mutex_unlock(&aclk_async_lock);
59 + return;
60 +}
61 +
62 +int aclk_worker_thread_exists(char *guid)
63 +{
64 + int rc = 0;
65 + uv_mutex_lock(&aclk_async_lock);
66 +
67 + struct aclk_database_worker_config *tmp = aclk_thread_head;
68 +
69 + while (tmp && !rc) {
70 + rc = strcmp(tmp->uuid_str, guid) == 0;
71 + tmp = tmp->next;
72 + }
73 + uv_mutex_unlock(&aclk_async_lock);
74 + return rc;
75 +}
76 +
77 +void aclk_database_init_cmd_queue(struct aclk_database_worker_config *wc)
78 +{
79 + wc->cmd_queue.head = wc->cmd_queue.tail = 0;
80 + wc->queue_size = 0;
81 + fatal_assert(0 == uv_cond_init(&wc->cmd_cond));
82 + fatal_assert(0 == uv_mutex_init(&wc->cmd_mutex));
83 +}
84 +
85 +void aclk_database_enq_cmd_nowake(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd)
86 +{
87 + unsigned queue_size;
88 +
89 + /* wait for free space in queue */
90 + uv_mutex_lock(&wc->cmd_mutex);
91 + while ((queue_size = wc->queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
92 + uv_cond_wait(&wc->cmd_cond, &wc->cmd_mutex);
93 + }
94 + fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
95 + /* enqueue command */
96 + wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;
97 + wc->cmd_queue.tail = wc->cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
98 + wc->cmd_queue.tail + 1 : 0;
99 + wc->queue_size = queue_size + 1;
100 + uv_mutex_unlock(&wc->cmd_mutex);
101 +}
102 +
103 +int aclk_database_enq_cmd_noblock(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd)
104 +{
105 + unsigned queue_size;
106 +
107 + /* wait for free space in queue */
108 + uv_mutex_lock(&wc->cmd_mutex);
109 + if ((queue_size = wc->queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
110 + uv_mutex_unlock(&wc->cmd_mutex);
111 + return 1;
112 + }
113 +
114 + fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
115 + /* enqueue command */
116 + wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;
117 + wc->cmd_queue.tail = wc->cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
118 + wc->cmd_queue.tail + 1 : 0;
119 + wc->queue_size = queue_size + 1;
120 + uv_mutex_unlock(&wc->cmd_mutex);
121 + return 0;
122 +}
123 +
124 +void aclk_database_enq_cmd(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd)
125 +{
126 + unsigned queue_size;
127 +
128 + /* wait for free space in queue */
129 + uv_mutex_lock(&wc->cmd_mutex);
130 + while ((queue_size = wc->queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
131 + uv_cond_wait(&wc->cmd_cond, &wc->cmd_mutex);
132 + }
133 + fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
134 + /* enqueue command */
135 + wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;
136 + wc->cmd_queue.tail = wc->cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
137 + wc->cmd_queue.tail + 1 : 0;
138 + wc->queue_size = queue_size + 1;
139 + uv_mutex_unlock(&wc->cmd_mutex);
140 +
141 + /* wake up event loop */
142 + fatal_assert(0 == uv_async_send(&wc->async));
143 +}
144 +
145 +struct aclk_database_cmd aclk_database_deq_cmd(struct aclk_database_worker_config* wc)
146 +{
147 + struct aclk_database_cmd ret;
148 + unsigned queue_size;
149 +
150 + uv_mutex_lock(&wc->cmd_mutex);
151 + queue_size = wc->queue_size;
152 + if (queue_size == 0) {
153 + ret.opcode = ACLK_DATABASE_NOOP;
154 + ret.completion = NULL;
155 + } else {
156 + /* dequeue command */
157 + ret = wc->cmd_queue.cmd_array[wc->cmd_queue.head];
158 + if (queue_size == 1) {
159 + wc->cmd_queue.head = wc->cmd_queue.tail = 0;
160 + } else {
161 + wc->cmd_queue.head = wc->cmd_queue.head != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
162 + wc->cmd_queue.head + 1 : 0;
163 + }
164 + wc->queue_size = queue_size - 1;
165 +
166 + /* wake up producers */
167 + uv_cond_signal(&wc->cmd_cond);
168 + }
169 + uv_mutex_unlock(&wc->cmd_mutex);
170 +
171 + return ret;
172 +}
173 +
174 +int aclk_start_sync_thread(void *data, int argc, char **argv, char **column)
175 +{
176 + char uuid_str[GUID_LEN + 1];
177 + UNUSED(data);
178 + UNUSED(argc);
179 + UNUSED(column);
180 +
181 + uuid_unparse_lower(*((uuid_t *) argv[0]), uuid_str);
182 +
183 + if (rrdhost_find_by_guid(uuid_str, 0) == localhost)
184 + return 0;
185 +
186 + sql_create_aclk_table(NULL, (uuid_t *) argv[0], (uuid_t *) argv[1]);
187 + return 0;
188 +}
189 +
190 +void sql_aclk_sync_init(void)
191 +{
192 +#ifdef ACLK_NEWARCH_DEVMODE
193 + char *err_msg = NULL;
194 + int rc;
195 +
196 + if (unlikely(!db_meta)) {
197 + if (default_rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE) {
198 + return;
199 + }
200 + error_report("Database has not been initialized");
201 + return;
202 + }
203 +
204 + info("SQLite aclk sync initialization");
205 +
206 + for (int i = 0; aclk_sync_config[i]; i++) {
207 + debug(D_ACLK_SYNC, "Executing %s", aclk_sync_config[i]);
208 + rc = sqlite3_exec(db_meta, aclk_sync_config[i], 0, 0, &err_msg);
209 + if (rc != SQLITE_OK) {
210 + error_report("SQLite error aclk sync initialization setup, rc = %d (%s)", rc, err_msg);
211 + error_report("SQLite failed statement %s", aclk_sync_config[i]);
212 + sqlite3_free(err_msg);
213 + return;
214 + }
215 + }
216 + info("SQLite aclk sync initialization completed");
217 + fatal_assert(0 == uv_mutex_init(&aclk_async_lock));
218 +
219 + rc = sqlite3_exec(db_meta, "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni WHERE "
220 + "h.host_id = ni.host_id AND ni.node_id IS NOT NULL;", aclk_start_sync_thread, NULL, NULL);
221 +#endif
222 + return;
223 +}
224 +
225 +static void async_cb(uv_async_t *handle)
226 +{
227 + uv_stop(handle->loop);
228 + uv_update_time(handle->loop);
229 + debug(D_ACLK_SYNC, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
230 +}
231 +
232 +#define TIMER_PERIOD_MS (1000)
233 +
234 +static void timer_cb(uv_timer_t* handle)
235 +{
236 + struct aclk_database_worker_config *wc = handle->data;
237 + uv_stop(handle->loop);
238 + uv_update_time(handle->loop);
239 +
240 + struct aclk_database_cmd cmd;
241 + cmd.opcode = ACLK_DATABASE_TIMER;
242 + cmd.completion = NULL;
243 + aclk_database_enq_cmd_noblock(wc, &cmd);
244 +
245 + if (wc->cleanup_after && wc->cleanup_after < now_realtime_sec()) {
246 + cmd.opcode = ACLK_DATABASE_CLEANUP;
247 + cmd.completion = NULL;
248 + if (!aclk_database_enq_cmd_noblock(wc, &cmd))
249 + wc->cleanup_after += ACLK_DATABASE_CLEANUP_INTERVAL;
250 + }
251 +
252 + if (wc->chart_updates) {
253 + cmd.opcode = ACLK_DATABASE_PUSH_CHART;
254 + cmd.count = ACLK_MAX_CHART_BATCH;
255 + cmd.completion = NULL;
256 + cmd.param1 = ACLK_MAX_CHART_BATCH_COUNT;
257 + aclk_database_enq_cmd_noblock(wc, &cmd);
258 + }
259 +
260 + if (wc->alert_updates) {
261 + cmd.opcode = ACLK_DATABASE_PUSH_ALERT;
262 + cmd.count = ACLK_MAX_ALERT_UPDATES;
263 + cmd.completion = NULL;
264 + aclk_database_enq_cmd_noblock(wc, &cmd);
265 + }
266 +}
267 +
268 +#define MAX_CMD_BATCH_SIZE (256)
269 +
270 +void aclk_database_worker(void *arg)
271 +{
272 + struct aclk_database_worker_config *wc = arg;
273 + uv_loop_t *loop;
274 + int shutdown, ret;
275 + enum aclk_database_opcode opcode;
276 + uv_timer_t timer_req;
277 + struct aclk_database_cmd cmd;
278 + unsigned cmd_batch_size;
279 +
280 + aclk_database_init_cmd_queue(wc);
281 +
282 + char threadname[NETDATA_THREAD_NAME_MAX+1];
283 + if (wc->host)
284 + snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", wc->host->hostname);
285 + else {
286 + snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", wc->uuid_str);
287 + threadname[11] = '\0';
288 + }
289 + uv_thread_set_name_np(wc->thread, threadname);
290 +
291 + loop = wc->loop = mallocz(sizeof(uv_loop_t));
292 + ret = uv_loop_init(loop);
293 + if (ret) {
294 + error("uv_loop_init(): %s", uv_strerror(ret));
295 + goto error_after_loop_init;
296 + }
297 + loop->data = wc;
298 +
299 + ret = uv_async_init(wc->loop, &wc->async, async_cb);
300 + if (ret) {
301 + error("uv_async_init(): %s", uv_strerror(ret));
302 + goto error_after_async_init;
303 + }
304 + wc->async.data = wc;
305 +
306 + ret = uv_timer_init(loop, &timer_req);
307 + if (ret) {
308 + error("uv_timer_init(): %s", uv_strerror(ret));
309 + goto error_after_timer_init;
310 + }
311 + timer_req.data = wc;
312 + fatal_assert(0 == uv_timer_start(&timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
313 +
314 + wc->error = 0;
315 + shutdown = 0;
316 +
317 + aclk_add_worker_thread(wc);
318 +
319 + info("Starting ACLK sync event loop for host with GUID %s (Host is '%s')", wc->host_guid, wc->host ? "connected" : "not connected");
320 +// TODO: To be added
321 +// sql_get_last_chart_sequence(wc, cmd);
322 + while (likely(shutdown == 0)) {
323 + uv_run(loop, UV_RUN_DEFAULT);
324 +
325 + if (netdata_exit)
326 + shutdown = 1;
327 +
328 + /* wait for commands */
329 + cmd_batch_size = 0;
330 + do {
331 + if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE))
332 + break;
333 + cmd = aclk_database_deq_cmd(wc);
334 + opcode = cmd.opcode;
335 + ++cmd_batch_size;
336 + db_lock();
337 + switch (opcode) {
338 + case ACLK_DATABASE_NOOP:
339 + /* the command queue was empty, do nothing */
340 + break;
341 +
342 +// MAINTENANCE
343 + case ACLK_DATABASE_CLEANUP:
344 + debug(D_ACLK_SYNC, "Database cleanup for %s", wc->host_guid);
345 + sql_maint_aclk_sync_database(wc, cmd);
346 + if (wc->host == localhost)
347 + sql_check_aclk_table_list(wc);
348 + break;
349 + case ACLK_DATABASE_CHECK:
350 + debug(D_ACLK_SYNC, "Checking database dimensions for %s", wc->host_guid);
351 +// sql_check_dimension_state(wc, cmd);
352 + break;
353 + case ACLK_DATABASE_CHECK_ROTATION:
354 + debug(D_ACLK_SYNC, "Checking database for rotation %s", wc->host_guid);
355 +// sql_check_rotation_state(wc, cmd);
356 + break;
357 + case ACLK_DATABASE_DELETE_HOST:
358 + debug(D_ACLK_SYNC,"Cleaning ACLK tables for %s", (char *) cmd.data);
359 + sql_delete_aclk_table_list(wc, cmd);
360 + break;
361 +
362 +// CHART / DIMENSION OPERATIONS
363 + case ACLK_DATABASE_PUSH_CHART:
364 + debug(D_ACLK_SYNC, "Pushing chart info to the cloud for node %s", wc->host_guid);
365 +// aclk_push_chart_event(wc, cmd);
366 + break;
367 + case ACLK_DATABASE_PUSH_CHART_CONFIG:
368 + debug(D_ACLK_SYNC, "Pushing chart config info to the cloud for node %s", wc->host_guid);
369 +// aclk_push_chart_config(wc, cmd);
370 + break;
371 + case ACLK_DATABASE_CHART_ACK:
372 + debug(D_ACLK_SYNC, "ACK chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1);
373 +// sql_set_chart_ack(wc, cmd);
374 + break;
375 + case ACLK_DATABASE_RESET_CHART:
376 + debug(D_ACLK_SYNC, "RESET chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1);
377 +// sql_reset_chart_event(wc, cmd);
378 + break;
379 + case ACLK_DATABASE_STATUS_CHART:
380 + debug(D_ACLK_SYNC,"Requesting chart status for %s", wc->host_guid);
381 +// aclk_status_chart_event(wc, cmd);
382 + break;
383 + case ACLK_DATABASE_ADD_CHART:
384 + debug(D_ACLK_SYNC,"Adding chart event for %s", wc->host_guid);
385 +// aclk_add_chart_event(wc, cmd);
386 + break;
387 + case ACLK_DATABASE_ADD_DIMENSION:
388 + debug(D_ACLK_SYNC,"Adding dimension event for %s", wc->host_guid);
389 +// aclk_add_dimension_event(wc, cmd);
390 + break;
391 + case ACLK_DATABASE_DEDUP_CHART:
392 + debug(D_ACLK_SYNC,"Running chart deduplication for %s", wc->host_guid);
393 +// sql_chart_deduplicate(wc, cmd);
394 + break;
395 + case ACLK_DATABASE_SYNC_CHART_SEQ:
396 + debug(D_ACLK_SYNC,"Calculatting chart sequence for %s", wc->host_guid);
397 +// sql_get_last_chart_sequence(wc, cmd);
398 + break;
399 +
400 +// ALERTS
401 + case ACLK_DATABASE_ADD_ALERT:
402 + debug(D_ACLK_SYNC,"Adding alert event for %s", wc->host_guid);
403 +// aclk_add_alert_event(wc, cmd);
404 + break;
405 + case ACLK_DATABASE_PUSH_ALERT_CONFIG:
406 + debug(D_ACLK_SYNC,"Pushing chart config info to the cloud for %s", wc->host_guid);
407 +// aclk_push_alert_config_event(wc, cmd);
408 + break;
409 + case ACLK_DATABASE_PUSH_ALERT:
410 + debug(D_ACLK_SYNC, "Pushing alert info to the cloud for %s", wc->host_guid);
411 +// aclk_push_alert_event(wc, cmd);
412 + break;
413 + case ACLK_DATABASE_ALARM_HEALTH_LOG:
414 + debug(D_ACLK_SYNC, "Pushing alarm health log to the cloud for %s", wc->host_guid);
415 +// aclk_push_alarm_health_log(wc, cmd);
416 + break;
417 +
418 +// NODE OPERATIONS
419 + case ACLK_DATABASE_RESET_NODE:
420 + debug(D_ACLK_SYNC,"Resetting the node instance id of %s", (char *) cmd.data);
421 +// aclk_reset_node_event(wc, cmd);
422 + break;
423 + case ACLK_DATABASE_NODE_INFO:
424 + debug(D_ACLK_SYNC,"Sending node info for %s", wc->uuid_str);
425 +// sql_build_node_info(wc, cmd);
426 + break;
427 + case ACLK_DATABASE_UPD_STATS:
428 +// sql_update_metric_statistics(wc, cmd);
429 + break;
430 +
431 +// NODE_INSTANCE DETECTION
432 + case ACLK_DATABASE_TIMER:
433 + if (unlikely(localhost && !wc->host)) {
434 + if (claimed()) {
435 + wc->host = rrdhost_find_by_guid(wc->host_guid, 0);
436 + if (wc->host) {
437 + info("HOST %s detected as active and claimed !!!", wc->host->hostname);
438 + snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", wc->host->hostname);
439 + uv_thread_set_name_np(wc->thread, threadname);
440 + wc->host->dbsync_worker = wc;
441 + aclk_del_worker_thread(wc);
442 + if (wc->host->node_id) {
443 + cmd.opcode = ACLK_DATABASE_NODE_INFO;
444 + cmd.completion = NULL;
445 + aclk_database_enq_cmd(wc, &cmd);
446 + }
447 + }
448 + }
449 + }
450 + break;
451 + case ACLK_DATABASE_SHUTDOWN:
452 + shutdown = 1;
453 + fatal_assert(0 == uv_timer_stop(&timer_req));
454 + uv_close((uv_handle_t *)&timer_req, NULL);
455 + break;
456 + default:
457 + debug(D_ACLK_SYNC, "%s: default.", __func__);
458 + break;
459 + }
460 + db_unlock();
461 + if (cmd.completion)
462 + aclk_complete(cmd.completion);
463 + } while (opcode != ACLK_DATABASE_NOOP);
464 + }
465 +
466 + /* cleanup operations of the event loop */
467 + info("Shutting down ACLK_DATABASE engine event loop.");
468 +
469 + /*
470 + * uv_async_send after uv_close does not seem to crash in linux at the moment,
471 + * it is however undocumented behaviour and we need to be aware if this becomes
472 + * an issue in the future.
473 + */
474 + uv_close((uv_handle_t *)&wc->async, NULL);
475 + uv_run(loop, UV_RUN_DEFAULT);
476 +
477 + info("Shutting down ACLK_DATABASE engine event loop complete.");
478 + /* TODO: don't let the API block by waiting to enqueue commands */
479 + uv_cond_destroy(&wc->cmd_cond);
480 +/* uv_mutex_destroy(&wc->cmd_mutex); */
481 + //fatal_assert(0 == uv_loop_close(loop));
482 + int rc;
483 +
484 + do {
485 + rc = uv_loop_close(loop);
486 + } while (rc != UV_EBUSY);
487 +
488 + freez(loop);
489 +
490 + rrd_wrlock();
491 + if (likely(wc->host))
492 + wc->host->dbsync_worker = NULL;
493 + freez(wc);
494 + rrd_unlock();
495 + return;
496 +
497 +error_after_timer_init:
498 + uv_close((uv_handle_t *)&wc->async, NULL);
499 +error_after_async_init:
500 + fatal_assert(0 == uv_loop_close(loop));
501 +error_after_loop_init:
502 + freez(loop);
503 +
504 + wc->error = UV_EAGAIN;
505 +}
506 +
507 +// -------------------------------------------------------------
508 +
509 +void aclk_set_architecture(int mode)
510 +{
511 + aclk_architecture = mode;
512 +}
513 +
514 +#define SELECT_HOST_DIMENSION_LIST "SELECT d.dim_id, c.update_every, c.type||'.'||c.id FROM chart c, dimension d, host h " \
515 + "WHERE d.chart_id = c.chart_id AND c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
516 +
517 +#define SELECT_HOST_CHART_LIST "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
518 + "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
519 +//
520 +//void sql_update_metric_statistics(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
521 +//{
522 +// UNUSED(cmd);
523 +//
524 +// int rc;
525 +//
526 +// char *claim_id = is_agent_claimed();
527 +// if (unlikely(!claim_id))
528 +// return;
529 +//
530 +// sqlite3_stmt *res = NULL;
531 +//
532 +// if (!wc->host || wc->host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
533 +// rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
534 +// else
535 +// rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
536 +//
537 +// if (unlikely(rc != SQLITE_OK)) {
538 +// error_report("Failed to prepare statement to fetch host dimensions");
539 +// return;
540 +// }
541 +//
542 +// if (wc->host)
543 +// rc = sqlite3_bind_blob(res, 1, &wc->host->host_uuid , sizeof(wc->host->host_uuid), SQLITE_STATIC);
544 +// else {
545 +// uuid_t host_uuid;
546 +// rc = uuid_parse(wc->host_guid, host_uuid);
547 +// if (unlikely(rc))
548 +// goto failed;
549 +// rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
550 +// }
551 +// if (unlikely(rc != SQLITE_OK)) {
552 +// error_report("Failed to bind host parameter to fetch host dimensions");
553 +// goto failed;
554 +// }
555 +//
556 +// time_t start_time = LONG_MAX;
557 +// time_t first_entry_t;
558 +// uint32_t update_every = 0;
559 +//
560 +// struct retention_updated rotate_data;
561 +//
562 +// memset(&rotate_data, 0, sizeof(rotate_data));
563 +//
564 +// int max_intervals = 32;
565 +//
566 +// rotate_data.interval_duration_count = 0;
567 +// rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
568 +//
569 +// now_realtime_timeval(&rotate_data.rotation_timestamp);
570 +// rotate_data.memory_mode = wc->host ? wc->host->rrd_memory_mode : RRD_MEMORY_MODE_DBENGINE;
571 +// rotate_data.claim_id = claim_id;
572 +// rotate_data.node_id = strdupz(wc->node_id);
573 +//
574 +// while (sqlite3_step(res) == SQLITE_ROW) {
575 +// if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) {
576 +// if (update_every) {
577 +// debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
578 +// rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
579 +// rotate_data.interval_duration_count++;
580 +// }
581 +// update_every = (uint32_t) sqlite3_column_int(res, 1);
582 +// rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
583 +// start_time = LONG_MAX;
584 +// }
585 +//#ifdef ENABLE_DBENGINE
586 +// time_t last_entry_t;
587 +// if (!wc->host || wc->host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
588 +// rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
589 +// else
590 +//#endif
591 +// {
592 +// RRDSET *st = NULL;
593 +// rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
594 +// if (!rc) {
595 +// first_entry_t = rrdset_first_entry_t(st);
596 +//// info("DEBUG: Scanning SET = %s --> %ld", st->name, first_entry_t);
597 +// }
598 +// }
599 +//
600 +// if (likely(!rc && first_entry_t))
601 +// start_time = MIN(start_time, first_entry_t);
602 +// }
603 +// if (update_every) {
604 +// debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
605 +// rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
606 +// rotate_data.interval_duration_count++;
607 +// }
608 +//
609 +// info("DEBUG: Scan update every for host");
610 +// for (int i = 0; i < rotate_data.interval_duration_count; ++i) {
611 +// info("DEBUG: %d --> Update %s for %u Retention = %u", i, wc->host_guid,
612 +// rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention);
613 +// };
614 +// aclk_retention_updated(&rotate_data);
615 +// freez(rotate_data.node_id);
616 +// freez(rotate_data.claim_id);
617 +// freez(rotate_data.interval_durations);
618 +//
619 +//failed:
620 +// rc = sqlite3_finalize(res);
621 +// if (unlikely(rc != SQLITE_OK))
622 +// error_report("Failed to finalize the prepared statement when reading host dimensions");
623 +// return;
624 +//}
625 +
626 +
627 +void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
628 +{
629 + char uuid_str[GUID_LEN + 1];
630 + char host_guid[GUID_LEN + 1];
631 +
632 + uuid_unparse_lower_fix(host_uuid, uuid_str);
633 +
634 + if (aclk_worker_thread_exists(uuid_str))
635 + return;
636 +
637 + uuid_unparse_lower(*host_uuid, host_guid);
638 +
639 + BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
640 +
641 + buffer_sprintf(sql, TABLE_ACLK_CHART, uuid_str);
642 + db_execute(buffer_tostring(sql));
643 + buffer_flush(sql);
644 +
645 + buffer_sprintf(sql, TABLE_ACLK_CHART_PAYLOAD, uuid_str);
646 + db_execute(buffer_tostring(sql));
647 + buffer_flush(sql);
648 +
649 + buffer_sprintf(sql, TABLE_ACLK_CHART_LATEST, uuid_str);
650 + db_execute(buffer_tostring(sql));
651 + buffer_flush(sql);
652 +
653 + buffer_sprintf(sql, INDEX_ACLK_CHART, uuid_str, uuid_str);
654 + db_execute(buffer_tostring(sql));
655 + buffer_flush(sql);
656 +
657 + buffer_sprintf(sql, INDEX_ACLK_CHART_LATEST, uuid_str, uuid_str);
658 + db_execute(buffer_tostring(sql));
659 + buffer_flush(sql);
660 +
661 + buffer_sprintf(sql, TRIGGER_ACLK_CHART_PAYLOAD, uuid_str, uuid_str, uuid_str);
662 + db_execute(buffer_tostring(sql));
663 + buffer_flush(sql);
664 +
665 + buffer_sprintf(sql, TABLE_ACLK_ALERT, uuid_str, uuid_str, uuid_str);
666 + db_execute(buffer_tostring(sql));
667 + buffer_flush(sql);
668 +
669 + buffer_sprintf(sql, INDEX_ACLK_ALERT, uuid_str, uuid_str);
670 + db_execute(buffer_tostring(sql));
671 +
672 + buffer_free(sql);
673 +
674 + if (likely(host) && unlikely(host->dbsync_worker))
675 + return;
676 +
677 + struct aclk_database_worker_config *wc = callocz(1, sizeof(struct aclk_database_worker_config));
678 + if (likely(host))
679 + host->dbsync_worker = (void *) wc;
680 + wc->host = host;
681 + wc->chart_updates = 0;
682 + wc->alert_updates = 0;
683 + wc->startup_time = now_realtime_sec();
684 + wc->cleanup_after = wc->startup_time + ACLK_DATABASE_CLEANUP_FIRST;
685 + strcpy(wc->uuid_str, uuid_str);
686 + strcpy(wc->host_guid, host_guid);
687 + if (node_id && !uuid_is_null(*node_id))
688 + uuid_unparse_lower(*node_id, wc->node_id);
689 + fatal_assert(0 == uv_thread_create(&(wc->thread), aclk_database_worker, wc));
690 +}
691 +
692 +void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
693 +{
694 + UNUSED(cmd);
695 +
696 + debug(D_ACLK, "Checking database for %s", wc->host_guid);
697 +
698 + BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
699 +
700 + buffer_sprintf(sql,"DELETE FROM aclk_chart_%s WHERE date_submitted IS NOT NULL AND "
701 + "date_updated < strftime('%%s','now','-%d seconds');", wc->uuid_str, ACLK_DELETE_ACK_INTERNAL);
702 + db_execute(buffer_tostring(sql));
703 + buffer_flush(sql);
704 +
705 + buffer_sprintf(sql,"DELETE FROM aclk_chart_payload_%s WHERE unique_id NOT IN "
706 + "(SELECT unique_id FROM aclk_chart_%s) AND unique_id NOT IN (SELECT unique_id FROM aclk_chart_latest_%s);",
707 + wc->uuid_str, wc->uuid_str, wc->uuid_str);
708 + db_execute(buffer_tostring(sql));
709 +
710 + buffer_free(sql);
711 + return;
712 +}
713 +
714 +#define SQL_SELECT_HOST_BY_UUID "SELECT host_id FROM host WHERE host_id = @host_id;"
715 +
716 +static int is_host_available(uuid_t *host_id)
717 +{
718 + sqlite3_stmt *res = NULL;
719 + int rc;
720 +
721 + if (unlikely(!db_meta)) {
722 + if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
723 + error_report("Database has not been initialized");
724 + return 1;
725 + }
726 +
727 + rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_UUID, -1, &res, 0);
728 + if (unlikely(rc != SQLITE_OK)) {
729 + error_report("Failed to prepare statement to select node instance information for a node");
730 + return 1;
731 + }
732 +
733 + rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
734 + if (unlikely(rc != SQLITE_OK)) {
735 + error_report("Failed to bind host_id parameter to select node instance information");
736 + goto failed;
737 + }
738 + rc = sqlite3_step(res);
739 +
740 + failed:
741 + if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
742 + error_report("Failed to finalize the prepared statement when checking host existence");
743 +
744 + return (rc == SQLITE_ROW);
745 +}
746 +
747 +// OPCODE: ACLK_DATABASE_DELETE_HOST
748 +void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
749 +{
750 + UNUSED(wc);
751 + char uuid_str[GUID_LEN + 1];
752 + char host_str[GUID_LEN + 1];
753 +
754 + int rc;
755 + uuid_t host_uuid;
756 + char *host_guid = (char *)cmd.data;
757 +
758 + if (unlikely(!host_guid))
759 + return;
760 +
761 + rc = uuid_parse(host_guid, host_uuid);
762 + freez(host_guid);
763 + if (rc)
764 + return;
765 +
766 + uuid_unparse_lower(host_uuid, host_str);
767 + uuid_unparse_lower_fix(&host_uuid, uuid_str);
768 +
769 + debug(D_ACLK_SYNC, "Checking if I should delete aclk tables for node %s", host_str);
770 +
771 + if (is_host_available(&host_uuid)) {
772 + debug(D_ACLK_SYNC, "Host %s exists, not deleting aclk sync tables", host_str);
773 + return;
774 + }
775 +
776 + debug(D_ACLK_SYNC, "Host %s does NOT exist, can delete aclk sync tables", host_str);
777 +
778 + sqlite3_stmt *res = NULL;
779 + BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
780 +
781 + buffer_sprintf(sql,"SELECT 'drop '||type||' IF EXISTS '||name||';' FROM sqlite_schema " \
782 + "WHERE name LIKE 'aclk_%%_%s' AND type IN ('table', 'trigger', 'index');", uuid_str);
783 +
784 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
785 + if (rc != SQLITE_OK) {
786 + error_report("Failed to prepare statement to clean up aclk tables");
787 + goto fail;
788 + }
789 + buffer_flush(sql);
790 +
791 + while (sqlite3_step(res) == SQLITE_ROW)
792 + buffer_strcat(sql, (char *) sqlite3_column_text(res, 0));
793 +
794 + rc = sqlite3_finalize(res);
795 + if (unlikely(rc != SQLITE_OK))
796 + error_report("Failed to finalize statement to clean up aclk tables, rc = %d", rc);
797 +
798 + db_execute(buffer_tostring(sql));
799 +
800 +fail:
801 + buffer_free(sql);
802 + return;
803 +}
804 +
805 +static int sql_check_aclk_table(void *data, int argc, char **argv, char **column)
806 +{
807 + struct aclk_database_worker_config *wc = data;
808 + UNUSED(argc);
809 + UNUSED(column);
810 +
811 + debug(D_ACLK_SYNC,"Scheduling aclk sync table check for node %s", (char *) argv[0]);
812 + struct aclk_database_cmd cmd;
813 + cmd.opcode = ACLK_DATABASE_DELETE_HOST;
814 + cmd.completion = NULL;
815 + cmd.data = strdupz((char *) argv[0]);
816 + aclk_database_enq_cmd_noblock(wc, &cmd);
817 + return 0;
818 +}
819 +
820 +#define SQL_SELECT_ACLK_ACTIVE_LIST "SELECT REPLACE(SUBSTR(name,19),'_','-') FROM sqlite_schema " \
821 + "WHERE name LIKE 'aclk_chart_latest_%' AND type IN ('table');"
822 +
823 +void sql_check_aclk_table_list(struct aclk_database_worker_config *wc)
824 +{
825 + debug(D_ACLK_SYNC,"Cleaning tables for nodes that do not exist");
826 + (int) sqlite3_exec(db_meta, SQL_SELECT_ACLK_ACTIVE_LIST, sql_check_aclk_table, (void *) wc, NULL);
827 + return;
828 +}
829 +
830 +void aclk_data_rotated(RRDHOST *host)
831 +{
832 + UNUSED(host);
833 +
834 + debug(D_ACLK_SYNC,"Processing data base rotation event");
835 + struct aclk_database_cmd cmd;
836 + cmd.opcode = ACLK_DATABASE_UPD_STATS;
837 + cmd.data = NULL;
838 + cmd.count = 0;
839 + cmd.completion = NULL;
840 +
841 + rrd_wrlock();
842 + RRDHOST *this_host = localhost;
843 + while (this_host) {
844 + aclk_database_enq_cmd((struct aclk_database_worker_config *)this_host->dbsync_worker, &cmd);
845 + this_host = this_host->next;
846 + }
847 + rrd_unlock();
848 +
849 + struct aclk_database_worker_config *tmp = aclk_thread_head;
850 +
851 + uv_mutex_lock(&aclk_async_lock);
852 + while (tmp) {
853 + aclk_database_enq_cmd(tmp, &cmd);
854 + tmp = tmp->next;
855 + }
856 + uv_mutex_unlock(&aclk_async_lock);
857 + return;
858 +}
\ No newline at end of file
database/sqlite/sqlite_aclk.h new
+225
@@ -0,0 +1,225 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef NETDATA_SQLITE_ACLK_H
4 +#define NETDATA_SQLITE_ACLK_H
5 +
6 +#include "sqlite3.h"
7 +
8 +// TODO: To be added
9 +//#include "../../aclk/schema-wrappers/chart_stream.h"
10 +
11 +#ifndef ACLK_MAX_CHART_BATCH
12 +#define ACLK_MAX_CHART_BATCH (20)
13 +#endif
14 +#ifndef ACLK_MAX_CHART_BATCH_COUNT
15 +#define ACLK_MAX_CHART_BATCH_COUNT (5)
16 +#endif
17 +#define ACLK_MAX_ALERT_UPDATES (5)
18 +#define ACLK_SYNC_RETRY_COUNT "10"
19 +#define ACLK_DATABASE_CLEANUP_FIRST (60)
20 +#define ACLK_DATABASE_CLEANUP_INTERVAL (3600)
21 +#define ACLK_DELETE_ACK_INTERNAL (600)
22 +#define ACLK_SYNC_QUERY_SIZE 512
23 +
24 +struct aclk_completion {
25 + uv_mutex_t mutex;
26 + uv_cond_t cond;
27 + volatile unsigned completed;
28 +};
29 +
30 +static inline void init_aclk_completion(struct aclk_completion *p)
31 +{
32 + p->completed = 0;
33 + fatal_assert(0 == uv_cond_init(&p->cond));
34 + fatal_assert(0 == uv_mutex_init(&p->mutex));
35 +}
36 +
37 +static inline void destroy_aclk_completion(struct aclk_completion *p)
38 +{
39 + uv_cond_destroy(&p->cond);
40 + uv_mutex_destroy(&p->mutex);
41 +}
42 +
43 +static inline void wait_for_aclk_completion(struct aclk_completion *p)
44 +{
45 + uv_mutex_lock(&p->mutex);
46 + while (0 == p->completed) {
47 + uv_cond_wait(&p->cond, &p->mutex);
48 + }
49 + fatal_assert(1 == p->completed);
50 + uv_mutex_unlock(&p->mutex);
51 +}
52 +
53 +static inline void aclk_complete(struct aclk_completion *p)
54 +{
55 + uv_mutex_lock(&p->mutex);
56 + p->completed = 1;
57 + uv_mutex_unlock(&p->mutex);
58 + uv_cond_broadcast(&p->cond);
59 +}
60 +
61 +extern uv_mutex_t aclk_async_lock;
62 +
63 +extern int aclk_architecture;
64 +
65 +static inline void uuid_unparse_lower_fix(uuid_t *uuid, char *out)
66 +{
67 + uuid_unparse_lower(*uuid, out);
68 + out[8] = '_';
69 + out[13] = '_';
70 + out[18] = '_';
71 + out[23] = '_';
72 +}
73 +
74 +static inline char *get_str_from_uuid(uuid_t *uuid)
75 +{
76 + char uuid_str[GUID_LEN + 1];
77 + if (unlikely(!uuid)) {
78 + uuid_t zero_uuid;
79 + uuid_clear(zero_uuid);
80 + uuid_unparse_lower(zero_uuid, uuid_str);
81 + }
82 + else
83 + uuid_unparse_lower(*uuid, uuid_str);
84 + return strdupz(uuid_str);
85 +}
86 +
87 +#define TABLE_ACLK_CHART "CREATE TABLE IF NOT EXISTS aclk_chart_%s (sequence_id INTEGER PRIMARY KEY AUTOINCREMENT, " \
88 + "date_created, date_updated, date_submitted, status, uuid, type, unique_id, " \
89 + "update_count default 1, unique(uuid, status));"
90 +
91 +#define TABLE_ACLK_CHART_PAYLOAD "CREATE TABLE IF NOT EXISTS aclk_chart_payload_%s (unique_id BLOB PRIMARY KEY, " \
92 + "uuid, claim_id, type, date_created, payload);"
93 +
94 +#define TABLE_ACLK_CHART_LATEST "CREATE TABLE IF NOT EXISTS aclk_chart_latest_%s (uuid BLOB PRIMARY KEY, " \
95 + "unique_id, date_submitted);"
96 +
97 +#define TRIGGER_ACLK_CHART_PAYLOAD "CREATE TRIGGER IF NOT EXISTS aclk_tr_chart_payload_%s " \
98 + "after insert on aclk_chart_payload_%s " \
99 + "begin insert into aclk_chart_%s (uuid, unique_id, type, status, date_created) values " \
100 + " (new.uuid, new.unique_id, new.type, 'pending', strftime('%%s')) on conflict(uuid, status) " \
101 + " do update set unique_id = new.unique_id, update_count = update_count + 1; " \
102 + "end;"
103 +
104 +#define TABLE_ACLK_ALERT "CREATE TABLE IF NOT EXISTS aclk_alert_%s (sequence_id INTEGER PRIMARY KEY, " \
105 + "alert_unique_id, date_created, date_submitted, " \
106 + "unique(alert_unique_id)); " \
107 + "insert into aclk_alert_%s (alert_unique_id, date_created) " \
108 + "select unique_id alert_unique_id, strftime('%%s') date_created from health_log_%s where new_status <> 0 and new_status <> -2 order by unique_id asc on conflict (alert_unique_id) do nothing;"
109 +
110 +#define INDEX_ACLK_CHART "CREATE INDEX IF NOT EXISTS aclk_chart_index_%s ON aclk_chart_%s (unique_id);"
111 +
112 +#define INDEX_ACLK_CHART_LATEST "CREATE INDEX IF NOT EXISTS aclk_chart_latest_index_%s ON aclk_chart_latest_%s (unique_id);"
113 +
114 +#define INDEX_ACLK_ALERT "CREATE INDEX IF NOT EXISTS aclk_alert_index_%s ON aclk_alert_%s (alert_unique_id);"
115 +
116 +enum aclk_database_opcode {
117 + ACLK_DATABASE_NOOP = 0,
118 + ACLK_DATABASE_ADD_ALERT,
119 + ACLK_DATABASE_ADD_CHART,
120 + ACLK_DATABASE_ADD_DIMENSION,
121 + ACLK_DATABASE_ALARM_HEALTH_LOG,
122 + ACLK_DATABASE_CHART_ACK,
123 + ACLK_DATABASE_CHECK,
124 + ACLK_DATABASE_CHECK_ROTATION,
125 + ACLK_DATABASE_CLEANUP,
126 + ACLK_DATABASE_DEDUP_CHART,
127 + ACLK_DATABASE_DELETE_HOST,
128 + ACLK_DATABASE_NODE_INFO,
129 + ACLK_DATABASE_PUSH_ALERT,
130 + ACLK_DATABASE_PUSH_ALERT_CONFIG,
131 + ACLK_DATABASE_PUSH_CHART,
132 + ACLK_DATABASE_PUSH_CHART_CONFIG,
133 + ACLK_DATABASE_RESET_CHART,
134 + ACLK_DATABASE_RESET_NODE,
135 + ACLK_DATABASE_SHUTDOWN,
136 + ACLK_DATABASE_STATUS_CHART,
137 + ACLK_DATABASE_SYNC_CHART_SEQ,
138 + ACLK_DATABASE_TIMER,
139 + ACLK_DATABASE_UPD_STATS,
140 + ACLK_DATABASE_MAX_OPCODE
141 +};
142 +
143 +struct aclk_chart_payload_t {
144 + long sequence_id;
145 + long last_sequence_id;
146 + char *payload;
147 + struct aclk_chart_payload_t *next;
148 +};
149 +
150 +
151 +struct aclk_database_cmd {
152 + enum aclk_database_opcode opcode;
153 + void *data;
154 + void *data_param;
155 + int count;
156 + uint64_t param1;
157 + struct aclk_completion *completion;
158 +};
159 +
160 +#define ACLK_DATABASE_CMD_Q_MAX_SIZE (2048)
161 +
162 +struct aclk_database_cmdqueue {
163 + unsigned head, tail;
164 + struct aclk_database_cmd cmd_array[ACLK_DATABASE_CMD_Q_MAX_SIZE];
165 +};
166 +
167 +struct aclk_database_worker_config {
168 + uv_thread_t thread;
169 + char uuid_str[GUID_LEN + 1];
170 + char node_id[GUID_LEN + 1];
171 + char host_guid[GUID_LEN + 1];
172 + uint64_t chart_sequence_id; // last chart_sequence_id
173 + time_t chart_timestamp; // last chart timestamp
174 + time_t cleanup_after; // Start a cleanup after this timestamp
175 + time_t startup_time; // When the sync thread started
176 + uint64_t batch_id; // batch id to use
177 + uint64_t alerts_batch_id; // batch id for alerts to use
178 + uint64_t alerts_start_seq_id; // cloud has asked to start streaming from
179 + uint64_t alert_sequence_id; // last alert sequence_id
180 + uv_loop_t *loop;
181 + RRDHOST *host;
182 + uv_async_t async;
183 + /* FIFO command queue */
184 + uv_mutex_t cmd_mutex;
185 + uv_cond_t cmd_cond;
186 + volatile unsigned queue_size;
187 + struct aclk_database_cmdqueue cmd_queue;
188 + int error;
189 + int chart_updates;
190 + int alert_updates;
191 + time_t batch_created;
192 + struct aclk_database_worker_config *next;
193 +};
194 +
195 +static inline RRDHOST *find_host_by_node_id(char *node_id)
196 +{
197 + uuid_t node_uuid;
198 + if (unlikely(!node_id))
199 + return NULL;
200 +
201 + uuid_parse(node_id, node_uuid);
202 +
203 + RRDHOST *host = localhost;
204 + while(host) {
205 + if (host->node_id && !(uuid_compare(*host->node_id, node_uuid)))
206 + return host;
207 + host = host->next;
208 + }
209 + return NULL;
210 +}
211 +
212 +
213 +extern sqlite3 *db_meta;
214 +
215 +extern void aclk_database_enq_cmd(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd);
216 +extern int aclk_database_enq_cmd_noblock(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd);
217 +extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
218 +extern void aclk_set_architecture(int mode);
219 +void sql_aclk_sync_init(void);
220 +void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
221 +void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
222 +void sql_drop_host_aclk_table_list(uuid_t *host_uuid);
223 +void sql_check_aclk_table_list(struct aclk_database_worker_config *wc);
224 +void aclk_data_rotated(RRDHOST *host);
225 +#endif //NETDATA_SQLITE_ACLK_H
database/sqlite/sqlite_functions.c
+18 -7
@@ -977,16 +977,27 @@ failed:
977 return host;
978 }
979
980 -void db_execute(char *cmd)
980 +#define SQL_MAX_RETRY 100
981 +
982 +void db_execute(const char *cmd)
983 {
984 int rc;
983 - char *err_msg;
984 - rc = sqlite3_exec(db_meta, cmd, 0, 0, &err_msg);
985 - if (rc != SQLITE_OK) {
986 - error_report("Failed to execute '%s', rc = %d (%s)", cmd, rc, err_msg);
987 - sqlite3_free(err_msg);
985 + int cnt = 0;
986 + while (cnt < SQL_MAX_RETRY) {
987 + char *err_msg;
988 + rc = sqlite3_exec(db_meta, cmd, 0, 0, &err_msg);
989 + if (rc != SQLITE_OK) {
990 + error_report("Failed to execute '%s', rc = %d (%s) -- attempt %d", cmd, rc, err_msg, cnt);
991 + sqlite3_free(err_msg);
992 + if (likely(rc == SQLITE_BUSY || rc == SQLITE_LOCKED)) {
993 + usleep(SQLITE_INSERT_DELAY * USEC_PER_MS);
994 + }
995 + else break;
996 + }
997 + else
998 + break;
999 + ++cnt;
1000 }
989 -
1001 return;
1002 }
1003
database/sqlite/sqlite_functions.h
+1 -1
@@ -63,7 +63,7 @@ extern int find_uuid_type(uuid_t *uuid);
63 extern void sql_rrdset2json(RRDHOST *host, BUFFER *wb);
64
65 extern RRDHOST *sql_create_host_by_uuid(char *guid);
66 -extern void db_execute(char *cmd);
66 +extern void db_execute(const char *cmd);
67 extern int file_is_migrated(char *path);
68 extern void add_migrated_file(char *path, uint64_t file_size);
69 extern void db_unlock(void);
libnetdata/log/log.h
+1 -1
@@ -44,7 +44,7 @@ extern "C" {
44 #define D_RRDENGINE 0x0000000100000000
45 #define D_ACLK 0x0000000200000000
46 #define D_METADATALOG 0x0000000400000000
47 -#define D_GUIDLOG 0x0000000800000000
47 +#define D_ACLK_SYNC 0x0000000800000000
48 #define D_SYSTEM 0x8000000000000000
49
50 //#define DEBUG (D_WEB_CLIENT_ACCESS|D_LISTENER|D_RRD_STATS)