@cryptotaxi247 / netdata-1 / commits / ca6e7cd5a

Improve ACLK sync shutdown process (#19966)

* Refactor ACLK sync shutdown process * Mark all pending queries as cancelled Wait at most 5 seconds for queries to timeout before force stopping the ACLK sync thread * Refactor logging in ACLK synchronization process * Improve logging message for ACLK request snapshot creation

Stelios Fragkakis committed Mar 27, 2025 at 00:52 UTC ca6e7cd5a02f68058abca17b499292b598888e40

11 files changed +122 -76

src/aclk/aclk_query.c

+11

@@ -66,6 +66,17 @@ void pending_req_list_rm(const char *msg_id)

66 spinlock_unlock(&pending_req_list_lock);

67 }

69 +void mark_pending_req_cancel_all()

70 +{

71 + spinlock_lock(&pending_req_list_lock);

72 + struct pending_req_list *curr = pending_req_list_head;

73 + while (curr) {

74 + curr->canceled = 1;

75 + curr = curr->next;

76 + }

77 + spinlock_unlock(&pending_req_list_lock);

78 +}

79 +

80 int mark_pending_req_cancelled(const char *msg_id)

81 {

82 uint32_t hash = simple_hash(msg_id);

src/aclk/aclk_query.h

@@ -10,6 +10,7 @@

10 #include "aclk_query_queue.h"

12 int mark_pending_req_cancelled(const char *msg_id);

13 +void mark_pending_req_cancel_all();

15 void aclk_execute_query(aclk_query_t query);

16 void aclk_query_init(mqtt_wss_client client);

src/daemon/daemon-shutdown-watcher.c

+25 -39

@@ -112,15 +112,17 @@ void *watcher_main(void *arg)

112 usec_t shutdown_start_time = now_monotonic_usec();

113

114 watcher_wait_for_step(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS, shutdown_start_time);

115 - watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK, shutdown_start_time);

115 + watcher_wait_for_step(

116 + WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS, shutdown_start_time);

117 watcher_wait_for_step(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD, shutdown_start_time);

118 watcher_wait_for_step(WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS, shutdown_start_time);

119 watcher_wait_for_step(WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS, shutdown_start_time);

120 watcher_wait_for_step(WATCHER_STEP_ID_STOP_REPLICATION_THREADS, shutdown_start_time);

120 - watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS, shutdown_start_time);

121 + watcher_wait_for_step(WATCHER_STEP_ID_DISABLE_ML_DETEC_AND_TRAIN_THREADS, shutdown_start_time);

122 watcher_wait_for_step(WATCHER_STEP_ID_STOP_CONTEXT_THREAD, shutdown_start_time);

123 watcher_wait_for_step(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE, shutdown_start_time);

123 - watcher_wait_for_step(WATCHER_STEP_ID_STOP_ACLK_THREADS, shutdown_start_time);

124 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_ACLK_SYNC_THREAD, shutdown_start_time);

125 + watcher_wait_for_step(WATCHER_STEP_ID_STOP_ACLK_MQTT_THREAD, shutdown_start_time);

126 watcher_wait_for_step(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS, shutdown_start_time);

127 watcher_wait_for_step(WATCHER_STEP_ID_CANCEL_MAIN_THREADS, shutdown_start_time);

128 watcher_wait_for_step(WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN, shutdown_start_time);

@@ -145,46 +147,30 @@ void *watcher_main(void *arg)

