@cryptotaxi247 / netdata-1 / commits / 2085a518c

Add chart message support for ACLK new architecture (#11447)

Stelios Fragkakis committed Sep 21, 2021 at 22:37 UTC 2085a518c3fa415b9e4d08002c720465417c7c14
10 files changed +883 -165
CMakeLists.txt
+2
@@ -635,6 +635,8 @@ set(RRD_PLUGIN_FILES
635 database/sqlite/sqlite_health.h
636 database/sqlite/sqlite_aclk_node.c
637 database/sqlite/sqlite_aclk_node.h
638 + database/sqlite/sqlite_aclk_chart.c
639 + database/sqlite/sqlite_aclk_chart.h
640 database/sqlite/sqlite3.c
641 database/sqlite/sqlite3.h
642 database/engine/rrdengine.c
Makefile.am
+2
@@ -415,6 +415,8 @@ RRD_PLUGIN_FILES = \
415 database/sqlite/sqlite_health.h \
416 database/sqlite/sqlite_aclk_node.c \
417 database/sqlite/sqlite_aclk_node.h \
418 + database/sqlite/sqlite_aclk_chart.c \
419 + database/sqlite/sqlite_aclk_chart.h \
420 database/sqlite/sqlite3.c \
421 database/sqlite/sqlite3.h \
422 $(NULL)
database/rrd.h
+6 -1
@@ -163,7 +163,8 @@ typedef enum rrddim_flags {
163 RRDDIM_FLAG_OBSOLETE = (1 << 2), // this is marked by the collector/module as obsolete
164 // No new values have been collected for this dimension since agent start or it was marked RRDDIM_FLAG_OBSOLETE at
165 // least rrdset_free_obsolete_time seconds ago.
166 - RRDDIM_FLAG_ARCHIVED = (1 << 3)
166 + RRDDIM_FLAG_ARCHIVED = (1 << 3),
167 + RRDDIM_FLAG_ACLK = (1 << 4)
168 } RRDDIM_FLAGS;
169
170 #ifdef HAVE_C___ATOMIC
@@ -379,6 +380,9 @@ struct rrddim_volatile {
380 #ifdef ENABLE_DBENGINE
381 uuid_t *rrdeng_uuid; // database engine metric UUID
382 struct pg_cache_page_index *page_index;
383 +#endif
384 +#ifdef ENABLE_ACLK
385 + int aclk_live_status;
386 #endif
387 uuid_t metric_uuid; // global UUID for this metric (unique_across hosts)
388 union rrddim_collect_handle handle;
@@ -423,6 +427,7 @@ struct rrddim_volatile {
427 struct rrdset_volatile {
428 char *old_title;
429 char *old_context;
430 + uuid_t hash_id;
431 struct label *new_labels;
432 struct label_index labels;
433 };
database/rrdhost.c
+5
@@ -1443,6 +1443,11 @@ restart_after_removal:
1443 continue;
1444 }
1445
1446 + if (rrddim_flag_check(rd, RRDDIM_FLAG_ACLK)) {
1447 + last = rd;
1448 + rd = rd->next;
1449 + continue;
1450 + }
1451 rrddim_flag_set(rd, RRDDIM_FLAG_ARCHIVED);
1452 while (rd->variables)
1453 rrddimvar_free(rd->variables);
database/rrdset.c
+2 -1
@@ -1861,7 +1861,8 @@ after_second_database_work:
1861 rrdset_wrlock(st);
1862
1863 for( rd = st->dimensions, last = NULL ; likely(rd) ; ) {
1864 - if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE) && (rd->last_collected_time.tv_sec + rrdset_free_obsolete_time < now))) {
1864 + if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE) && !rrddim_flag_check(rd, RRDDIM_FLAG_ACLK)
1865 + && (rd->last_collected_time.tv_sec + rrdset_free_obsolete_time < now))) {
1866 info("Removing obsolete dimension '%s' (%s) of '%s' (%s).", rd->name, rd->id, st->name, st->id);
1867
1868 if(likely(rd->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || rd->rrd_memory_mode == RRD_MEMORY_MODE_MAP)) {
database/sqlite/sqlite_aclk.c
+48 -150
@@ -4,7 +4,7 @@
4 #include "sqlite_aclk.h"
5
6 // TODO: To be added
7 -//#include "sqlite_aclk_chart.h"
7 +#include "sqlite_aclk_chart.h"
8 //#include "sqlite_aclk_alert.h"
9 #include "sqlite_aclk_node.h"
10
@@ -170,6 +170,25 @@ struct aclk_database_cmd aclk_database_deq_cmd(struct aclk_database_worker_confi
170 return ret;
171 }
172
173 +int aclk_worker_enq_cmd(char *node_id, struct aclk_database_cmd *cmd)
174 +{
175 + if (unlikely(!node_id || !cmd))
176 + return 0;
177 +
178 + uv_mutex_lock(&aclk_async_lock);
179 + struct aclk_database_worker_config *wc = aclk_thread_head;
180 +
181 + while (wc) {
182 + if (!strcmp(wc->node_id, node_id))
183 + break;
184 + wc = wc->next;
185 + }
186 + if (wc)
187 + aclk_database_enq_cmd(wc, cmd);
188 + uv_mutex_unlock(&aclk_async_lock);
189 + return (wc == NULL);
190 +}
191 +
192 int aclk_start_sync_thread(void *data, int argc, char **argv, char **column)
193 {
194 char uuid_str[GUID_LEN + 1];
@@ -247,11 +266,13 @@ static void timer_cb(uv_timer_t* handle)
266 wc->cleanup_after += ACLK_DATABASE_CLEANUP_INTERVAL;
267 }
268
250 - if (wc->chart_updates) {
269 + if (wc->chart_updates && !wc->chart_pending) {
270 cmd.opcode = ACLK_DATABASE_PUSH_CHART;
271 cmd.count = ACLK_MAX_CHART_BATCH;
272 + cmd.completion = NULL;
273 cmd.param1 = ACLK_MAX_CHART_BATCH_COUNT;
254 - aclk_database_enq_cmd_noblock(wc, &cmd);
274 + if (!aclk_database_enq_cmd_noblock(wc, &cmd))
275 + wc->chart_pending = 1;
276 }
277
278 if (wc->alert_updates) {
@@ -310,9 +331,9 @@ void aclk_database_worker(void *arg)
331 wc->error = 0;
332 shutdown = 0;
333
334 + wc->node_info_send = (wc->host && !localhost);
335 aclk_add_worker_thread(wc);
314 -
315 - info("Starting ACLK sync event loop for host with GUID %s (Host is '%s')", wc->host_guid, wc->host ? "connected" : "not connected");
336 + info("Starting ACLK sync thread for host %s -- scratch area %lu bytes", wc->host_guid, sizeof(*wc));
337 // TODO: To be added
338 // sql_get_last_chart_sequence(wc, cmd);
339 while (likely(shutdown == 0)) {
@@ -324,12 +345,14 @@ void aclk_database_worker(void *arg)
345 /* wait for commands */
346 cmd_batch_size = 0;
347 do {
327 - if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE))
348 + if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE)) {
349 + info("DEBUG: %s Processed %u commands, current queue about %u",
350 + wc->uuid_str, cmd_batch_size, wc->queue_size);
351 break;
352 + }
353 cmd = aclk_database_deq_cmd(wc);
354 opcode = cmd.opcode;
355 ++cmd_batch_size;
332 - db_lock();
356 switch (opcode) {
357 case ACLK_DATABASE_NOOP:
358 /* the command queue was empty, do nothing */
@@ -356,41 +379,29 @@ void aclk_database_worker(void *arg)
379 break;
380
381 // CHART / DIMENSION OPERATIONS
382 + case ACLK_DATABASE_ADD_CHART:
383 + debug(D_ACLK_SYNC, "Adding chart event for %s", wc->host_guid);
384 + aclk_add_chart_event(wc, cmd);
385 + break;
386 + case ACLK_DATABASE_ADD_DIMENSION:
387 + debug(D_ACLK_SYNC, "Adding dimension event for %s", wc->host_guid);
388 + aclk_add_dimension_event(wc, cmd);
389 + break;
390 case ACLK_DATABASE_PUSH_CHART:
391 debug(D_ACLK_SYNC, "Pushing chart info to the cloud for node %s", wc->host_guid);
361 -// aclk_push_chart_event(wc, cmd);
392 + aclk_send_chart_event(wc, cmd);
393 break;
394 case ACLK_DATABASE_PUSH_CHART_CONFIG:
395 debug(D_ACLK_SYNC, "Pushing chart config info to the cloud for node %s", wc->host_guid);
365 -// aclk_push_chart_config(wc, cmd);
396 + aclk_send_chart_config(wc, cmd);
397 break;
398 case ACLK_DATABASE_CHART_ACK:
399 debug(D_ACLK_SYNC, "ACK chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1);
369 -// sql_set_chart_ack(wc, cmd);
400 + aclk_receive_chart_ack(wc, cmd);
401 break;
402 case ACLK_DATABASE_RESET_CHART:
403 debug(D_ACLK_SYNC, "RESET chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1);
373 -// sql_reset_chart_event(wc, cmd);
374 - break;
375 - case ACLK_DATABASE_STATUS_CHART:
376 - debug(D_ACLK_SYNC,"Requesting chart status for %s", wc->host_guid);
377 -// aclk_status_chart_event(wc, cmd);
378 - break;
379 - case ACLK_DATABASE_ADD_CHART:
380 - debug(D_ACLK_SYNC,"Adding chart event for %s", wc->host_guid);
381 -// aclk_add_chart_event(wc, cmd);
382 - break;
383 - case ACLK_DATABASE_ADD_DIMENSION:
384 - debug(D_ACLK_SYNC,"Adding dimension event for %s", wc->host_guid);
385 -// aclk_add_dimension_event(wc, cmd);
386 - break;
387 - case ACLK_DATABASE_DEDUP_CHART:
388 - debug(D_ACLK_SYNC,"Running chart deduplication for %s", wc->host_guid);
389 -// sql_chart_deduplicate(wc, cmd);
390 - break;
391 - case ACLK_DATABASE_SYNC_CHART_SEQ:
392 - debug(D_ACLK_SYNC,"Calculatting chart sequence for %s", wc->host_guid);
393 -// sql_get_last_chart_sequence(wc, cmd);
404 + aclk_receive_chart_reset(wc, cmd);
405 break;
406
407 // ALERTS
@@ -431,14 +442,15 @@ void aclk_database_worker(void *arg)
442 uv_thread_set_name_np(wc->thread, threadname);
443 wc->host->dbsync_worker = wc;
444 aclk_del_worker_thread(wc);
434 - if (wc->host->node_id) {
435 - cmd.opcode = ACLK_DATABASE_NODE_INFO;
436 - cmd.completion = NULL;
437 - aclk_database_enq_cmd(wc, &cmd);
438 - }
445 + wc->node_info_send = 1;
446 }
447 }
448 }
449 + if (wc->node_info_send && wc->host && localhost && claimed()) {
450 + cmd.opcode = ACLK_DATABASE_NODE_INFO;
451 + cmd.completion = NULL;
452 + wc->node_info_send = aclk_database_enq_cmd_noblock(wc, &cmd);
453 + }
454 break;
455 case ACLK_DATABASE_SHUTDOWN:
456 shutdown = 1;
@@ -449,7 +461,6 @@ void aclk_database_worker(void *arg)
461 debug(D_ACLK_SYNC, "%s: default.", __func__);
462 break;
463 }
452 - db_unlock();
464 if (cmd.completion)
465 aclk_complete(cmd.completion);
466 } while (opcode != ACLK_DATABASE_NOOP);
@@ -503,119 +514,6 @@ void aclk_set_architecture(int mode)
514 aclk_architecture = mode;
515 }
516
506 -#define SELECT_HOST_DIMENSION_LIST "SELECT d.dim_id, c.update_every, c.type||'.'||c.id FROM chart c, dimension d, host h " \
507 - "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;"
508 -
509 -#define SELECT_HOST_CHART_LIST "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
510 - "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
511 -//
512 -//void sql_update_metric_statistics(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
513 -//{
514 -// UNUSED(cmd);
515 -//
516 -// int rc;
517 -//
518 -// char *claim_id = is_agent_claimed();
519 -// if (unlikely(!claim_id))
520 -// return;
521 -//
522 -// sqlite3_stmt *res = NULL;
523 -//
524 -// if (!wc->host || wc->host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
525 -// rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
526 -// else
527 -// rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
528 -//
529 -// if (unlikely(rc != SQLITE_OK)) {
530 -// error_report("Failed to prepare statement to fetch host dimensions");
531 -// return;
532 -// }
533 -//
534 -// if (wc->host)
535 -// rc = sqlite3_bind_blob(res, 1, &wc->host->host_uuid , sizeof(wc->host->host_uuid), SQLITE_STATIC);
536 -// else {
537 -// uuid_t host_uuid;
538 -// rc = uuid_parse(wc->host_guid, host_uuid);
539 -// if (unlikely(rc))
540 -// goto failed;
541 -// rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
542 -// }
543 -// if (unlikely(rc != SQLITE_OK)) {
544 -// error_report("Failed to bind host parameter to fetch host dimensions");
545 -// goto failed;
546 -// }
547 -//
548 -// time_t start_time = LONG_MAX;
549 -// time_t first_entry_t;
550 -// uint32_t update_every = 0;
551 -//
552 -// struct retention_updated rotate_data;
553 -//
554 -// memset(&rotate_data, 0, sizeof(rotate_data));
555 -//
556 -// int max_intervals = 32;
557 -//
558 -// rotate_data.interval_duration_count = 0;
559 -// rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
560 -//
561 -// now_realtime_timeval(&rotate_data.rotation_timestamp);
562 -// rotate_data.memory_mode = wc->host ? wc->host->rrd_memory_mode : RRD_MEMORY_MODE_DBENGINE;
563 -// rotate_data.claim_id = claim_id;
564 -// rotate_data.node_id = strdupz(wc->node_id);
565 -//
566 -// while (sqlite3_step(res) == SQLITE_ROW) {
567 -// if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) {
568 -// if (update_every) {
569 -// debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
570 -// rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
571 -// rotate_data.interval_duration_count++;
572 -// }
573 -// update_every = (uint32_t) sqlite3_column_int(res, 1);
574 -// rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
575 -// start_time = LONG_MAX;
576 -// }
577 -//#ifdef ENABLE_DBENGINE
578 -// time_t last_entry_t;
579 -// if (!wc->host || wc->host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
580 -// rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
581 -// else
582 -//#endif
583 -// {
584 -// RRDSET *st = NULL;
585 -// rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
586 -// if (!rc) {
587 -// first_entry_t = rrdset_first_entry_t(st);
588 -//// info("DEBUG: Scanning SET = %s --> %ld", st->name, first_entry_t);
589 -// }
590 -// }
591 -//
592 -// if (likely(!rc && first_entry_t))
593 -// start_time = MIN(start_time, first_entry_t);
594 -// }
595 -// if (update_every) {
596 -// debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
597 -// rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
598 -// rotate_data.interval_duration_count++;
599 -// }
600 -//
601 -// info("DEBUG: Scan update every for host");
602 -// for (int i = 0; i < rotate_data.interval_duration_count; ++i) {
603 -// info("DEBUG: %d --> Update %s for %u Retention = %u", i, wc->host_guid,
604 -// rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention);
605 -// };
606 -// aclk_retention_updated(&rotate_data);
607 -// freez(rotate_data.node_id);
608 -// freez(rotate_data.claim_id);
609 -// freez(rotate_data.interval_durations);
610 -//
611 -//failed:
612 -// rc = sqlite3_finalize(res);
613 -// if (unlikely(rc != SQLITE_OK))
614 -// error_report("Failed to finalize the prepared statement when reading host dimensions");
615 -// return;
616 -//}
617 -
618 -
517 void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
518 {
519 char uuid_str[GUID_LEN + 1];
database/sqlite/sqlite_aclk.h
+14 -13
@@ -6,13 +6,13 @@
6 #include "sqlite3.h"
7
8 // TODO: To be added
9 -//#include "../../aclk/schema-wrappers/chart_stream.h"
9 +#include "../../aclk/schema-wrappers/chart_stream.h"
10
11 #ifndef ACLK_MAX_CHART_BATCH
12 -#define ACLK_MAX_CHART_BATCH (20)
12 +#define ACLK_MAX_CHART_BATCH (200)
13 #endif
14 #ifndef ACLK_MAX_CHART_BATCH_COUNT
15 -#define ACLK_MAX_CHART_BATCH_COUNT (5)
15 +#define ACLK_MAX_CHART_BATCH_COUNT (10)
16 #endif
17 #define ACLK_MAX_ALERT_UPDATES (5)
18 #define ACLK_SYNC_RETRY_COUNT "10"
@@ -84,7 +84,7 @@ static inline char *get_str_from_uuid(uuid_t *uuid)
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, " \
87 +#define TABLE_ACLK_CHART "CREATE TABLE IF NOT EXISTS aclk_chart_%s (sequence_id INTEGER PRIMARY KEY, " \
88 "date_created, date_updated, date_submitted, status, uuid, type, unique_id, " \
89 "update_count default 1, unique(uuid, status));"
90
@@ -123,7 +123,6 @@ enum aclk_database_opcode {
123 ACLK_DATABASE_CHECK,
124 ACLK_DATABASE_CHECK_ROTATION,
125 ACLK_DATABASE_CLEANUP,
126 - ACLK_DATABASE_DEDUP_CHART,
126 ACLK_DATABASE_DELETE_HOST,
127 ACLK_DATABASE_NODE_INFO,
128 ACLK_DATABASE_PUSH_ALERT,
@@ -133,8 +132,6 @@ enum aclk_database_opcode {
132 ACLK_DATABASE_RESET_CHART,
133 ACLK_DATABASE_RESET_NODE,
134 ACLK_DATABASE_SHUTDOWN,
136 - ACLK_DATABASE_STATUS_CHART,
137 - ACLK_DATABASE_SYNC_CHART_SEQ,
135 ACLK_DATABASE_TIMER,
136 ACLK_DATABASE_UPD_STATS,
137 ACLK_DATABASE_MAX_OPCODE
@@ -157,7 +154,7 @@ struct aclk_database_cmd {
154 struct aclk_completion *completion;
155 };
156
160 -#define ACLK_DATABASE_CMD_Q_MAX_SIZE (2048)
157 +#define ACLK_DATABASE_CMD_Q_MAX_SIZE (16384)
158
159 struct aclk_database_cmdqueue {
160 unsigned head, tail;
@@ -189,6 +186,9 @@ struct aclk_database_worker_config {
186 int chart_updates;
187 int alert_updates;
188 time_t batch_created;
189 + int node_info_send;
190 + int chart_pending;
191 + int chart_reset_count;
192 struct aclk_database_worker_config *next;
193 };
194
@@ -212,14 +212,15 @@ static inline RRDHOST *find_host_by_node_id(char *node_id)
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);
215 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);
216 +extern void aclk_database_enq_cmd(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd);
217 extern void aclk_set_architecture(int mode);
218 +extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
219 +int aclk_worker_enq_cmd(char *node_id, struct aclk_database_cmd *cmd);
220 +void aclk_data_rotated(RRDHOST *host);
221 void sql_aclk_sync_init(void);
220 -void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
222 +void sql_check_aclk_table_list(struct aclk_database_worker_config *wc);
223 void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
224 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 +void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
226 #endif //NETDATA_SQLITE_ACLK_H
database/sqlite/sqlite_aclk_chart.c new
+762
@@ -0,0 +1,762 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "sqlite_functions.h"
4 +#include "sqlite_aclk_chart.h"
5 +
6 +#include "../../aclk/aclk_charts_api.h"
7 +
8 +#define CHECK_SQLITE_CONNECTION(db_meta) \
9 + if (unlikely(!db_meta)) { \
10 + if (default_rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE) { \
11 + return 1; \
12 + } \
13 + error_report("Database has not been initialized"); \
14 + return 1; \
15 + }
16 +
17 +static inline int sql_queue_chart_payload(struct aclk_database_worker_config *wc,
18 + void *data, enum aclk_database_opcode opcode)
19 +{
20 + if (unlikely(!wc))
21 + return 1;
22 +
23 + struct aclk_database_cmd cmd;
24 + memset(&cmd, 0, sizeof(cmd));
25 + cmd.opcode = opcode;
26 + cmd.data = data;
27 + aclk_database_enq_cmd(wc, &cmd);
28 + return 0;
29 +}
30 +
31 +static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size)
32 +{
33 + static __thread sqlite3_stmt *res = NULL;
34 + int rc;
35 + int send_status = 0;
36 +
37 + if (unlikely(!res)) {
38 + BUFFER *sql = buffer_create(1024);
39 + buffer_sprintf(sql,"SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
40 + "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",
41 + uuid_str, uuid_str);
42 + rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res);
43 + buffer_free(sql);
44 + if (rc != SQLITE_OK) {
45 + error_report("Failed to prepare statement to check payload data");
46 + return 0;
47 + }
48 + }
49 +
50 + rc = sqlite3_bind_blob(res, 1, uuid , sizeof(*uuid), SQLITE_STATIC);
51 + if (unlikely(rc != SQLITE_OK))
52 + goto bind_fail;
53 +
54 + rc = sqlite3_bind_blob(res, 2, payload , payload_size, SQLITE_STATIC);
55 + if (unlikely(rc != SQLITE_OK))
56 + goto bind_fail;
57 +
58 + while (sqlite3_step(res) == SQLITE_ROW) {
59 + send_status = sqlite3_column_int(res, 0);
60 + }
61 +
62 +bind_fail:
63 + if (unlikely(sqlite3_reset(res) != SQLITE_OK))
64 + error_report("Failed to reset statement in check payload, rc = %d", rc);
65 + return send_status;
66 +}
67 +
68 +static int aclk_add_chart_payload(char *uuid_str, uuid_t *uuid, char *claim_id, ACLK_PAYLOAD_TYPE payload_type,
69 + void *payload, size_t payload_size)
70 +{
71 + static __thread sqlite3_stmt *res_chart = NULL;
72 + int rc;
73 +
74 + rc = payload_sent(uuid_str, uuid, payload, payload_size);
75 + if (rc == 1)
76 + return 0;
77 +
78 + if (unlikely(!res_chart)) {
79 + BUFFER *sql = buffer_create(1024);
80 +
81 + buffer_sprintf(sql,"INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
82 + "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", uuid_str);
83 +
84 + rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res_chart);
85 + buffer_free(sql);
86 +
87 + if (rc != SQLITE_OK) {
88 + error_report("Failed to prepare statement to store chart payload data");
89 + return 1;
90 + }
91 + }
92 +
93 + uuid_t unique_uuid;
94 + uuid_generate(unique_uuid);
95 +
96 + uuid_t claim_uuid;
97 + uuid_parse(claim_id, claim_uuid);
98 +
99 + rc = sqlite3_bind_blob(res_chart, 1, &unique_uuid , sizeof(unique_uuid), SQLITE_STATIC);
100 + if (unlikely(rc != SQLITE_OK))
101 + goto bind_fail;
102 +
103 + rc = sqlite3_bind_blob(res_chart, 2, uuid , sizeof(*uuid), SQLITE_STATIC);
104 + if (unlikely(rc != SQLITE_OK))
105 + goto bind_fail;
106 +
107 + rc = sqlite3_bind_blob(res_chart, 3, &claim_uuid , sizeof(claim_uuid), SQLITE_STATIC);
108 + if (unlikely(rc != SQLITE_OK))
109 + goto bind_fail;
110 +
111 + rc = sqlite3_bind_int(res_chart, 4, payload_type);
112 + if (unlikely(rc != SQLITE_OK))
113 + goto bind_fail;
114 +
115 + rc = sqlite3_bind_blob(res_chart, 5, payload, payload_size, SQLITE_STATIC);
116 + if (unlikely(rc != SQLITE_OK))
117 + goto bind_fail;
118 +
119 + rc = execute_insert(res_chart);
120 + if (unlikely(rc != SQLITE_DONE))
121 + error_report("Failed store chart payload event, rc = %d", rc);
122 +
123 +bind_fail:
124 + if (unlikely(sqlite3_reset(res_chart) != SQLITE_OK))
125 + error_report("Failed to reset statement in store chart payload, rc = %d", rc);
126 + return (rc != SQLITE_DONE);
127 +}
128 +
129 +int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
130 +{
131 + int rc = 0;
132 + CHECK_SQLITE_CONNECTION(db_meta);
133 +
134 +#ifdef ACLK_NG
135 + char *claim_id = is_agent_claimed();
136 +
137 + RRDSET *st = cmd.data;
138 +
139 + if (likely(claim_id)) {
140 + struct chart_instance_updated chart_payload;
141 + memset(&chart_payload, 0, sizeof(chart_payload));
142 + chart_payload.config_hash = get_str_from_uuid(&st->state->hash_id);
143 + chart_payload.update_every = st->update_every;
144 + chart_payload.memory_mode = st->rrd_memory_mode;
145 + chart_payload.name = strdupz((char *)st->name);
146 + chart_payload.node_id = strdupz(wc->node_id);
147 + chart_payload.claim_id = claim_id;
148 + chart_payload.id = strdupz(st->id);
149 +
150 + struct label_index *labels = &st->state->labels;
151 + netdata_rwlock_wrlock(&labels->labels_rwlock);
152 + struct label *label_list = labels->head;
153 + struct label *chart_label = NULL;
154 + while (label_list) {
155 + chart_label = add_label_to_list(chart_label, label_list->key, label_list->value, label_list->label_source);
156 + label_list = label_list->next;
157 + }
158 + netdata_rwlock_unlock(&labels->labels_rwlock);
159 + chart_payload.label_head = chart_label;
160 +
161 + size_t size;
162 + char *payload = generate_chart_instance_updated(&size, &chart_payload);
163 + if (likely(payload))
164 + rc = aclk_add_chart_payload(wc->uuid_str, st->chart_uuid, claim_id, ACLK_PAYLOAD_CHART, (void *) payload, size);
165 + freez(payload);
166 + chart_instance_updated_destroy(&chart_payload);
167 + }
168 +#else
169 + UNUSED(wc);
170 + UNUSED(cmd);
171 +#endif
172 + return rc;
173 +}
174 +
175 +int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
176 +{
177 + int rc = 0;
178 + CHECK_SQLITE_CONNECTION(db_meta);
179 +
180 +#ifdef ACLK_NG
181 + char *claim_id = is_agent_claimed();
182 +
183 + RRDDIM *rd = cmd.data;
184 +
185 + if (likely(claim_id)) {
186 + time_t now = now_realtime_sec();
187 +
188 + time_t first_t = rd->state->query_ops.oldest_time(rd);
189 + time_t last_t = rd->state->query_ops.latest_time(rd);
190 +
191 + int live = ((now - last_t) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every));
192 +
193 + struct chart_dimension_updated dim_payload;
194 + size_t size;
195 +
196 + memset(&dim_payload, 0, sizeof(dim_payload));
197 + dim_payload.node_id = strdupz(wc->node_id);
198 + dim_payload.claim_id = claim_id;
199 + dim_payload.name = strdupz(rd->name);
200 + dim_payload.id = strdupz(rd->id);
201 +
202 + dim_payload.chart_id = strdupz(rd->rrdset->name);
203 + dim_payload.created_at.tv_sec = first_t;
204 + if (unlikely(!live))
205 + dim_payload.last_timestamp.tv_sec = last_t;
206 +
207 + char *payload = generate_chart_dimension_updated(&size, &dim_payload);
208 + if (likely(payload))
209 + rc = aclk_add_chart_payload(wc->uuid_str, &rd->state->metric_uuid, claim_id, ACLK_PAYLOAD_DIMENSION, (void *)payload, size);
210 + freez((char *)dim_payload.node_id);
211 + freez((char *)dim_payload.chart_id);
212 + freez((char *)dim_payload.name);
213 + freez((char *)dim_payload.id);
214 + freez(payload);
215 + freez(claim_id);
216 + }
217 + rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK);
218 +#else
219 + UNUSED(wc);
220 + UNUSED(cmd);
221 +#endif
222 + return rc;
223 +}
224 +
225 +void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
226 +{
227 +#ifdef ACLK_NG
228 + int rc;
229 +
230 + wc->chart_pending = 0;
231 + if (unlikely(!wc->chart_updates)) {
232 + debug(D_ACLK_SYNC,"Ignoring chart push event, updates have been turned off for node %s", wc->node_id);
233 + return;
234 + }
235 +
236 + char *claim_id = is_agent_claimed();
237 + if (unlikely(!claim_id))
238 + return;
239 +
240 + int limit = cmd.count > 0 ? cmd.count : 1;
241 +
242 + uint64_t first_sequence;
243 + uint64_t last_sequence;
244 + time_t last_timestamp;
245 +
246 + BUFFER *sql = buffer_create(1024);
247 +
248 + sqlite3_stmt *res = NULL;
249 +
250 + buffer_sprintf(sql, "SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid " \
251 + "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \
252 + "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \
253 + "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);
254 +
255 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
256 + if (rc != SQLITE_OK) {
257 + error_report("Failed to prepare statement when trying to send a chart update via ACLK");
258 + buffer_free(sql);
259 + freez(claim_id);
260 + return;
261 + }
262 +
263 + rc = sqlite3_bind_text(res, 1, claim_id , -1, SQLITE_STATIC);
264 + if (unlikely(rc != SQLITE_OK))
265 + goto bind_fail;
266 +
267 + char **payload_list = callocz(limit+1, sizeof(char *));
268 + size_t *payload_list_size = callocz(limit+1, sizeof(size_t));
269 + size_t *payload_list_max_size = callocz(limit+1, sizeof(size_t));
270 + struct aclk_message_position *position_list = callocz(limit+1, sizeof(*position_list));
271 + int *is_dim = callocz(limit+1, sizeof(*is_dim));
272 +
273 + int loop = cmd.param1;
274 +
275 + while (loop > 0) {
276 + uint64_t previous_sequence_id = wc->chart_sequence_id;
277 + int count = 0;
278 + first_sequence = 0;
279 + last_sequence = 0;
280 + while (count < limit && sqlite3_step(res) == SQLITE_ROW) {
281 + size_t payload_size = sqlite3_column_bytes(res, 1);
282 + if (payload_list_max_size[count] < payload_size) {
283 + freez(payload_list[count]);
284 + payload_list_max_size[count] = payload_size;
285 + payload_list[count] = mallocz(payload_size);
286 + }
287 + payload_list_size[count] = payload_size;
288 + memcpy(payload_list[count], sqlite3_column_blob(res, 1), payload_size);
289 + position_list[count].sequence_id = (uint64_t)sqlite3_column_int64(res, 0);
290 + position_list[count].previous_sequence_id = previous_sequence_id;
291 + position_list[count].seq_id_creation_time.tv_sec = sqlite3_column_int64(res, 2);
292 + position_list[count].seq_id_creation_time.tv_usec = 0;
293 + if (!first_sequence)
294 + first_sequence = position_list[count].sequence_id;
295 + last_sequence = position_list[count].sequence_id;
296 + last_timestamp = position_list[count].seq_id_creation_time.tv_sec;
297 + previous_sequence_id = last_sequence;
298 + is_dim[count] = sqlite3_column_int(res, 3) > 0;
299 + count++;
300 + }
301 + freez(payload_list[count]);
302 + payload_list_max_size[count] = 0;
303 + payload_list[count] = NULL;
304 +
305 + rc = sqlite3_reset(res);
306 + if (unlikely(rc != SQLITE_OK))
307 + error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
308 +
309 + if (likely(first_sequence)) {
310 + buffer_flush(sql);
311 +
312 + db_lock();
313 + buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "
314 + "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",
315 + wc->uuid_str, first_sequence, last_sequence);
316 + db_execute(buffer_tostring(sql));
317 +
318 + buffer_flush(sql);
319 + buffer_sprintf(sql, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
320 + " SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s s "
321 + " WHERE date_submitted IS NOT NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64
322 + " ;",
323 + wc->uuid_str, wc->uuid_str, first_sequence, last_sequence);
324 + db_execute(buffer_tostring(sql));
325 + db_unlock();
326 +
327 + aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id);
328 + wc->chart_sequence_id = last_sequence;
329 + wc->chart_timestamp = last_timestamp;
330 + }
331 + --loop;
332 + }
333 +
334 + for (int i = 0; i <= limit; ++i)
335 + freez(payload_list[i]);
336 +
337 + freez(payload_list);
338 + freez(payload_list_size);
339 + freez(position_list);
340 + freez(is_dim);
341 +
342 +bind_fail:
343 + rc = sqlite3_finalize(res);
344 + if (unlikely(rc != SQLITE_OK))
345 + error_report("Failed to finalize statement when pushing chart events, rc = %d", rc);
346 +
347 + buffer_free(sql);
348 + freez(claim_id);
349 +#else
350 + UNUSED(wc);
351 + UNUSED(cmd);
352 +#endif
353 + return;
354 +}
355 +
356 +
357 +// Push one chart config to the cloud
358 +int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
359 +{
360 + UNUSED(wc);
361 +#ifdef ACLK_NG
362 +
363 + CHECK_SQLITE_CONNECTION(db_meta);
364 +
365 + sqlite3_stmt *res = NULL;
366 + int rc = 0;
367 +
368 + char *hash_id = (char *) cmd.data_param;
369 +
370 + uuid_t hash_uuid;
371 + rc = uuid_parse(hash_id, hash_uuid);
372 +
373 + if (unlikely(rc)) {
374 + freez((char *) cmd.data_param);
375 + return 1;
376 + }
377 +
378 + BUFFER *sql = buffer_create(1024);
379 + buffer_sprintf(sql, "SELECT type, family, context, title, priority, plugin, module, unit, chart_type " \
380 + "FROM chart_hash WHERE hash_id = @hash_id;");
381 +
382 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
383 + if (rc != SQLITE_OK) {
384 + error_report("Failed to prepare statement when trying to fetch a chart hash configuration");
385 + goto fail;
386 + }
387 +
388 + rc = sqlite3_bind_blob(res, 1, &hash_uuid , sizeof(hash_uuid), SQLITE_STATIC);
389 + if (unlikely(rc != SQLITE_OK))
390 + goto bind_fail;
391 +
392 + struct chart_config_updated chart_config;
393 + chart_config.config_hash = NULL;
394 +
395 + while (sqlite3_step(res) == SQLITE_ROW) {
396 + chart_config.type = strdupz((char *)sqlite3_column_text(res, 0));
397 + chart_config.family = strdupz((char *)sqlite3_column_text(res, 1));
398 + chart_config.context = strdupz((char *)sqlite3_column_text(res, 2));
399 + chart_config.title = strdupz((char *)sqlite3_column_text(res, 3));
400 + chart_config.priority = sqlite3_column_int64(res, 4);
401 + chart_config.plugin = strdupz((char *)sqlite3_column_text(res, 5));
402 + chart_config.module = sqlite3_column_bytes(res, 6) > 0 ? strdupz((char *)sqlite3_column_text(res, 6)) : NULL;
403 + chart_config.chart_type = (RRDSET_TYPE) sqlite3_column_int(res,8);
404 + chart_config.units = strdupz((char *)sqlite3_column_text(res, 7));
405 + chart_config.config_hash = strdupz(hash_id);
406 + }
407 +
408 + if (likely(chart_config.config_hash)) {
409 + debug(D_ACLK_SYNC, "Sending chart config for %s", hash_id);
410 + aclk_chart_config_updated(&chart_config, 1);
411 + destroy_chart_config_updated(&chart_config);
412 + }
413 + else
414 + info("DEBUG: Chart config for %s not found", hash_id);
415 +
416 + bind_fail:
417 + rc = sqlite3_finalize(res);
418 + if (unlikely(rc != SQLITE_OK))
419 + error_report("Failed to reset statement when pushing chart config hash, rc = %d", rc);
420 + fail:
421 + freez((char *) cmd.data_param);
422 + buffer_free(sql);
423 + return rc;
424 +#else
425 + UNUSED(cmd);
426 + return 0;
427 +#endif
428 +}
429 +
430 +void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
431 +{
432 + int rc;
433 + sqlite3_stmt *res = NULL;
434 +
435 + BUFFER *sql = buffer_create(1024);
436 +
437 + buffer_sprintf(sql, "UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "
438 + "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);
439 +
440 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
441 + if (rc != SQLITE_OK) {
442 + error_report("Failed to prepare statement count sequence ids in the database");
443 + goto prepare_fail;
444 + }
445 +
446 + rc = sqlite3_bind_int64(res, 1, (uint64_t) cmd.param1);
447 + if (unlikely(rc != SQLITE_OK))
448 + goto bind_fail;
449 +
450 + rc = execute_insert(res);
451 + if (rc != SQLITE_DONE)
452 + error_report("Failed to ACK sequence id, rc = %d", rc);
453 +
454 + bind_fail:
455 + if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
456 + error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);
457 +
458 + prepare_fail:
459 + buffer_free(sql);
460 + return;
461 +}
462 +
463 +void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
464 +{
465 + BUFFER *sql = buffer_create(1024);
466 + buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted = NULL WHERE sequence_id >= %"PRIu64";",
467 + wc->uuid_str, cmd.param1);
468 + db_execute(buffer_tostring(sql));
469 + if (cmd.param1 == 1) {
470 + db_lock();
471 + buffer_flush(sql);
472 + info("DEBUG: Deleting all data for %s", wc->uuid_str);
473 + buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s; DELETE FROM aclk_chart_%s; DELETE FROM aclk_chart_latest_%s;",
474 + wc->uuid_str, wc->uuid_str, wc->uuid_str);
475 + db_execute(buffer_tostring(sql));
476 + db_unlock();
477 + wc->chart_sequence_id = 0;
478 + wc->chart_timestamp = 0;
479 +
480 +#ifdef ACLK_NG
481 + RRDHOST *host = wc->host;
482 + rrdhost_rdlock(host);
483 + RRDSET *st;
484 + rrdset_foreach_read(st, host) {
485 + rrdset_rdlock(st);
486 + RRDDIM *rd;
487 + rrddim_foreach_read(rd, st) {
488 + rd->state->aclk_live_status = (rd->state->aclk_live_status == 0);
489 + }
490 + rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
491 + rrdset_unlock(st);
492 + }
493 + rrdhost_unlock(host);
494 +#endif
495 + }
496 + else {
497 + //sql_chart_deduplicate(wc, cmd);
498 + sql_get_last_chart_sequence(wc, cmd);
499 + }
500 + buffer_free(sql);
501 + wc->chart_updates = 1;
502 + return;
503 +}
504 +
505 +//
506 +// Functions called directly from ACLK threads and will queue commands
507 +//
508 +void aclk_get_chart_config(char **hash_id)
509 +{
510 + struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *)localhost->dbsync_worker;
511 +
512 + if (unlikely(!wc || !hash_id))
513 + return;
514 +
515 + struct aclk_database_cmd cmd;
516 + memset(&cmd, 0, sizeof(cmd));
517 + cmd.opcode = ACLK_DATABASE_PUSH_CHART_CONFIG;
518 + for (int i = 0; hash_id[i]; ++i) {
519 + // TODO: Verify that we have a valid hash_id
520 + debug(D_ACLK_SYNC,"Request %d for chart config with hash [%s] received", i, hash_id[i]);
521 + cmd.data_param = (void *)strdupz(hash_id[i]);
522 + aclk_database_enq_cmd(wc, &cmd);
523 + }
524 + return;
525 +}
526 +
527 +// Send a command to a node_id
528 +// Need to discover the thread that will handle the request
529 +// if thread not in active hosts, then try to find in the queue
530 +static void aclk_submit_param_command(char *node_id, enum aclk_database_opcode aclk_command, uint64_t param)
531 +{
532 + if (unlikely(!node_id))
533 + return;
534 +
535 + struct aclk_database_worker_config *wc = NULL;
536 + struct aclk_database_cmd cmd;
537 + memset(&cmd, 0, sizeof(cmd));
538 + cmd.opcode = aclk_command;
539 + cmd.param1 = param;
540 +
541 + rrd_wrlock();
542 + RRDHOST *host = find_host_by_node_id(node_id);
543 + if (likely(host))
544 + wc = (struct aclk_database_worker_config *)host->dbsync_worker;
545 + rrd_unlock();
546 + if (wc)
547 + aclk_database_enq_cmd(wc, &cmd);
548 + else {
549 + if (aclk_worker_enq_cmd(node_id, &cmd))
550 + error_report("ACLK synchronization thread is not active for node id %s", node_id);
551 + }
552 + return;
553 +}
554 +
555 +void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id)
556 +{
557 + if (unlikely(!node_id))
558 + return;
559 +
560 + debug(D_ACLK_SYNC, "NODE %s reports last sequence id received %"PRIu64, node_id, last_sequence_id);
561 + aclk_submit_param_command(node_id, ACLK_DATABASE_CHART_ACK, last_sequence_id);
562 + return;
563 +}
564 +
565 +void aclk_reset_chart_event(char *node_id, uint64_t last_sequence_id)
566 +{
567 + if (unlikely(!node_id))
568 + return;
569 +
570 + debug(D_ACLK_SYNC, "NODE %s wants to resync from %"PRIu64, node_id, last_sequence_id);
571 + aclk_submit_param_command(node_id, ACLK_DATABASE_RESET_CHART, last_sequence_id);
572 + return;
573 +}
574 +
575 +// ST is read locked
576 +int sql_queue_chart_to_aclk(RRDSET *st)
577 +{
578 + return sql_queue_chart_payload((struct aclk_database_worker_config *) st->rrdhost->dbsync_worker,
579 + st, ACLK_DATABASE_ADD_CHART);
580 +}
581 +
582 +int sql_queue_dimension_to_aclk(RRDDIM *rd)
583 +{
584 + int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker,
585 + rd, ACLK_DATABASE_ADD_DIMENSION);
586 + if (likely(!rc))
587 + rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
588 + return rc;
589 +}
590 +
591 +void sql_chart_deduplicate(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
592 +{
593 + UNUSED(cmd);
594 +
595 + BUFFER *sql = buffer_create(1024);
596 +
597 + db_lock();
598 + buffer_sprintf(sql, "DROP TABLE IF EXISTS t_%s;", wc->uuid_str);
599 + db_execute(buffer_tostring(sql));
600 +
601 + buffer_flush(sql);
602 + buffer_sprintf(sql, "CREATE TABLE t_%s AS SELECT * FROM aclk_chart_payload_%s WHERE unique_id IN "
603 + "(SELECT unique_id from aclk_chart_%s WHERE date_submitted IS NULL AND update_count > 0);",
604 + wc->uuid_str, wc->uuid_str, wc->uuid_str);
605 + db_execute(buffer_tostring(sql));
606 +
607 + buffer_flush(sql);
608 + buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s WHERE unique_id IN (SELECT unique_id FROM t_%s); " ,
609 + wc->uuid_str, wc->uuid_str);
610 + db_execute(buffer_tostring(sql));
611 +
612 + buffer_flush(sql);
613 + buffer_sprintf(sql, "DELETE FROM aclk_chart_%s WHERE unique_id IN (SELECT unique_id FROM t_%s);",
614 + wc->uuid_str, wc->uuid_str);
615 + db_execute(buffer_tostring(sql));
616 +
617 + buffer_flush(sql);
618 + buffer_sprintf(sql, "DELETE FROM aclk_chart_latest_%s WHERE unique_id IN (SELECT unique_id FROM t_%s);",
619 + wc->uuid_str, wc->uuid_str);
620 + db_execute(buffer_tostring(sql));
621 +
622 + buffer_flush(sql);
623 + buffer_sprintf(sql, "INSERT INTO aclk_chart_payload_%s SELECT * FROM t_%s ORDER BY DATE_CREATED ASC;",
624 + wc->uuid_str, wc->uuid_str);
625 + db_execute(buffer_tostring(sql));
626 +
627 + buffer_flush(sql);
628 + buffer_sprintf(sql, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
629 + "SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s where sequence_id IN "
630 + "(SELECT sequence_id FROM aclk_chart_%s WHERE date_submitted IS NOT NULL "
631 + "GROUP BY uuid HAVING sequence_id = MAX(sequence_id));"
632 + , wc->uuid_str, wc->uuid_str, wc->uuid_str);
633 + db_execute(buffer_tostring(sql));
634 +
635 + buffer_flush(sql);
636 + buffer_sprintf(sql, "DROP TABLE IF EXISTS t_%s;", wc->uuid_str);
637 + db_execute(buffer_tostring(sql));
638 + db_unlock();
639 +
640 + sql_get_last_chart_sequence(wc, cmd);
641 +
642 + buffer_free(sql);
643 + return;
644 +}
645 +
646 +void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
647 +{
648 + UNUSED(cmd);
649 +
650 + BUFFER *sql = buffer_create(1024);
651 +
652 + buffer_sprintf(sql,"SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \
653 + "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);
654 +
655 + int rc;
656 + sqlite3_stmt *res = NULL;
657 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
658 + if (rc != SQLITE_OK) {
659 + error_report("Failed to prepare statement to find last chart sequence id");
660 + goto fail;
661 + }
662 +
663 + wc->chart_sequence_id = 0;
664 + wc->chart_timestamp = 0;
665 + while (sqlite3_step(res) == SQLITE_ROW) {
666 + wc->chart_sequence_id = (uint64_t) sqlite3_column_int64(res, 0);
667 + wc->chart_timestamp = (time_t) sqlite3_column_int64(res, 1);
668 + }
669 +
670 + debug(D_ACLK_SYNC,"Node %s reports last sequence_id=%"PRIu64, wc->node_id, wc->chart_sequence_id);
671 +
672 + rc = sqlite3_finalize(res);
673 + if (unlikely(rc != SQLITE_OK))
674 + error_report("Failed to reset statement when fetching chart sequence info, rc = %d", rc);
675 +
676 +fail:
677 + buffer_free(sql);
678 + return;
679 +}
680 +
681 +// Start streaming charts / dimensions for node_id
682 +void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at, uint64_t batch_id)
683 +{
684 +#ifdef ACLK_NG
685 + if (unlikely(!node_id))
686 + return;
687 +
688 + debug(D_ACLK_SYNC,"START streaming charts for node %s from sequence %"PRIu64" t=%ld, batch=%"PRIu64, node_id,
689 + sequence_id, created_at, batch_id);
690 + uuid_t node_uuid;
691 + uuid_parse(node_id, node_uuid);
692 +
693 + struct aclk_database_worker_config *wc = NULL;
694 + rrd_wrlock();
695 + RRDHOST *host = localhost;
696 + while(host) {
697 + if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) {
698 + rrd_unlock();
699 + wc = (struct aclk_database_worker_config *)host->dbsync_worker;
700 + if (likely(wc)) {
701 + // if (unlikely(!wc->chart_updates)) {
702 + // struct aclk_database_cmd cmd;
703 + // cmd.opcode = ACLK_DATABASE_NODE_INFO;
704 + // cmd.completion = NULL;
705 + // aclk_database_enq_cmd(wc, &cmd);
706 + // }
707 +
708 + wc->chart_reset_count++;
709 + wc->chart_updates = 0;
710 + wc->batch_id = batch_id;
711 + wc->batch_created = now_realtime_sec();
712 + info("DEBUG: START streaming charts for %s (%s) enabled -- last streamed sequence %"PRIu64" t=%ld (reset count=%d)", node_id, wc->uuid_str,
713 + wc->chart_sequence_id, wc->chart_timestamp, wc->chart_reset_count);
714 + // If mismatch detected
715 + if (sequence_id > wc->chart_sequence_id || wc->chart_reset_count > 10) {
716 + info("DEBUG: Full resync requested -- reset_count=%d", wc->chart_reset_count);
717 + chart_reset_t chart_reset;
718 + chart_reset.node_id = strdupz(node_id);
719 + chart_reset.claim_id = is_agent_claimed();
720 + chart_reset.reason = SEQ_ID_NOT_EXISTS;
721 + aclk_chart_reset(chart_reset);
722 +// wc->chart_updates = 0;
723 + wc->chart_reset_count = -1;
724 + return;
725 + } else {
726 + struct aclk_database_cmd cmd;
727 + // TODO: handle timestamp
728 +// if (!wc->chart_reset_count)
729 +// wc->chart_delay = now_realtime_sec() + 60;
730 +// else
731 +// wc->chart_delay = 0;
732 +
733 + if (sequence_id < wc->chart_sequence_id) { // || created_at != wc->chart_timestamp) {
734 +// wc->chart_updates = 0;
735 + if (sequence_id)
736 + info("DEBUG: Synchonization mismatch detected");
737 + else
738 + info("DEBUG: Synchonization mismatch detected; full resync ACKed from the cloud");
739 + cmd.opcode = ACLK_DATABASE_RESET_CHART;
740 + cmd.param1 = sequence_id + 1;
741 + cmd.completion = NULL;
742 + aclk_database_enq_cmd(wc, &cmd);
743 + }
744 + else
745 + wc->chart_updates = 1;
746 + }
747 + }
748 + else
749 + error("ACLK synchronization thread is not active for host %s", host->hostname);
750 + return;
751 + }
752 + host = host->next;
753 + }
754 + rrd_unlock();
755 +#else
756 + UNUSED(node_id);
757 + UNUSED(sequence_id);
758 + UNUSED(created_at);
759 + UNUSED(batch_id);
760 +#endif
761 + return;
762 +}
database/sqlite/sqlite_aclk_chart.h new
+40
@@ -0,0 +1,40 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef NETDATA_SQLITE_ACLK_CHART_H
4 +#define NETDATA_SQLITE_ACLK_CHART_H
5 +
6 +
7 +typedef enum payload_type {
8 + ACLK_PAYLOAD_CHART,
9 + ACLK_PAYLOAD_DIMENSION,
10 + ACLK_PAYLOAD_DIMENSION_ROTATED
11 +} ACLK_PAYLOAD_TYPE;
12 +
13 +extern sqlite3 *db_meta;
14 +
15 +#ifndef RRDSET_MINIMUM_LIVE_COUNT
16 +#define RRDSET_MINIMUM_LIVE_COUNT 3
17 +#endif
18 +
19 +//void aclk_status_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
20 +extern int sql_queue_chart_to_aclk(RRDSET *st);
21 +extern int sql_queue_dimension_to_aclk(RRDDIM *rd);
22 +extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
23 +extern void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae);
24 +int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
25 +int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
26 +int aclk_add_offline_dimension_event(struct aclk_database_worker_config *wc, char *node_id, char *chart_name, uuid_t *dim_uuid, char *rd_id, char *rd_name, time_t first_entry_t, time_t last_entry_t);
27 +int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
28 +void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id);
29 +void aclk_get_chart_config(char **hash_id_list);
30 +void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
31 +void aclk_reset_chart_event(char *node_id, uint64_t last_sequence_id);
32 +void aclk_start_streaming(char *node_id, uint64_t seq_id, time_t created_at, uint64_t batch_id);
33 +void sql_chart_deduplicate(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
34 +void sql_check_dimension_state(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
35 +void sql_check_rotation_state(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
36 +void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
37 +void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
38 +void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
39 +void sql_update_metric_statistics(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
40 +#endif //NETDATA_SQLITE_ACLK_CHART_H
database/sqlite/sqlite_functions.h
+2
@@ -63,6 +63,8 @@ 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 int prepare_statement(sqlite3 *database, char *query, sqlite3_stmt **statement);
67 +extern int execute_insert(sqlite3_stmt *res);
68 extern void db_execute(const char *cmd);
69 extern int file_is_migrated(char *path);
70 extern void add_migrated_file(char *path, uint64_t file_size);