@cryptotaxi247 / netdata-1 / commits / 154cf74d6

Improve agent cloud chart synchronization (#12655)

* Try to queue dimension always when: Trying to clean obsolete charts If chart has been sent and liveness apparently changed * delay rotation and skip chart check if not send to cloud * No need to CLEAR flag during database rotation Do not clear chart ACLK status for dimension requests * Change payload_sent to return timestamp of submitted message * Clear the dimension ACLK flag if we are processing all the charts again * Check if dimension is already queued to ACLK and ignore it If queue fails then reset it to retry Already try to queue the dimension * Improve dimension cleanup during the retention message calculation * Change queue_dimension_to_aclk to return void * If no time range for this dimension then assume it is deleted * Start streaming for inactive nodes * Remove dead code * Correctly report hostname in the access log * Schedule a dimension deletion without trying to submit a message immediately * Enable dimension cleanup -- also delete dimension if not found in the dbengine files Free hostname

Stelios Fragkakis committed May 3, 2022 at 21:38 UTC 154cf74d6a97e0712f0d6b6a970d3e5aaf439f0e
6 files changed +125 -103
database/rrddim.c
-22
@@ -194,9 +194,6 @@ void rrdcalc_link_to_rrddim(RRDDIM *rd, RRDSET *st, RRDHOST *host) {
194 }
195 }
196 }
197 -#ifdef ENABLE_ACLK
198 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
199 -#endif
197 }
198
199 RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collected_number multiplier,
@@ -441,9 +438,6 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
438 ml_new_dimension(rd);
439
440 rrdset_unlock(st);
444 -#ifdef ENABLE_ACLK
445 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
446 -#endif
441 return(rd);
442 }
443
@@ -516,10 +510,6 @@ void rrddim_free_custom(RRDSET *st, RRDDIM *rd, int db_rotated)
510 freez(rd);
511 break;
512 }
519 -#ifdef ENABLE_ACLK
520 - if (db_rotated || RRD_MEMORY_MODE_DBENGINE != rrd_memory_mode)
521 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
522 -#endif
513 }
514
515
@@ -539,9 +529,6 @@ int rrddim_hide(RRDSET *st, const char *id) {
529 (void) sql_set_dimension_option(&rd->state->metric_uuid, "hidden");
530
531 rrddim_flag_set(rd, RRDDIM_FLAG_HIDDEN);
542 -#ifdef ENABLE_ACLK
543 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
544 -#endif
532 return 0;
533 }
534
@@ -557,9 +544,6 @@ int rrddim_unhide(RRDSET *st, const char *id) {
544 (void) sql_set_dimension_option(&rd->state->metric_uuid, NULL);
545
546 rrddim_flag_clear(rd, RRDDIM_FLAG_HIDDEN);
560 -#ifdef ENABLE_ACLK
561 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
562 -#endif
547 return 0;
548 }
549
@@ -572,18 +556,12 @@ inline void rrddim_is_obsolete(RRDSET *st, RRDDIM *rd) {
556 }
557 rrddim_flag_set(rd, RRDDIM_FLAG_OBSOLETE);
558 rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS);
575 -#ifdef ENABLE_ACLK
576 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
577 -#endif
559 }
560
561 inline void rrddim_isnot_obsolete(RRDSET *st __maybe_unused, RRDDIM *rd) {
562 debug(D_RRD_CALLS, "rrddim_isnot_obsolete() for chart %s, dimension %s", st->name, rd->name);
563
564 rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE);
584 -#ifdef ENABLE_ACLK
585 - rrdset_flag_clear(st, RRDSET_FLAG_ACLK);
586 -#endif
565 }
566
567 // ----------------------------------------------------------------------------
database/rrdhost.c
+2 -3
@@ -1488,9 +1488,8 @@ restart_after_removal:
1488 continue;
1489 }
1490 #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1491 - else {
1492 - aclk_send_dimension_update(rd);
1493 - }
1491 + else
1492 + queue_dimension_to_aclk(rd);
1493 #endif
1494 }
1495 last = rd;
database/rrdset.c
+9 -14
@@ -1368,8 +1368,9 @@ void rrdset_done(RRDSET *st) {
1368 #ifdef ENABLE_ACLK
1369 if (likely(!st->state->is_ar_chart)) {
1370 if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1371 - if (likely(st->dimensions && st->counter_done && !queue_chart_to_aclk(st)))
1371 + if (likely(st->dimensions && st->counter_done && !queue_chart_to_aclk(st))) {
1372 rrdset_flag_set(st, RRDSET_FLAG_ACLK);
1373 + }
1374 }
1375 }
1376 #endif
@@ -1796,20 +1797,14 @@ after_second_database_work:
1797 continue;
1798
1799 #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1799 - if (likely(!st->state->is_ar_chart)) {
1800 - if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)) {
1801 - int live =
1802 - ((mark - rd->last_collected_time.tv_sec) < RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every);
1803 - if (unlikely(live != rd->state->aclk_live_status)) {
1804 - if (likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1805 - if (likely(!queue_dimension_to_aclk(rd))) {
1806 - rd->state->aclk_live_status = live;
1807 - rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
1808 - }
1809 - }
1800 + if (likely(!st->state->is_ar_chart)) {
1801 + if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN) && likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1802 + int live =
1803 + ((mark - rd->last_collected_time.tv_sec) < RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every);
1804 + if (unlikely(live != rd->state->aclk_live_status))
1805 + queue_dimension_to_aclk(rd);
1806 }
1807 }
1812 - }
1808 #endif
1809 if(unlikely(!rd->updated))
1810 continue;
@@ -1911,7 +1906,7 @@ after_second_database_work:
1906 } else {
1907 /* Do not delete this dimension */
1908 #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
1914 - aclk_send_dimension_update(rd);
1909 + queue_dimension_to_aclk(rd);
1910 #endif
1911 last = rd;
1912 rd = rd->next;
database/sqlite/sqlite_aclk.h
+1 -1
@@ -16,7 +16,7 @@
16 #endif
17 #define ACLK_MAX_ALERT_UPDATES (5)
18 #define ACLK_DATABASE_CLEANUP_FIRST (60)
19 -#define ACLK_DATABASE_ROTATION_DELAY (60)
19 +#define ACLK_DATABASE_ROTATION_DELAY (180)
20 #define ACLK_DATABASE_CLEANUP_INTERVAL (3600)
21 #define ACLK_DATABASE_ROTATION_INTERVAL (3600)
22 #define ACLK_DELETE_ACK_INTERNAL (600)
database/sqlite/sqlite_aclk_chart.c
+112 -62
@@ -22,20 +22,20 @@ sql_queue_chart_payload(struct aclk_database_worker_config *wc, void *data, enum
22 return rc;
23 }
24
25 -static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size)
25 +static time_t payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size)
26 {
27 static __thread sqlite3_stmt *res = NULL;
28 int rc;
29 - int send_status = 0;
29 + time_t send_status = 0;
30
31 if (unlikely(!res)) {
32 char sql[ACLK_SYNC_QUERY_SIZE];
33 - snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
34 - "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",
35 - uuid_str, uuid_str);
33 + snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT acl.date_submitted FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
34 + "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",
35 + uuid_str, uuid_str);
36 rc = prepare_statement(db_meta, sql, &res);
37 if (rc != SQLITE_OK) {
38 - error_report("Failed to prepare statement to check payload data");
38 + error_report("Failed to prepare statement to check payload data on %s", sql);
39 return 0;
40 }
41 }
@@ -49,7 +49,7 @@ static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payl
49 goto bind_fail;
50
51 while (sqlite3_step(res) == SQLITE_ROW) {
52 - send_status = sqlite3_column_int(res, 0);
52 + send_status = (time_t) sqlite3_column_int64(res, 0);
53 }
54
55 bind_fail:
@@ -59,22 +59,23 @@ bind_fail:
59 }
60
61 static int aclk_add_chart_payload(struct aclk_database_worker_config *wc, uuid_t *uuid, char *claim_id,
62 - ACLK_PAYLOAD_TYPE payload_type, void *payload, size_t payload_size, int *send_status)
62 + ACLK_PAYLOAD_TYPE payload_type, void *payload, size_t payload_size, time_t *send_status)
63 {
64 static __thread sqlite3_stmt *res_chart = NULL;
65 int rc;
66 + time_t date_submitted;
67
67 - rc = payload_sent(wc->uuid_str, uuid, payload, payload_size);
68 + date_submitted = payload_sent(wc->uuid_str, uuid, payload, payload_size);
69 if (send_status)
69 - *send_status = rc;
70 - if (rc == 1)
70 + *send_status = date_submitted;
71 + if (date_submitted)
72 return 0;
73
74 if (unlikely(!res_chart)) {
75 char sql[ACLK_SYNC_QUERY_SIZE];
76 snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,
76 - "INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
77 - "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);
77 + "INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
78 + "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);
79 rc = prepare_statement(db_meta, sql, &res_chart);
80 if (rc != SQLITE_OK) {
81 error_report("Failed to prepare statement to store chart payload data");
@@ -168,7 +169,7 @@ int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_dat
169
170 static inline int aclk_upd_dimension_event(struct aclk_database_worker_config *wc, char *claim_id, uuid_t *dim_uuid,
171 const char *dim_id, const char *dim_name, const char *chart_type_id, time_t first_time, time_t last_time,
171 - int *send_status)
172 + time_t *send_status)
173 {
174 int rc = 0;
175 size_t size;
@@ -279,7 +280,7 @@ int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk
280 RRDDIM *rd = cmd.data;
281
282 if (likely(claim_id)) {
282 - int send_status = 0;
283 + time_t send_status = 0;
284 time_t now = now_realtime_sec();
285
286 time_t first_t = rd->state->query_ops.oldest_time(rd);
@@ -337,6 +338,12 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
338 char sql[ACLK_SYNC_QUERY_SIZE];
339 static __thread sqlite3_stmt *res = NULL;
340
341 + char *hostname = NULL;
342 + if (wc->host)
343 + hostname = strdupz(wc->host->hostname);
344 + else
345 + hostname = get_hostname_by_node_id(wc->node_id);
346 +
347 if (unlikely(!res)) {
348 snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid " \
349 "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \
@@ -419,7 +426,7 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
426 log_access(
427 "ACLK RES [%s (%s)]: CHARTS SENT from %" PRIu64 " to %" PRIu64 " batch=%" PRIu64,
428 wc->node_id,
422 - wc->host ? wc->host->hostname : "N/A",
429 + hostname ? hostname : "N/A",
430 first_sequence,
431 last_sequence,
432 wc->batch_id);
@@ -440,7 +447,7 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
447 log_access(
448 "ACLK STA [%s (%s)]: Sync of charts and dimensions done in %ld seconds.",
449 wc->node_id,
443 - wc->host ? wc->host->hostname : "N/A",
450 + hostname ? hostname : "N/A",
451 now_realtime_sec() - wc->startup_time);
452 }
453
@@ -459,6 +466,7 @@ bind_fail:
466 error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
467
468 freez(claim_id);
469 + freez(hostname);
470 return;
471 }
472
@@ -583,8 +591,13 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
591 cmd.param1);
592 db_execute(buffer_tostring(sql));
593 if (cmd.param1 == 1) {
594 + char *hostname = NULL;
595 + if (wc->host)
596 + hostname = strdupz(wc->host->hostname);
597 + else
598 + hostname = get_hostname_by_node_id(wc->node_id);
599 buffer_flush(sql);
587 - log_access("ACLK REQ [%s (%s)]: Received chart full resync.", wc->node_id, wc->host ? wc->host->hostname : "N/A");
600 + log_access("ACLK REQ [%s (%s)]: Received chart full resync.", wc->node_id, hostname? hostname : "N/A");
601 buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s; DELETE FROM aclk_chart_%s; " \
602 "DELETE FROM aclk_chart_latest_%s;", wc->uuid_str, wc->uuid_str, wc->uuid_str);
603 db_lock();
@@ -609,6 +622,7 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
622 RRDDIM *rd;
623 rrddim_foreach_read(rd, st)
624 {
625 + rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK);
626 rd->state->aclk_live_status = (rd->state->aclk_live_status == 0);
627 }
628 rrdset_unlock(st);
@@ -616,6 +630,7 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
630 rrdhost_unlock(host);
631 } else
632 error_report("ACLK synchronization thread for %s is not linked to HOST", wc->host_guid);
633 + freez(hostname);
634 } else {
635 log_access(
636 "ACLK STA [%s (%s)]: Restarting chart sync from sequence %" PRIu64,
@@ -705,25 +720,28 @@ void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at
720 if (unlikely(!node_id))
721 return;
722
708 - // log_access("ACLK REQ [%s (N/A)]: CHARTS STREAM from %"PRIu64" t=%ld batch=%"PRIu64, node_id,
709 - // sequence_id, created_at, batch_id);
710 -
723 uuid_t node_uuid;
724 if (uuid_parse(node_id, node_uuid)) {
725 log_access("ACLK REQ [%s (N/A)]: CHARTS STREAM ignored, invalid node id", node_id);
726 return;
727 }
728
717 - struct aclk_database_worker_config *wc = NULL;
729 + struct aclk_database_worker_config *wc = find_inactive_wc_by_node_id(node_id);
730 rrd_rdlock();
731 RRDHOST *host = localhost;
732 while(host) {
721 - if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) {
733 + if (wc || (host->node_id && !(uuid_compare(*host->node_id, node_uuid)))) {
734 rrd_unlock();
723 - wc = (struct aclk_database_worker_config *)host->dbsync_worker ?
724 - (struct aclk_database_worker_config *)host->dbsync_worker :
725 - (struct aclk_database_worker_config *)find_inactive_wc_by_node_id(node_id);
735 + if (!wc)
736 + wc = (struct aclk_database_worker_config *)host->dbsync_worker ?
737 + (struct aclk_database_worker_config *)host->dbsync_worker :
738 + (struct aclk_database_worker_config *)find_inactive_wc_by_node_id(node_id);
739 + char *hostname = NULL;
740 if (likely(wc)) {
741 + if (wc->host)
742 + hostname = strdupz(wc->host->hostname);
743 + else
744 + hostname = get_hostname_by_node_id(node_id);
745 wc->chart_reset_count++;
746 __sync_synchronize();
747 wc->chart_updates = 0;
@@ -733,7 +751,7 @@ void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at
751 log_access(
752 "ACLK REQ [%s (%s)]: CHARTS STREAM from %" PRIu64 " t=%ld resets=%d",
753 wc->node_id,
736 - wc->host ? wc->host->hostname : "N/A",
754 + hostname ? hostname : "N/A",
755 wc->chart_sequence_id,
756 wc->chart_timestamp,
757 wc->chart_reset_count);
@@ -742,7 +760,7 @@ void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at
760 "ACLK RES [%s (%s)]: CHARTS FULL RESYNC REQUEST "
761 "remote_seq=%" PRIu64 " local_seq=%" PRIu64 " resets=%d ",
762 wc->node_id,
745 - wc->host ? wc->host->hostname : "N/A",
763 + hostname ? hostname : "N/A",
764 sequence_id,
765 wc->chart_sequence_id,
766 wc->chart_reset_count);
@@ -766,7 +784,7 @@ void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at
784 log_access(
785 "ACLK REQ [%s (%s)]: CHART RESET from %" PRIu64 " t=%ld batch=%" PRIu64,
786 wc->node_id,
769 - wc->host ? wc->host->hostname : "N/A",
787 + hostname ? hostname : "N/A",
788 wc->chart_sequence_id,
789 wc->chart_timestamp,
790 wc->batch_id);
@@ -775,20 +793,16 @@ void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at
793 cmd.completion = NULL;
794 aclk_database_enq_cmd(wc, &cmd);
795 } else {
778 -// log_access(
779 -// "ACLK RES [%s (%s)]: CHARTS STREAM from %" PRIu64
780 -// " t=%ld resets=%d",
781 -// wc->node_id,
782 -// wc->host ? wc->host->hostname : "N/A",
783 -// wc->chart_sequence_id,
784 -// wc->chart_timestamp,
785 -// wc->chart_reset_count);
796 wc->chart_reset_count = 0;
797 wc->chart_updates = 1;
798 }
799 }
790 - } else
791 - log_access("ACLK STA [%s (N/A)]: ACLK synchronization thread is not active.", node_id);
800 + freez(hostname);
801 + } else {
802 + hostname = get_hostname_by_node_id(node_id);
803 + log_access("ACLK STA [%s (%s)]: ACLK synchronization thread is not active.", node_id, hostname ? hostname : "N/A");
804 + freez(hostname);
805 + }
806 return;
807 }
808 host = host->next;
@@ -887,7 +901,10 @@ void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_d
901 time_t last_entry_t;
902 uint32_t update_every = 0;
903 uint32_t dimension_update_count = 0;
890 - int send_status;
904 + uint32_t total_checked = 0;
905 + uint32_t total_deleted= 0;
906 + uint32_t total_stopped= 0;
907 + time_t send_status;
908
909 struct retention_updated rotate_data;
910
@@ -942,23 +959,40 @@ void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_d
959 if (likely(!rc && first_entry_t))
960 start_time = MIN(start_time, first_entry_t);
961
945 - if (memory_mode == RRD_MEMORY_MODE_DBENGINE && wc->chart_updates) {
962 + if (memory_mode == RRD_MEMORY_MODE_DBENGINE && wc->chart_updates && (dimension_update_count < ACLK_MAX_DIMENSION_CLEANUP)) {
963 int live = ((now - last_entry_t) < (RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * update_every));
947 - if ((!live || !first_entry_t) && (dimension_update_count < ACLK_MAX_DIMENSION_CLEANUP)) {
948 - (void)aclk_upd_dimension_event(
949 - wc,
950 - claim_id,
951 - (uuid_t *)sqlite3_column_blob(res, 0),
952 - (const char *)(const char *)sqlite3_column_text(res, 3),
953 - (const char *)(const char *)sqlite3_column_text(res, 4),
954 - (const char *)(const char *)sqlite3_column_text(res, 2),
955 - first_entry_t,
956 - live ? 0 : last_entry_t,
957 - &send_status);
958 - if (!send_status)
964 + if (rc) {
965 + first_entry_t = 0;
966 + last_entry_t = 0;
967 + live = 0;
968 + }
969 + if (!wc->host || !first_entry_t) {
970 + if (!first_entry_t) {
971 + delete_dimension_uuid((uuid_t *)sqlite3_column_blob(res, 0));
972 + total_deleted++;
973 dimension_update_count++;
974 + }
975 + else {
976 + (void)aclk_upd_dimension_event(
977 + wc,
978 + claim_id,
979 + (uuid_t *)sqlite3_column_blob(res, 0),
980 + (const char *)(const char *)sqlite3_column_text(res, 3),
981 + (const char *)(const char *)sqlite3_column_text(res, 4),
982 + (const char *)(const char *)sqlite3_column_text(res, 2),
983 + first_entry_t,
984 + live ? 0 : last_entry_t,
985 + &send_status);
986 +
987 + if (!send_status) {
988 + if (last_entry_t)
989 + total_stopped++;
990 + dimension_update_count++;
991 + }
992 + }
993 }
994 }
995 + total_checked++;
996 }
997 if (update_every) {
998 debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
@@ -970,7 +1004,16 @@ void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_d
1004 rotate_data.interval_duration_count++;
1005 }
1006
1007 + char *hostname = NULL;
1008 + if (!wc->host)
1009 + hostname = get_hostname_by_node_id(wc->node_id);
1010 +
1011 + log_access("ACLK STA [%s (%s)]: UPDATES %d RETENTION MESSAGE SENT. CHECKED %u DIMENSIONS. %u DELETED, %u STOPPED COLLECTING",
1012 + wc->node_id, wc->host ? wc->host->hostname : hostname ? hostname : "N/A", wc->chart_updates, total_checked, total_deleted, total_stopped);
1013 + freez(hostname);
1014 +
1015 #ifdef NETDATA_INTERNAL_CHECKS
1016 + info("Retention update for %s (chart updates = %d)", wc->host_guid, wc->chart_updates);
1017 for (int i = 0; i < rotate_data.interval_duration_count; ++i)
1018 info(
1019 "Update for host %s (node %s) for %u Retention = %u",
@@ -1048,11 +1091,17 @@ void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)
1091 return;
1092 }
1093
1051 -int queue_dimension_to_aclk(RRDDIM *rd)
1094 +void queue_dimension_to_aclk(RRDDIM *rd)
1095 {
1096 + if (rrddim_flag_check(rd, RRDDIM_FLAG_ACLK))
1097 + return;
1098 +
1099 + rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
1100 int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker,
1101 rd, ACLK_DATABASE_ADD_DIMENSION);
1055 - return rc;
1102 + if (unlikely(rc))
1103 + rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK);
1104 + return;
1105 }
1106
1107 void aclk_send_dimension_update(RRDDIM *rd)
@@ -1203,6 +1252,12 @@ void sql_check_chart_liveness(RRDSET *st) {
1252 return;
1253
1254 rrdset_rdlock(st);
1255 +
1256 + if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1257 + rrdset_unlock(st);
1258 + return;
1259 + }
1260 +
1261 if (unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1262 if (likely(st->dimensions && st->counter_done && !queue_chart_to_aclk(st))) {
1263 debug(D_ACLK_SYNC,"Check chart liveness [%s] submit chart definition", st->name);
@@ -1218,13 +1273,8 @@ void sql_check_chart_liveness(RRDSET *st) {
1273 if (!rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)) {
1274 int live = (mark - rd->last_collected_time.tv_sec) < RRDSET_MINIMUM_DIM_LIVE_MULTIPLIER * rd->update_every;
1275 if (unlikely(live != rd->state->aclk_live_status)) {
1221 - if (likely(rrdset_flag_check(st, RRDSET_FLAG_ACLK))) {
1222 - if (likely(!queue_dimension_to_aclk(rd))) {
1223 - debug(D_ACLK_SYNC,"Dimension change [%s] on [%s] from live %d --> %d", rd->id, rd->rrdset->name, rd->state->aclk_live_status, live);
1224 - rd->state->aclk_live_status = live;
1225 - rrddim_flag_set(rd, RRDDIM_FLAG_ACLK);
1226 - }
1227 - }
1276 + debug(D_ACLK_SYNC,"Dimension change [%s] on [%s] from live %d --> %d", rd->id, rd->rrdset->name, rd->state->aclk_live_status, live);
1277 + queue_dimension_to_aclk(rd);
1278 }
1279 else
1280 debug(D_ACLK_SYNC,"Dimension check [%s] on [%s] liveness matches", rd->id, st->name);
database/sqlite/sqlite_aclk_chart.h
+1 -1
@@ -37,7 +37,7 @@ struct aclk_chart_sync_stats {
37 };
38
39 extern int queue_chart_to_aclk(RRDSET *st);
40 -extern int queue_dimension_to_aclk(RRDDIM *rd);
40 +extern void queue_dimension_to_aclk(RRDDIM *rd);
41 extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
42 int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
43 int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);