@cryptotaxi247 / netdata-1 / commits / 0e73b7f06

ACLK synchronization improvements (#20077)

ACLK synchronization configuration handling and update related functions Add OPCODEs to properly create/stop/destroy ACLK host notification related timer

Stelios Fragkakis committed Apr 7, 2025 at 23:59 UTC 0e73b7f06c29adaad2f615aec8f2d00dc9895a24

11 files changed +207 -133

src/aclk/aclk.c

+15 -22

@@ -1035,20 +1035,20 @@ void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, con

1035

1036 static void fill_alert_status_for_host(BUFFER *wb, RRDHOST *host)

1037 {

1038 - struct aclk_sync_cfg_t *wc = host->aclk_config;

1039 - if (!wc)

1038 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

1039 + if (!aclk_host_config)

1040 return;

1041

1042 - buffer_sprintf(wb,

1042 + buffer_sprintf(

1043 + wb,

1044 "\n\t\tUpdates: %d"

1045 "\n\t\tCheckpoints: %d"

1046 "\n\t\tAlert count: %d"

1047 "\n\t\tAlert snapshot count: %d",

1047 - wc->stream_alerts,

1048 - wc->checkpoint_count,

1049 - wc->alert_count,

1050 - wc->snapshot_count

1051 - );

1048 + aclk_host_config->stream_alerts,

1049 + aclk_host_config->checkpoint_count,

1050 + aclk_host_config->alert_count,

1051 + aclk_host_config->snapshot_count);

1052 }

1053

1054 char *aclk_state(void)

@@ -1137,22 +1137,22 @@ char *aclk_state(void)

1137

1138 static void fill_alert_status_for_host_json(json_object *obj, RRDHOST *host)

1139 {

1140 - struct aclk_sync_cfg_t *wc = host->aclk_config;

1141 - if (!wc)

1140 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

1141 + if (!aclk_host_config)

1142 return;

1143

1144 - json_object *tmp = json_object_new_int(wc->stream_alerts);

1144 + json_object *tmp = json_object_new_int(aclk_host_config->stream_alerts);

1145 json_object_object_add(obj, "updates", tmp);

1146

1147 - tmp = json_object_new_int(wc->checkpoint_count);

1147 + tmp = json_object_new_int(aclk_host_config->checkpoint_count);

1148 json_object_object_add(obj, "checkpoint-count", tmp);

1149

1150 - tmp = json_object_new_int(wc->alert_count);

1150 + tmp = json_object_new_int(aclk_host_config->alert_count);

1151 json_object_object_add(obj, "alert-count", tmp);

1152

1153 - tmp = json_object_new_int(wc->snapshot_count);

1153 + tmp = json_object_new_int(aclk_host_config->snapshot_count);

1154 json_object_object_add(obj, "alert-snapshot-count", tmp);

1155 - tmp = json_object_new_int64(calculate_node_alert_version(wc->host));

1155 + tmp = json_object_new_int64(calculate_node_alert_version(aclk_host_config->host));

1156 json_object_object_add(obj, "alert-version", tmp);

1157 }

1158

@@ -1302,10 +1302,3 @@ void add_aclk_host_labels(void) {

1302 rrdlabels_add(labels, "_aclk_proxy", proxy_str, RRDLABEL_SRC_AUTO);

1303 rrdlabels_add(labels, "_aclk_ng_new_cloud_protocol", "true", RRDLABEL_SRC_AUTO|RRDLABEL_SRC_ACLK);

1304 }

1305 -

1306 -void aclk_queue_node_info(RRDHOST *host, bool immediate)

1307 -{

1308 - struct aclk_sync_cfg_t *wc = host->aclk_config;

1309 - if (wc)

1310 - wc->node_info_send_time = (host == localhost || immediate) ? 1 : now_realtime_sec();

1311 -}

src/aclk/aclk.h

-1

@@ -103,7 +103,6 @@ void aclk_send_bin_msg(char *msg, size_t msg_len, enum aclk_topics subtopic, con

103 char *aclk_state(void);

104 char *aclk_state_json(void);

105 void add_aclk_host_labels(void);

106 -void aclk_queue_node_info(RRDHOST *host, bool immediate);

106

107 struct mqtt_wss_stats aclk_statistics(void);

108 void aclk_status_set(ACLK_STATUS status);

src/database/rrdhost.h

+1 -1

@@ -253,7 +253,7 @@ struct rrdhost {

253

254 // ------------------------------------------------------------------------

255

256 - struct aclk_sync_cfg_t *aclk_config;

256 + struct aclk_sync_cfg_t *aclk_host_config;

257

258 // ------------------------------------------------------------------------

259 // health monitoring options

src/database/sqlite/sqlite_aclk.c

+107 -41

@@ -76,7 +76,7 @@ static struct aclk_database_cmd aclk_database_deq_cmd(void)

77 static void aclk_database_enq_cmd(struct aclk_database_cmd *cmd)

78 {

79 - if(unlikely(!aclk_sync_config.initialized))

79 + if(unlikely(!__atomic_load_n(&aclk_sync_config.initialized, __ATOMIC_RELAXED)))

80 return;

82 struct aclk_database_cmd *t = aral_mallocz(aclk_sync_config.ar);

@@ -505,6 +505,11 @@ struct worker_data {

505 struct aclk_sync_config_s *config;

506 };

507

508 +struct notify_timer_cb_data {

509 + void *payload;

510 + struct completion *completion;

511 +};

512 +

513 static void after_do_unregister_node(uv_work_t *req, int status __maybe_unused)

514 {

515 struct worker_data *data = req->data;

@@ -524,16 +529,25 @@ static void do_unregister_node(uv_work_t *req)

529 worker_is_idle();

530 }

531

532 +static void notify_timer_close_callback(uv_handle_t *handle)

533 +{

534 + struct notify_timer_cb_data *data = handle->data;

535 + if (data->completion) {

536 + completion_mark_complete(data->completion);

537 + }

538 + freez(data);

539 +}

540 +

541 static void node_update_timer_cb(uv_timer_t *handle)

542 {

529 - struct aclk_sync_cfg_t *ahc = handle->data;

530 - if (unlikely(!ahc))

543 + struct aclk_sync_cfg_t *aclk_host_config = handle->data;

544 + if (unlikely(!aclk_host_config))

545 return;

546

533 - RRDHOST *host = ahc->host;

547 + RRDHOST *host = aclk_host_config->host;

548

549 if(!host || aclk_host_state_update_auto(host))

536 - uv_timer_stop(&ahc->timer);

550 + uv_timer_stop(&aclk_host_config->timer);

551 }

552

553 static void after_start_alert_push(uv_work_t *req, int status __maybe_unused)

@@ -621,6 +635,8 @@ static void aclk_synchronization_event_loop(void *arg)

635 worker_register_job_name(ACLK_MQTT_WSS_CLIENT_SET, "config mqtt client");

636 worker_register_job_name(ACLK_MQTT_WSS_CLIENT_RESET, "reset mqtt client");

637 worker_register_job_name(ACLK_DATABASE_NODE_UNREGISTER, "unregister node");

638 + worker_register_job_name(ACLK_CANCEL_NODE_UPDATE_TIMER, "cancel node update timer");

639 + worker_register_job_name(ACLK_QUEUE_NODE_INFO, "queue node info");

640

641 uv_loop_t *loop = &config->loop;

642 fatal_assert(0 == uv_loop_init(loop));

@@ -640,6 +656,7 @@ static void aclk_synchronization_event_loop(void *arg)

656 netdata_log_info("Starting ACLK synchronization thread with %d parallel query threads", query_thread_count);

657

658 struct worker_data *data;

659 + struct notify_timer_cb_data *timer_cb_data;

660 aclk_query_t query;

661

662 // This holds queries that need to be executed one by one

@@ -659,6 +676,8 @@ static void aclk_synchronization_event_loop(void *arg)

676

677 while (likely(ACLK_SYNC_SHOULD_BE_RUNNING)) {

678 enum aclk_database_opcode opcode;

679 + RRDHOST *host;

680 + struct aclk_sync_cfg_t *aclk_host_config;

681 worker_is_idle();

682 uv_run(loop, UV_RUN_DEFAULT);

683

@@ -695,30 +714,30 @@ static void aclk_synchronization_event_loop(void *arg)

714 /* the command queue was empty, do nothing */

715 break;

716 // NODE STATE

698 - case ACLK_DATABASE_NODE_STATE:;

699 - RRDHOST *host = cmd.param[0];

700 - struct aclk_sync_cfg_t *ahc = host->aclk_config;

701 - if (unlikely(!ahc)) {

717 + case ACLK_DATABASE_NODE_STATE:

718 + host = cmd.param[0];

719 + aclk_host_config = host->aclk_host_config;

720 + if (unlikely(!aclk_host_config)) {

721 create_aclk_config(host, &host->host_id.uuid, &host->node_id.uuid);

703 - ahc = host->aclk_config;

722 + aclk_host_config = host->aclk_host_config;

723 }

724

706 - if (ahc) {

725 + if (aclk_host_config) {

726 uint64_t schedule_time = (uint64_t)(uintptr_t)cmd.param[1];

708 - if (!ahc->timer_initialized) {

709 - int rc = uv_timer_init(loop, &ahc->timer);

727 + if (!aclk_host_config->timer_initialized) {

728 + int rc = uv_timer_init(loop, &aclk_host_config->timer);

729 if (!rc) {

711 - ahc->timer_initialized = true;

712 - ahc->timer.data = ahc;

730 + aclk_host_config->timer_initialized = true;

731 + aclk_host_config->timer.data = aclk_host_config;

732 }

733 }

734

716 - if (ahc->timer_initialized) {

717 - if (uv_is_active((uv_handle_t *)&ahc->timer))

718 - uv_timer_stop(&ahc->timer);

735 + if (aclk_host_config->timer_initialized) {

736 + if (uv_is_active((uv_handle_t *)&aclk_host_config->timer))

737 + uv_timer_stop(&aclk_host_config->timer);

738

720 - ahc->timer.data = ahc;

721 - int rc = uv_timer_start(&ahc->timer, node_update_timer_cb, schedule_time, 5000);

739 + aclk_host_config->timer.data = aclk_host_config;

740 + int rc = uv_timer_start(&aclk_host_config->timer, node_update_timer_cb, schedule_time, 5000);

741 if (!rc)

742 break; // Timer started, exit

743 }

@@ -727,6 +746,31 @@ static void aclk_synchronization_event_loop(void *arg)

746 // This is fallback if timer fails

747 aclk_host_state_update_auto(host);

748 break;

749 + case ACLK_QUEUE_NODE_INFO:

750 + host = cmd.param[0];

751 + bool immediate = (bool)(uintptr_t)cmd.param[1];

752 + aclk_host_config = host->aclk_host_config;

753 + if (unlikely(!aclk_host_config)) {

754 + create_aclk_config(host, &host->host_id.uuid, &host->node_id.uuid);

755 + aclk_host_config = host->aclk_host_config;

756 + }

757 + aclk_host_config->node_info_send_time = (host == localhost ||(void *)(uintptr_t) immediate) ? 1 : now_realtime_sec();

758 + break;

759 + case ACLK_CANCEL_NODE_UPDATE_TIMER:

760 + host = cmd.param[0];

761 + aclk_host_config = host->aclk_host_config;

762 + if (!aclk_host_config || !aclk_host_config->timer_initialized)

763 + break;

764 + if (uv_is_active((uv_handle_t *)&aclk_host_config->timer))

765 + uv_timer_stop(&aclk_host_config->timer);

766 +

767 + aclk_host_config->timer_initialized = false;

768 + timer_cb_data = mallocz(sizeof(*data));

769 + timer_cb_data->payload = host;

770 + timer_cb_data->completion = (struct completion *)cmd.param[1];

771 + aclk_host_config->timer.data = data;

772 + uv_close((uv_handle_t *)&aclk_host_config->timer, notify_timer_close_callback);

773 + break;

774

775 case ACLK_DATABASE_NODE_UNREGISTER:

776 data = mallocz(sizeof(*data));

@@ -739,7 +783,6 @@ static void aclk_synchronization_event_loop(void *arg)

783 freez(data);

784 }

785 break;

742 - // ALERTS

786 case ACLK_DATABASE_PUSH_ALERT_CONFIG:

787 aclk_push_alert_config_event(cmd.param[0], cmd.param[1]);

788 break;

@@ -885,7 +928,7 @@ static void aclk_synchronization_event_loop(void *arg)

928 uv_close((uv_handle_t *)&config->timer_req, NULL);

929

930 uv_close((uv_handle_t *)&config->async, NULL);

888 - uv_walk(loop, libuv_close_callback, NULL);

931 + uv_walk(loop, libuv_close_callback, notify_timer_close_callback);

932 uv_run(loop, UV_RUN_NOWAIT);

933

934 (void) uv_loop_close(loop);

@@ -932,34 +975,30 @@ static void aclk_initialize_event_loop(void)

975

976 // -------------------------------------------------------------

977

935 -void create_aclk_config(RRDHOST *host __maybe_unused, nd_uuid_t *host_uuid __maybe_unused, nd_uuid_t *node_id __maybe_unused)

978 +void create_aclk_config(RRDHOST *host, nd_uuid_t *host_uuid __maybe_unused, nd_uuid_t *node_id __maybe_unused)

979 {

980

938 - if (!host || host->aclk_config)

981 + if (!host || host->aclk_host_config)

982 return;

983

941 - struct aclk_sync_cfg_t *wc = callocz(1, sizeof(struct aclk_sync_cfg_t));

984 + struct aclk_sync_cfg_t *aclk_host_config = callocz(1, sizeof(struct aclk_sync_cfg_t));

985 if (node_id && !uuid_is_null(*node_id))

943 - uuid_unparse_lower(*node_id, wc->node_id);

986 + uuid_unparse_lower(*node_id, aclk_host_config->node_id);

987

945 - host->aclk_config = wc;

946 - if (node_id && UUIDiszero(host->node_id)) {

947 - uuid_copy(host->node_id.uuid, *node_id);

988 + struct aclk_sync_cfg_t *expected = NULL;

989 + if (__atomic_compare_exchange_n(&host->aclk_host_config, &expected, aclk_host_config, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) {

990 + if (node_id && UUIDiszero(host->node_id))

991 + uuid_copy(host->node_id.uuid, *node_id);

992 }

949 -

950 - wc->host = host;

951 - wc->stream_alerts = false;

952 - time_t now = now_realtime_sec();

953 - wc->node_info_send_time = (host == localhost || NULL == localhost) ? now - 25 : now;

954 -}

955 -

956 -void destroy_aclk_config(RRDHOST *host)

957 -{

958 - if (!host || !host->aclk_config)

993 + else {

994 + freez(aclk_host_config);

995 return;

996 + }

997

961 - freez(host->aclk_config);

962 - host->aclk_config = NULL;

998 + aclk_host_config->host = host;

999 + aclk_host_config->stream_alerts = false;

1000 + time_t now = now_realtime_sec();

1001 + aclk_host_config->node_info_send_time = (host == localhost || NULL == localhost) ? now - 25 : now;

1002 }

1003

1004 #define SQL_FETCH_ALL_HOSTS \

@@ -1088,3 +1127,30 @@ void unregister_node(const char *machine_guid)

1127 return;

1128 queue_aclk_sync_cmd(ACLK_DATABASE_NODE_UNREGISTER, strdupz(machine_guid), NULL);

1129 }

1130 +

1131 +void destroy_aclk_config(RRDHOST *host)

1132 +{

1133 + if (!host)

1134 + return;

1135 +

1136 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

1137 + if (!aclk_host_config)

1138 + return;

1139 +

1140 + if(likely(__atomic_load_n(&aclk_sync_config.initialized, __ATOMIC_RELAXED))) {

1141 + struct completion compl;

1142 + completion_init(&compl);

1143 +

1144 + queue_aclk_sync_cmd(ACLK_CANCEL_NODE_UPDATE_TIMER, (void *)host, (void *)&compl);

1145 + completion_wait_for(&compl);

1146 + completion_destroy(&compl);

1147 + }

1148 +

1149 + struct aclk_sync_cfg_t *old_aclk_host_config = __atomic_exchange_n(&host->aclk_host_config, NULL, __ATOMIC_RELAXED);

1150 + freez(old_aclk_host_config);

1151 +}

1152 +

1153 +void aclk_queue_node_info(RRDHOST *host, bool immediate)

1154 +{

1155 + queue_aclk_sync_cmd(ACLK_QUEUE_NODE_INFO, (void *)host, (void *)(uintptr_t)immediate);

1156 +}

src/database/sqlite/sqlite_aclk.h

@@ -23,6 +23,9 @@ enum aclk_database_opcode {

23 ACLK_DATABASE_NODE_UNREGISTER,

24 ACLK_MQTT_WSS_CLIENT_SET,

25 ACLK_MQTT_WSS_CLIENT_RESET,

26 + ACLK_CANCEL_NODE_UPDATE_TIMER,

27 + ACLK_QUEUE_NODE_INFO,

28 + ACLK_MQTT_WSS_CLIENT,

29 ACLK_QUERY_EXECUTE,

30 ACLK_QUERY_EXECUTE_SYNC,

31 ACLK_QUERY_BATCH_ADD,

@@ -61,5 +64,7 @@ void aclk_synchronization_shutdown(void);

64 void aclk_push_alert_config(const char *node_id, const char *config_hash);

65 void schedule_node_state_update(RRDHOST *host, uint64_t delay);

66 void unregister_node(const char *machine_guid);

67 +void cancel_node_update_timer(const RRDHOST *host, struct completion *completion);

68 +void aclk_queue_node_info(RRDHOST *host, bool immediate);

70 #endif //NETDATA_SQLITE_ACLK_H

src/database/sqlite/sqlite_aclk_alert.c

+50 -43

@@ -504,11 +504,11 @@ static void aclk_push_alert_event(RRDHOST *host, sqlite3_stmt **res, sqlite3_stm

504

505 param = 0;

506 RRDCALC_STATUS status;

507 - struct aclk_sync_cfg_t *wc = host->aclk_config;

507 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

508 while (sqlite3_step_monitored(*res) == SQLITE_ROW) {

509 health_alarm_log_populate(&alarm_log, *res, host, &status);

510 aclk_send_alarm_log_entry(&alarm_log);

511 - wc->alert_count++;

511 + aclk_host_config->alert_count++;

512

513 last_id = alarm_log.sequence_id;

514 if (first_id == 0)

@@ -651,7 +651,8 @@ bool process_alert_pending_queue(RRDHOST *host)

651 RRDCALC_STATUS new_status = sqlite3_column_int(res, 2);

652 int64_t row = sqlite3_column_int64(res, 3);

653

654 - if (host->aclk_config) {

654 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

655 + if (aclk_host_config) {

656 int ret = insert_alert_to_submit_queue(host, health_log_id, unique_id, new_status);

657 if (ret == 0)

658 added++;

@@ -686,23 +687,23 @@ void aclk_push_alert_events_for_all_hosts(void)

687

688 rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

689

689 - struct aclk_sync_cfg_t *wc = host->aclk_config;

690 - if (!wc || false == wc->stream_alerts || rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {

690 + struct aclk_sync_cfg_t *aclk_host_config = host->aclk_host_config;

691 + if (!aclk_host_config || false == aclk_host_config->stream_alerts || rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {

692 (void)process_alert_pending_queue(host);

693 commit_alert_events(host);

694 continue;

695 }

696

696 - if (wc->send_snapshot) {

697 + if (aclk_host_config->send_snapshot) {

698 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

698 - if (wc->send_snapshot == 1)

699 + if (aclk_host_config->send_snapshot == 1)

700 continue;

701 (void)process_alert_pending_queue(host);

702 commit_alert_events(host);

703 rebuild_host_alert_version_table(host);

704 send_alert_snapshot_to_cloud(host);

704 - wc->snapshot_count++;

705 - wc->send_snapshot = 0;

705 + aclk_host_config->snapshot_count++;

706 + aclk_host_config->send_snapshot = 0;

707 }

708 else

709 aclk_push_alert_event(host, &res, &res_version);

@@ -717,18 +718,18 @@ void aclk_send_alert_configuration(char *config_hash)

718 if (unlikely(!config_hash))

719 return;

720

720 - struct aclk_sync_cfg_t *wc = localhost->aclk_config;

721 + struct aclk_sync_cfg_t *aclk_host_config = localhost->aclk_host_config;

722

722 - if (unlikely(!wc))

723 + if (unlikely(!aclk_host_config))

724 return;

725

726 nd_log(NDLS_ACCESS, NDLP_DEBUG,

727 "ACLK REQ [%s (%s)]: Request to send alert config %s.",

727 - wc->node_id,

728 - wc->host ? rrdhost_hostname(wc->host) : "N/A",

728 + aclk_host_config->node_id,

729 + aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A",

730 config_hash);

731

731 - aclk_push_alert_config(wc->node_id, config_hash);

732 + aclk_push_alert_config(aclk_host_config->node_id, config_hash);

733 }

734

735 #define SQL_SELECT_ALERT_CONFIG \

@@ -740,11 +741,11 @@ void aclk_send_alert_configuration(char *config_hash)

741 void aclk_push_alert_config_event(char *node_id __maybe_unused, char *config_hash __maybe_unused)

742 {

743 sqlite3_stmt *res = NULL;

743 - struct aclk_sync_cfg_t *wc;

744 + struct aclk_sync_cfg_t *aclk_host_config;

745

746 RRDHOST *host = rrdhost_find_by_node_id(node_id);

747

747 - if (unlikely(!host || !(wc = host->aclk_config))) {

748 + if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED))) {

749 freez(config_hash);

750 freez(node_id);

751 return;

@@ -827,13 +828,17 @@ void aclk_push_alert_config_event(char *node_id __maybe_unused, char *config_has

828 param = 0;

829

830 if (likely(p_alarm_config.cfg_hash)) {

830 - nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK RES [%s (%s)]: Sent alert config %s.", wc->node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A", config_hash);

831 + nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK RES [%s (%s)]: Sent alert config %s.",

832 + aclk_host_config->node_id,

833 + aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A", config_hash);

834 aclk_send_provide_alarm_cfg(&p_alarm_config);

835 freez(p_alarm_config.cfg_hash);

836 destroy_aclk_alarm_configuration(&alarm_config);

837 }

838 else

836 - nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK STA [%s (%s)]: Alert config for %s not found.", wc->node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A", config_hash);

839 + nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK STA [%s (%s)]: Alert config for %s not found.",

840 + aclk_host_config->node_id,

841 + aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A", config_hash);

842

843 done:

844 REPORT_BIND_FAIL(res, param);

@@ -868,42 +873,42 @@ done:

873 return version;

874 }

875

871 -static void schedule_alert_snapshot_if_needed(struct aclk_sync_cfg_t *wc, uint64_t cloud_version)

876 +static void schedule_alert_snapshot_if_needed(struct aclk_sync_cfg_t *aclk_host_config, uint64_t cloud_version)

877 {

878 if (cloud_version == 1) {

879 nd_log(

880 NDLS_ACCESS,

881 NDLP_NOTICE,

882 "Cloud requested to skip alert version verification for host \"%s\", node \"%s\"",

878 - rrdhost_hostname(wc->host),

879 - wc->node_id);

883 + rrdhost_hostname(aclk_host_config->host),

884 + aclk_host_config->node_id);

885 return;

886 }

887

883 - uint64_t local_version = calculate_node_alert_version(wc->host);

888 + uint64_t local_version = calculate_node_alert_version(aclk_host_config->host);

889 if (local_version != cloud_version) {

890 nd_log(

891 NDLS_ACCESS,

892 NDLP_NOTICE,

893 "Scheduling alert snapshot for host \"%s\", node \"%s\" (version: cloud %llu, local %llu)",

889 - rrdhost_hostname(wc->host),

890 - wc->node_id,

894 + rrdhost_hostname(aclk_host_config->host),

895 + aclk_host_config->node_id,

896 (long long unsigned)cloud_version,

897 (long long unsigned)local_version);

898

894 - wc->send_snapshot = 1;

895 - rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

899 + aclk_host_config->send_snapshot = 1;

900 + rrdhost_flag_set(aclk_host_config->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

901 }

902 else

903 nd_log(

904 NDLS_ACCESS,

905 NDLP_DEBUG,

906 "Alert check on \"%s\", node \"%s\" (version: cloud %llu, local %llu)",

902 - rrdhost_hostname(wc->host),

903 - wc->node_id,

907 + rrdhost_hostname(aclk_host_config->host),

908 + aclk_host_config->node_id,

909 (unsigned long long)cloud_version,

910 (unsigned long long)local_version);

906 - wc->checkpoint_count++;

911 + aclk_host_config->checkpoint_count++;

912 }

913

914 #define SQL_COUNT_SNAPSHOT_ENTRIES \

@@ -950,10 +955,10 @@ done:

955 #define ALARM_EVENTS_PER_CHUNK 1000

956 void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)

957 {

953 - struct aclk_sync_cfg_t *wc = host->aclk_config;

958 + struct aclk_sync_cfg_t *aclk_host_config = host->aclk_host_config;

959

960 if (unlikely(!host)) {

956 - nd_log(NDLS_ACCESS, NDLP_WARNING, "AC [%s (N/A)]: Node id not found", wc->node_id);

961 + nd_log(NDLS_ACCESS, NDLP_WARNING, "AC [%s (N/A)]: Node id not found", aclk_host_config->node_id);

962 return;

963 }

964

@@ -980,7 +985,8 @@ void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)

985 char *snapshot_uuid = &snapshot_uuid_str[0];

986

987 nd_log(NDLS_ACCESS, NDLP_DEBUG,

983 - "ACLK REQ [%s (%s)]: Sending %d alerts snapshot, snapshot_uuid %s", wc->node_id, rrdhost_hostname(host),

988 + "ACLK REQ [%s (%s)]: Sending %d alerts snapshot, snapshot_uuid %s",

989 + aclk_host_config->node_id, rrdhost_hostname(host),

990 cnt, snapshot_uuid);

991

992 uint32_t chunks;

@@ -990,13 +996,13 @@ void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)

996 struct alarm_snapshot alarm_snap;

997 struct alarm_log_entry alarm_log;

998

993 - alarm_snap.node_id = wc->node_id;

999 + alarm_snap.node_id = aclk_host_config->node_id;

1000 alarm_snap.claim_id = claim_id.str;

1001 alarm_snap.snapshot_uuid = snapshot_uuid;

1002 alarm_snap.chunks = chunks;

1003 alarm_snap.chunk = 1;

1004

999 - alarm_log.node_id = wc->node_id;

1005 + alarm_log.node_id = aclk_host_config->node_id;

1006 alarm_log.claim_id = claim_id.str;

1007

1008 cnt = 0;

@@ -1033,7 +1039,7 @@ void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)

1039 NDLS_ACCESS,

1040 NDLP_DEBUG,

1041 "ACLK REQ [%s (%s)]: Created snapshot %s with %d alerts (version = %llu)",

1036 - wc->node_id,

1042 + aclk_host_config->node_id,

1043 rrdhost_hostname(host),

1044 snapshot_uuid,

1045 total_count,

@@ -1052,10 +1058,10 @@ void aclk_start_alert_streaming(char *node_id, uint64_t cloud_version)

1058 if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))

1059 return;

1060

1055 - struct aclk_sync_cfg_t *wc;

1061 + struct aclk_sync_cfg_t *aclk_host_config;

1062 RRDHOST *host = rrdhost_find_by_node_id(node_id);

1063

1058 - if (unlikely(!host || !(wc = host->aclk_config))) {

1064 + if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED))) {

1065 nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, invalid node.", node_id);

1066 return;

1067 }

@@ -1065,9 +1071,10 @@ void aclk_start_alert_streaming(char *node_id, uint64_t cloud_version)

1071 return;

1072 }

1073

1068 - nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");

1069 - schedule_alert_snapshot_if_needed(wc, cloud_version);

1070 - wc->stream_alerts = true;

1074 + nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id,

1075 + aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A");

1076 + schedule_alert_snapshot_if_needed(aclk_host_config, cloud_version);

1077 + aclk_host_config->stream_alerts = true;

1078 }

1079

1080 // Do checkpoint alert version check

@@ -1086,13 +1093,13 @@ void aclk_alert_version_check(char *node_id, char *claim_id, uint64_t cloud_vers

1093 return;

1094 }

1095

1089 - struct aclk_sync_cfg_t *wc;

1096 + struct aclk_sync_cfg_t *aclk_host_config;

1097 RRDHOST *host = rrdhost_find_by_node_id(node_id);

1098

1092 - if ((!host || !(wc = host->aclk_config)))

1099 + if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED)))

1100 nd_log(NDLS_ACCESS, NDLP_NOTICE,

1101 "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT VALIDATION REQUEST RECEIVED FOR INVALID NODE",

1102 node_id);

1103 else

1097 - schedule_alert_snapshot_if_needed(wc, cloud_version);

1104 + schedule_alert_snapshot_if_needed(aclk_host_config, cloud_version);

1105 }

src/database/sqlite/sqlite_aclk_node.c

+17 -15

@@ -25,13 +25,13 @@ DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) {

26 static void build_node_collectors(RRDHOST *host)

27 {

28 - struct aclk_sync_cfg_t *wc = host->aclk_config;

28 + struct aclk_sync_cfg_t *aclk_host_config = host->aclk_host_config;

30 struct update_node_collectors upd_node_collectors;

31 DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);

33 CLAIM_ID claim_id = claim_id_get();

34 - upd_node_collectors.node_id = wc->node_id;

34 + upd_node_collectors.node_id = aclk_host_config->node_id;

35 upd_node_collectors.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL;

37 upd_node_collectors.node_collectors = collectors_from_charts(host, dict);

@@ -41,19 +41,19 @@ static void build_node_collectors(RRDHOST *host)

42 nd_log(NDLS_ACCESS, NDLP_DEBUG,

43 "ACLK RES [%s (%s)]: NODE COLLECTORS SENT",

44 - wc->node_id, rrdhost_hostname(host));

44 + aclk_host_config->node_id, rrdhost_hostname(host));

45 }

47 static void build_node_info(RRDHOST *host)

48 {

49 struct update_node_info node_info;

51 - struct aclk_sync_cfg_t *wc = host->aclk_config;

51 + struct aclk_sync_cfg_t *aclk_host_config = host->aclk_host_config;

53 CLAIM_ID claim_id = claim_id_get();

55 rrd_rdlock();

56 - node_info.node_id = wc->node_id;

56 + node_info.node_id = aclk_host_config->node_id;

57 node_info.claim_id = claim_id_is_set(claim_id) ? claim_id.str : NULL;

58 node_info.machine_guid = host->machine_guid;

59 node_info.child = (host != localhost);

@@ -85,7 +85,7 @@ static void build_node_info(RRDHOST *host)

85 NDLS_ACCESS,

86 NDLP_DEBUG,

87 "ACLK RES [%s (%s)]: NODE INFO SENT for guid [%s] (%s)",

88 - wc->node_id,

88 + aclk_host_config->node_id,

89 rrdhost_hostname(host),

90 host->machine_guid,

91 host == localhost ? "parent" : "child");

@@ -94,7 +94,7 @@ static void build_node_info(RRDHOST *host)

94 freez(node_info.node_instance_capabilities);

95 freez(host_version);

97 - wc->node_collectors_send = now_realtime_sec();

97 + aclk_host_config->node_collectors_send = now_realtime_sec();

98 }