147 void watcher_thread_start() {

148 watcher_steps = callocz(WATCHER_STEP_ID_MAX, sizeof(watcher_step_t));

149

148 - watcher_steps[WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS].msg =

149 - "close webrtc connections";

150 - watcher_steps[WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK].msg =

151 - "disable maintenance, new queries, new web requests, new streaming connections and aclk";

152 - watcher_steps[WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD].msg =

153 - "stop maintenance thread";

150 + watcher_steps[WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS].msg = "close webrtc connections";

151 + watcher_steps[WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS]

152 + .msg = "disable maintenance, new queries, new web requests, new streaming connections and aclk";

153 + watcher_steps[WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD].msg = "stop maintenance thread";

154 watcher_steps[WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS].msg =

155 "stop exporters, health and web servers threads";

156 - watcher_steps[WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS].msg =

157 - "stop collectors and streaming threads";

158 - watcher_steps[WATCHER_STEP_ID_STOP_REPLICATION_THREADS].msg =

159 - "stop replication threads";

160 - watcher_steps[WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN].msg =

161 - "prepare metasync shutdown";

162 - watcher_steps[WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS].msg =

163 - "disable ML detection and training threads";

164 - watcher_steps[WATCHER_STEP_ID_STOP_CONTEXT_THREAD].msg =

165 - "stop context thread";

166 - watcher_steps[WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE].msg =

167 - "clear web client cache";

168 - watcher_steps[WATCHER_STEP_ID_STOP_ACLK_THREADS].msg =

169 - "stop aclk threads";

170 - watcher_steps[WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS].msg =

171 - "stop all remaining worker threads";

172 - watcher_steps[WATCHER_STEP_ID_CANCEL_MAIN_THREADS].msg =

173 - "cancel main threads";

174 - watcher_steps[WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS].msg =

175 - "stop collection for all hosts";

156 + watcher_steps[WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS].msg = "stop collectors and streaming threads";

157 + watcher_steps[WATCHER_STEP_ID_STOP_REPLICATION_THREADS].msg = "stop replication threads";

158 + watcher_steps[WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN].msg = "prepare metasync shutdown";

159 + watcher_steps[WATCHER_STEP_ID_DISABLE_ML_DETEC_AND_TRAIN_THREADS].msg = "disable ML detection and training threads";

160 + watcher_steps[WATCHER_STEP_ID_STOP_CONTEXT_THREAD].msg = "stop context thread";

161 + watcher_steps[WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE].msg = "clear web client cache";

162 + watcher_steps[WATCHER_STEP_ID_STOP_ACLK_SYNC_THREAD].msg = "stop ACLK sync thread";

163 + watcher_steps[WATCHER_STEP_ID_STOP_ACLK_MQTT_THREAD].msg = "stop ACLK MQTT connection thread";

164 + watcher_steps[WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS].msg = "stop all remaining worker threads";

165 + watcher_steps[WATCHER_STEP_ID_CANCEL_MAIN_THREADS].msg = "cancel main threads";

166 + watcher_steps[WATCHER_STEP_ID_STOP_COLLECTION_FOR_ALL_HOSTS].msg = "stop collection for all hosts";

167 watcher_steps[WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH].msg =

168 "wait for dbengine collectors to finish";

178 - watcher_steps[WATCHER_STEP_ID_STOP_DBENGINE_TIERS].msg =

179 - "stop dbengine tiers";

180 - watcher_steps[WATCHER_STEP_ID_STOP_METASYNC_THREADS].msg =

181 - "stop metasync threads";

182 - watcher_steps[WATCHER_STEP_ID_CLOSE_SQL_DATABASES].msg =

183 - "close SQL databases";

184 - watcher_steps[WATCHER_STEP_ID_REMOVE_PID_FILE].msg =

185 - "remove pid file";

186 - watcher_steps[WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES].msg =

187 - "free openssl structures";

169 + watcher_steps[WATCHER_STEP_ID_STOP_DBENGINE_TIERS].msg = "stop dbengine tiers";

170 + watcher_steps[WATCHER_STEP_ID_STOP_METASYNC_THREADS].msg = "stop metasync threads";

171 + watcher_steps[WATCHER_STEP_ID_CLOSE_SQL_DATABASES].msg = "close SQL databases";

172 + watcher_steps[WATCHER_STEP_ID_REMOVE_PID_FILE].msg = "remove pid file";

173 + watcher_steps[WATCHER_STEP_ID_FREE_OPENSSL_STRUCTURES].msg = "free openssl structures";

174

175 for (size_t i = 0; i != WATCHER_STEP_ID_MAX; i++) {

176 completion_init(&watcher_steps[i].p);

src/daemon/daemon-shutdown-watcher.h

+4 -3

@@ -7,15 +7,16 @@

8 typedef enum {

9 WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS,

10 - WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK,

10 + WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS,

11 WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD,

12 WATCHER_STEP_ID_STOP_EXPORTERS_HEALTH_AND_WEB_SERVERS_THREADS,

13 WATCHER_STEP_ID_STOP_COLLECTORS_AND_STREAMING_THREADS,

14 WATCHER_STEP_ID_STOP_REPLICATION_THREADS,

15 - WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS,

15 + WATCHER_STEP_ID_DISABLE_ML_DETEC_AND_TRAIN_THREADS,

16 WATCHER_STEP_ID_STOP_CONTEXT_THREAD,

17 WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE,

18 - WATCHER_STEP_ID_STOP_ACLK_THREADS,

18 + WATCHER_STEP_ID_STOP_ACLK_SYNC_THREAD,

19 + WATCHER_STEP_ID_STOP_ACLK_MQTT_THREAD,

20 WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS,

21 WATCHER_STEP_ID_CANCEL_MAIN_THREADS,

22 WATCHER_STEP_ID_PREPARE_METASYNC_SHUTDOWN,

src/daemon/daemon-shutdown.c

+9 -4

@@ -207,8 +207,8 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

207 watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);

208

209 service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS |

210 - ABILITY_STREAMING_CONNECTIONS | SERVICE_ACLK | SERVICE_SYSTEMD);

211 - watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS_AND_ACLK);

