@cryptotaxi247 / netdata / commits / 1bf5dbee6

Additional checks on metasync thread shutdown (#20455)

* update metadata scan to use correct worker status * Avoid scheduling additional metadata scan if we timed out * Improve shutdown handling for ACLK synchronization Unregister metasync service thread

Stelios Fragkakis committed Jun 17, 2025 at 16:38 UTC 1bf5dbee6619308254121e25f8903025cc2d501f

2 files changed +34 -33

src/database/sqlite/sqlite_aclk.c

+20 -25

@@ -41,6 +41,7 @@ struct aclk_sync_config_s {

41 uv_timer_t timer_req;

42 uv_async_t async;

43 bool initialized;

44 + bool shutdown_requested;

45 mqtt_wss_client client;

46 int aclk_queries_running;

47 bool run_query_batch;

@@ -587,11 +588,11 @@ static void timer_cb(uv_timer_t *handle)

588 }

589

590 #define MAX_SHUTDOWN_TIMEOUT_SECONDS (5)

591 +#define SHUTDOWN_SLEEP_INTERVAL_MS (100)

592 #define CMD_POOL_SIZE (2048)

593

592 -#define ACLK_SYNC_SHOULD_BE_RUNNING \

593 - (!shutdown_requested || config->aclk_queries_running || config->alert_push_running || \

594 - config->aclk_batch_job_is_running)

594 +#define ACLK_JOBS_ARE_RUNNING \

595 + (config->aclk_queries_running || config->alert_push_running || config->aclk_batch_job_is_running)

596

597 static void *aclk_synchronization_event_loop(void *arg)

598 {

@@ -624,18 +625,16 @@ static void *aclk_synchronization_event_loop(void *arg)

625

626 netdata_log_info("Starting ACLK synchronization thread");

627

627 - config->initialized = true;

628 -

628 sql_delete_aclk_table_list();

629

630 int query_thread_count = (int) netdata_conf_cloud_query_threads();

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

632

633 struct notify_timer_cb_data *timer_cb_data;

635 - aclk_query_t *query;

634

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

636 struct judy_list_t *aclk_query_batch = NULL;

637 +

638 // This holds queries that can be dispatched in parallel in ACLK QUERY worker threads

639 struct judy_list_t *aclk_query_execute = callocz(1, sizeof(*aclk_query_execute));

640 size_t pending_queries = 0;

@@ -645,27 +644,18 @@ static void *aclk_synchronization_event_loop(void *arg)

644

645 unsigned cmd_batch_size;

646

647 + config->shutdown_requested = false;

648 + config->initialized = true;

649 completion_mark_complete(&config->start_stop_complete);

649 - int shutdown_requested = 0;

650 - time_t shutdown_initiated = 0;

650

652 - while (likely(ACLK_SYNC_SHOULD_BE_RUNNING)) {

651 + while (likely(config->shutdown_requested == false)) {

652 enum aclk_database_opcode opcode;

653 RRDHOST *host;

654 struct aclk_sync_cfg_t *aclk_host_config;

655 + aclk_query_t *query;

656 worker_is_idle();

657 uv_run(loop, UV_RUN_DEFAULT);

658

659 - if (unlikely(shutdown_requested)) {

660 - nd_log_limit_static_thread_var(erl, 1, 0);

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

662 - if (now_realtime_sec() - shutdown_initiated > MAX_SHUTDOWN_TIMEOUT_SECONDS) {

663 - nd_log_daemon(NDLP_INFO, "ACLKSYNC: Shutdown timeout, forcing exit");

664 - break;

665 - }

666 - continue;

667 - }

668 -

659 /* wait for commands */

660 cmd_batch_size = 0;

661 do {

@@ -882,8 +872,7 @@ static void *aclk_synchronization_event_loop(void *arg)

872 }

873 break;

874 case ACLK_SYNC_SHUTDOWN:

885 - shutdown_requested = 1;

886 - shutdown_initiated = now_realtime_sec();

875 + config->shutdown_requested = true;

876 mark_pending_req_cancel_all();

877 break;

878 default:

@@ -898,7 +887,15 @@ static void *aclk_synchronization_event_loop(void *arg)

887

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

889 uv_walk(loop, libuv_close_callback, notify_timer_close_callback);

901 - uv_run(loop, UV_RUN_NOWAIT);

890 +

891 + size_t loop_count = (MAX_SHUTDOWN_TIMEOUT_SECONDS * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;

892 +

893 + while (ACLK_JOBS_ARE_RUNNING && loop_count > 0) {

894 + if (!uv_run(loop, UV_RUN_NOWAIT))

895 + break; // No pending callbacks

896 + sleep_usec(SHUTDOWN_SLEEP_INTERVAL_MS * USEC_PER_MS);

897 + loop_count--;

898 + }

899

900 (void) uv_loop_close(loop);

901

@@ -915,11 +912,9 @@ static void *aclk_synchronization_event_loop(void *arg)

912 }

913

914 release_cmd_pool(&config->cmd_pool);

918 - completion_mark_complete(&config->start_stop_complete);

919 -

915 worker_unregister();

916 service_exits();

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

917 + completion_mark_complete(&config->start_stop_complete);

918 return NULL;

919 }

920

src/database/sqlite/sqlite_metadata.c

+14 -8

@@ -2372,7 +2372,7 @@ static void store_hosts_metadata(BUFFER *work_buffer, bool is_worker)

2372 if (is_worker)

2373 worker_is_idle();

2374

2375 - metadata_scan_host(host, work_buffer, true);

2375 + metadata_scan_host(host, work_buffer, is_worker);

2376

2377 if (!is_worker)

2378 nd_log_daemon(NDLP_INFO, "METADATA: Progress of metadata storage: %6.2f%% completed", (100.0 * count / host_count));

@@ -2634,8 +2634,13 @@ static void *metadata_event_loop(void *arg)

2634 uv_walk(loop, libuv_close_callback, NULL);

2635

2636 size_t loop_count = (MAX_SHUTDOWN_TIMEOUT_SECONDS * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;

2637 +

2638 + // are we waiting for callbacks?

2639 + bool callbacks_pending = (config->metadata_running || config->ctx_load_running);

2640 +

2641 while ((config->metadata_running || config->ctx_load_running) && loop_count > 0) {

2638 - if (!uv_run(loop, UV_RUN_NOWAIT))

2642 + callbacks_pending = uv_run(loop, UV_RUN_NOWAIT);

2643 + if (!callbacks_pending)

2644 break; // No pending callbacks

2645 sleep_usec(SHUTDOWN_SLEEP_INTERVAL_MS * USEC_PER_MS);

2646 loop_count--;

@@ -2643,11 +2648,12 @@ static void *metadata_event_loop(void *arg)

2648

2649 (void)uv_loop_close(loop);

2650

2646 - store_hosts_metadata(work_buffer, false);

2647 -

2648 - store_alert_transitions(pending_alert_list, false);

2649 -

2650 - store_sql_statements(pending_sql_statement, false);

2651 + // If we are still waitinng for callbacks we timed out, don't run these

2652 + if (!callbacks_pending) {

2653 + store_hosts_metadata(work_buffer, false);

2654 + store_alert_transitions(pending_alert_list, false);

2655 + store_sql_statements(pending_sql_statement, false);

2656 + }

2657

2658 if (pending_ctx_cleanup_list) {

2659 Word_t Index = 0;

@@ -2666,7 +2672,7 @@ static void *metadata_event_loop(void *arg)

2672 buffer_free(work_buffer);

2673 release_cmd_pool(&config->cmd_pool);

2674 worker_unregister();

2669 -

2675 + service_exits();

2676 completion_mark_complete(&config->start_stop_complete);

2677

2678 return NULL;