100 void aclk_check_node_info_and_collectors(void)

@@ -121,8 +121,8 @@ void aclk_check_node_info_and_collectors(void)

121 time_t now = now_realtime_sec();

122 dfe_start_reentrant(rrdhost_root_index, host)

123 {

124 - struct aclk_sync_cfg_t *wc = host->aclk_config;

125 - if (unlikely(!wc))

124 + struct aclk_sync_cfg_t *aclk_host_config = host->aclk_host_config;

125 + if (unlikely(!aclk_host_config))

126 continue;

127

128 if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) {

@@ -132,7 +132,7 @@ void aclk_check_node_info_and_collectors(void)

132 continue;

133 }

134

135 - if (!wc->node_info_send_time && !wc->node_collectors_send)

135 + if (!aclk_host_config->node_info_send_time && !aclk_host_config->node_collectors_send)

136 continue;

137

138 if (unlikely(rrdhost_receiver_replicating_charts(host))) {

@@ -156,22 +156,24 @@ void aclk_check_node_info_and_collectors(void)

156

157 bool pp_queue_empty = !(host->rrdctx.pp_queue && dictionary_entries(host->rrdctx.pp_queue));

158

159 - if (!pp_queue_empty && (wc->node_info_send_time || wc->node_collectors_send)) {

159 + if (!pp_queue_empty && (aclk_host_config->node_info_send_time || aclk_host_config->node_collectors_send)) {

160 context_pp++;

161 context_pp_host = host->hostname;

162 }

163

164 - if (pp_queue_empty && wc->node_info_send_time && wc->node_info_send_time + 30 < now) {

165 - wc->node_info_send_time = 0;

164 + if (pp_queue_empty && aclk_host_config->node_info_send_time &&

165 + aclk_host_config->node_info_send_time + 30 < now) {

166 + aclk_host_config->node_info_send_time = 0;

167 build_node_info(host);

168 schedule_node_state_update(host, 10000);

169 internal_error(true, "ACLK SYNC: Sending node info for %s", rrdhost_hostname(host));

170 }

171

171 - if (pp_queue_empty && wc->node_collectors_send && wc->node_collectors_send + 30 < now) {

172 + if (pp_queue_empty && aclk_host_config->node_collectors_send &&

173 + aclk_host_config->node_collectors_send + 30 < now) {

174 build_node_collectors(host);

175 internal_error(true, "ACLK SYNC: Sending collectors for %s", rrdhost_hostname(host));

174 - wc->node_collectors_send = 0;

176 + aclk_host_config->node_collectors_send = 0;

177 }

178 }

179 dfe_done(host);

src/database/sqlite/sqlite_health.c

+2 -1

@@ -183,7 +183,8 @@ static void insert_alert_queue(

183

184 int rc;

185

186 - if (!host->aclk_config)

186 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

187 + if (!aclk_host_config)

188 return;

189

190 time_t submit_delay = trigger_time + calculate_delay(old_status, new_status);

src/database/sqlite/sqlite_metadata.c

+3 -3

@@ -279,14 +279,14 @@ static inline void set_host_node_id(RRDHOST *host, nd_uuid_t *node_id)

279 return;

280 }

281

282 - struct aclk_sync_cfg_t *wc = host->aclk_config;

282 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

283

284 uuid_copy(host->node_id.uuid, *node_id);

285

286 - if (unlikely(!wc))

286 + if (unlikely(!aclk_host_config))

287 create_aclk_config(host, &host->host_id.uuid, node_id);

288 else

289 - uuid_unparse_lower(*node_id, wc->node_id);

289 + uuid_unparse_lower(*node_id, aclk_host_config->node_id);

290

291 stream_receiver_send_node_and_claim_id_to_child(host);

292 stream_path_node_id_updated(host);

src/health/health_event_loop.c

+5 -5

@@ -283,8 +283,8 @@ static void health_event_loop_for_host(RRDHOST *host, bool apply_hibernation_del

283

284 worker_is_busy(WORKER_HEALTH_JOB_HOST_LOCK);

285 {

286 - struct aclk_sync_cfg_t *wc = host->aclk_config;

287 - if (wc && wc->send_snapshot == 2)

286 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

287 + if (aclk_host_config && aclk_host_config->send_snapshot == 2)

288 return;

289 }

290

@@ -634,9 +634,9 @@ static void health_event_loop_for_host(RRDHOST *host, bool apply_hibernation_del

634 commit_alert_transitions(host);

635

636 if (!__atomic_load_n(&host->health.pending_transitions, __ATOMIC_RELAXED)) {

637 - struct aclk_sync_cfg_t *wc = host->aclk_config;

638 - if (wc && wc->send_snapshot == 1) {

639 - wc->send_snapshot = 2;

637 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

638 + if (aclk_host_config && aclk_host_config->send_snapshot == 1) {

639 + aclk_host_config->send_snapshot = 2;

640 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);

641 } else {

642 worker_is_busy(WORKER_HEALTH_JOB_ALARM_LOG_QUEUE);

src/plugins.d/pluginsd_parser.c

+2 -1

@@ -245,7 +245,8 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si

245

246 rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);

247 rrdcontext_host_child_connected(host);

248 - if (host->aclk_config)

248 + struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);

249 + if (aclk_host_config)

250 aclk_queue_node_info(host, true);

251 else

252 schedule_node_state_update(host, 100);