210 + ABILITY_STREAMING_CONNECTIONS | SERVICE_SYSTEMD);

211 + watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS);

212

213 service_wait_exit(SERVICE_MAINTENANCE | SERVICE_SYSTEMD, 3 * USEC_PER_SEC);

214 watcher_step_complete(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD);

@@ -232,7 +232,7 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

232

233 ml_stop_threads();

234 ml_fini();

235 - watcher_step_complete(WATCHER_STEP_ID_DISABLE_ML_DETECTION_AND_TRAINING_THREADS);

235 + watcher_step_complete(WATCHER_STEP_ID_DISABLE_ML_DETEC_AND_TRAIN_THREADS);

236

237 service_wait_exit(SERVICE_CONTEXT, 3 * USEC_PER_SEC);

238 watcher_step_complete(WATCHER_STEP_ID_STOP_CONTEXT_THREAD);

@@ -240,8 +240,13 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

240 web_client_cache_destroy();

241 watcher_step_complete(WATCHER_STEP_ID_CLEAR_WEB_CLIENT_CACHE);

242

243 + aclk_synchronization_shutdown();

244 + watcher_step_complete(WATCHER_STEP_ID_STOP_ACLK_SYNC_THREAD);

245 +

246 + service_signal_exit(SERVICE_ACLK);

247 +

248 service_wait_exit(SERVICE_ACLK, 3 * USEC_PER_SEC);

244 - watcher_step_complete(WATCHER_STEP_ID_STOP_ACLK_THREADS);

249 + watcher_step_complete(WATCHER_STEP_ID_STOP_ACLK_MQTT_THREAD);

250

251 service_wait_exit(~0, 10 * USEC_PER_SEC);

252 watcher_step_complete(WATCHER_STEP_ID_STOP_ALL_REMAINING_WORKER_THREADS);

src/database/rrd.c

+1 -1

@@ -161,7 +161,7 @@ int rrd_init(const char *hostname, struct rrdhost_system_info *system_info, bool

161

162 if (likely(system_info)) {

163 detect_machine_guid_change(&localhost->host_id.uuid);

164 - sql_aclk_sync_init();

164 + aclk_synchronization_init();

165 api_v1_management_init();

166 }

167

src/database/sqlite/sqlite_aclk.c

+56 -26

@@ -594,13 +594,19 @@ static void free_query_list(Pvoid_t JudyL)

594 }

595 }

596

597 -static void aclk_synchronization(void *arg)

597 +#define MAX_SHUTDOWN_TIMEOUT_SECONDS (5)

598 +

599 +#define ACLK_SYNC_SHOULD_BE_RUNNING \

600 + (!shutdown_requested || config->aclk_queries_running || config->alert_push_running || \

601 + config->aclk_batch_job_is_running)

602 +

603 +static void aclk_synchronization_event_loop(void *arg)

604 {

605 struct aclk_sync_config_s *config = arg;

606 uv_thread_set_name_np("ACLKSYNC");

607 config->ar = aral_by_size_acquire(sizeof(struct aclk_database_cmd));

602 -

608 worker_register("ACLKSYNC");

609 +

610 service_register(SERVICE_THREAD_TYPE_EVENT_LOOP, NULL, NULL, NULL, true);

611

612 worker_register_job_name(ACLK_DATABASE_NOOP, "noop");

@@ -645,23 +651,31 @@ static void aclk_synchronization(void *arg)

651 unsigned cmd_batch_size;

652

653 completion_mark_complete(&config->start_stop_complete);

648 - while (likely(service_running(SERVICE_ACLK))) {

654 + int shutdown_requested = 0;

655 + time_t shutdown_initiated = 0;

656 +

657 + while (likely(ACLK_SYNC_SHOULD_BE_RUNNING)) {

658 enum aclk_database_opcode opcode;

659 worker_is_idle();

660 uv_run(loop, UV_RUN_DEFAULT);

661

662 + if (unlikely(shutdown_requested)) {

663 + nd_log_limit_static_thread_var(erl, 1, 0);

664 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO, "ACLKSYNC: Waiting for pending queries to finish before shutdown");

665 + if (now_realtime_sec() - shutdown_initiated > MAX_SHUTDOWN_TIMEOUT_SECONDS) {

666 + nd_log_daemon(NDLP_INFO, "ACLKSYNC: Shutdown timeout, forcing exit");

667 + break;

668 + }

669 + continue;

670 + }

671 +

672 /* wait for commands */

673 cmd_batch_size = 0;

674 do {

656 - if (unlikely(cmd_batch_size >= MAX_BATCH_SIZE))

675 + if (unlikely(++cmd_batch_size >= MAX_BATCH_SIZE))

676 break;

677

678 struct aclk_database_cmd cmd = aclk_database_deq_cmd();

660 -

661 - if (unlikely(!service_running(SERVICE_ACLK)))

662 - break;

663 -

664 - ++cmd_batch_size;

679 opcode = cmd.opcode;

680

681 if(likely(opcode != ACLK_DATABASE_NOOP && opcode != ACLK_QUERY_EXECUTE))

@@ -848,7 +862,11 @@ static void aclk_synchronization(void *arg)

862 config->aclk_batch_job_is_running = false;

863 }

864 break;

851 -

865 + case ACLK_SYNC_SHUTDOWN:

866 + shutdown_requested = 1;

867 + shutdown_initiated = now_realtime_sec();

868 + mark_pending_req_cancel_all();

869 + break;

870 default:

871 break;

872 }

@@ -880,19 +898,20 @@ static void aclk_synchronization(void *arg)

898 }

899

900 aral_by_size_release(config->ar);

901 + completion_mark_complete(&config->start_stop_complete);

902

903 worker_unregister();

904 service_exits();

905 netdata_log_info("ACLK SYNC: Shutting down ACLK synchronization event loop");

906 }

907

889 -static void aclk_synchronization_init(void)

908 +static void aclk_initialize_event_loop(void)

909 {

910 memset(&aclk_sync_config, 0, sizeof(aclk_sync_config));

911 completion_init(&aclk_sync_config.start_stop_complete);

912

913 int retries = 0;

895 - int create_uv_thread_rc = create_uv_thread(&aclk_sync_config.thread, aclk_synchronization, &aclk_sync_config, &retries);

914 + int create_uv_thread_rc = create_uv_thread(&aclk_sync_config.thread, aclk_synchronization_event_loop, &aclk_sync_config, &retries);

915 if (create_uv_thread_rc)

916 nd_log_daemon(NDLP_ERR, "Failed to create ACLK synchronization thread, error %s, after %d retries", uv_err_name(create_uv_thread_rc), retries);

917

@@ -900,9 +919,10 @@ static void aclk_synchronization_init(void)

919

920 if (retries)

921 nd_log_daemon(NDLP_WARNING, "ACLK synchronization thread was created after %d attempts", retries);

903 -

922 completion_wait_for(&aclk_sync_config.start_stop_complete);

905 - completion_destroy(&aclk_sync_config.start_stop_complete);

923 +

924 + // Keep completion, just reset it for next use during shutdown

925 + completion_reset(&aclk_sync_config.start_stop_complete);

926 }

927

928 // -------------------------------------------------------------

@@ -950,38 +970,37 @@ void destroy_aclk_config(RRDHOST *host)

970 "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni " \

971 "WHERE h.host_id = ni.host_id AND ni.node_id IS NOT NULL"

972

953 -void sql_aclk_sync_init(void)

973 +void aclk_synchronization_init(void)

974 {

975 char *err_msg = NULL;

976 int rc;

977

958 - REQUIRE_DB(db_meta);

959 -

960 - netdata_log_info("Creating archived hosts");

978 + nd_log_daemon(NDLP_INFO, "Creating archived hosts");

979 int number_of_children = 0;

980 rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_HOSTS, create_host_callback, &number_of_children, &err_msg);

981

982 if (rc != SQLITE_OK) {

965 - error_report("SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg);

983 + nd_log_daemon(NDLP_ERR, "SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg);

984 sqlite3_free(err_msg);

985 }

986

969 - netdata_log_info("Created %d archived hosts", number_of_children);

987 + nd_log_daemon(NDLP_INFO, "Created %d archived hosts", number_of_children);

988 // Trigger host context load for hosts that have been created

989 metadata_queue_load_host_context();

990

973 - if (!number_of_children)

974 - aclk_queue_node_info(localhost, true);

975 -

991 rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_INSTANCES, aclk_config_parameters, NULL, &err_msg);

992

993 if (rc != SQLITE_OK) {

979 - error_report("SQLite error when configuring host ACLK synchonization parameters, rc = %d (%s)", rc, err_msg);

994 + nd_log_daemon(NDLP_ERR, "SQLite error when configuring host ACLK synchonization parameters, rc = %d (%s)", rc, err_msg);

995 sqlite3_free(err_msg);

996 }

982 - aclk_synchronization_init();

997

984 - netdata_log_info("ACLK sync initialization completed");

998 + aclk_initialize_event_loop();

999 +

1000 + if (!number_of_children)

1001 + aclk_queue_node_info(localhost, true);

1002 +

1003 + nd_log_daemon(NDLP_INFO, "ACLK sync initialization completed");

1004 }

1005

1006 static inline void queue_aclk_sync_cmd(enum aclk_database_opcode opcode, const void *param0, const void *param1)

@@ -993,6 +1012,17 @@ static inline void queue_aclk_sync_cmd(enum aclk_database_opcode opcode, const v

1012 aclk_database_enq_cmd(&cmd);

1013 }

1014

1015 +void aclk_synchronization_shutdown(void)

1016 +{

1017 + // Send shutdown command, not that the completion is initialized

1018 + // on init and still valid

1019 + queue_aclk_sync_cmd(ACLK_SYNC_SHUTDOWN, NULL, NULL);

1020 +

1021 + completion_wait_for(&aclk_sync_config.start_stop_complete);

1022 + completion_destroy(&aclk_sync_config.start_stop_complete);

1023 + nd_log_daemon(NDLP_INFO, "ACLK sync shutdown completed");

1024 +}

1025 +

1026 // Public

1027 void aclk_push_alert_config(const char *node_id, const char *config_hash)

1028 {

src/database/sqlite/sqlite_aclk.h

+3 -1

@@ -26,6 +26,7 @@ enum aclk_database_opcode {

26 ACLK_QUERY_EXECUTE_SYNC,

27 ACLK_QUERY_BATCH_ADD,

28 ACLK_QUERY_BATCH_EXECUTE,

29 + ACLK_SYNC_SHUTDOWN,

31 // leave this last

32 // we need it to check for worker utilization

@@ -54,7 +55,8 @@ typedef struct aclk_sync_cfg_t {

56 void create_aclk_config(RRDHOST *host, nd_uuid_t *host_uuid, nd_uuid_t *node_id);

57 void destroy_aclk_config(RRDHOST *host);

57 -void sql_aclk_sync_init(void);

58 +void aclk_synchronization_init(void);

59 +void aclk_synchronization_shutdown(void);

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

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

62 void unregister_node(const char *machine_guid);

src/database/sqlite/sqlite_aclk_alert.c

+2 -2

@@ -1032,11 +1032,11 @@ void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)

1032 nd_log(

1033 NDLS_ACCESS,

1034 NDLP_DEBUG,

1035 - "ACLK REQ [%s (%s)]: Sent! %d alerts snapshot, snapshot_uuid %s (version = %llu)",

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

1036 wc->node_id,

1037 rrdhost_hostname(host),

1038 - cnt,

1038 snapshot_uuid,

1039 + total_count,

1040 (long long unsigned)version);

1041

1042 done:

src/libnetdata/completion/completion.c

@@ -2,6 +2,14 @@

3 #include "completion.h"

5 +ALWAYS_INLINE void completion_reset(struct completion *p)

6 +{

7 + if (!p)

8 + return;

9 + p->completed = 0;

10 + p->completed_jobs = 0;

11 +}

12 +

13 ALWAYS_INLINE void completion_init(struct completion *p)

14 {

15 p->completed = 0;

src/libnetdata/completion/completion.h

@@ -12,6 +12,8 @@ struct completion {

12 volatile unsigned completed_jobs;

13 };

15 +void completion_reset(struct completion *p);

16 +

17 void completion_init(struct completion *p);

19 void completion_destroy(struct completion *p);