@cryptotaxi247 / netdata-1 / commits / 3e0227819

Unregister node from the agent to run in a worker thread (#19471)

Stelios Fragkakis committed Jan 23, 2025 at 23:02 UTC 3e022781999955c02c06897064f6e91dd2fe3027

3 files changed +37 -8

src/daemon/libuv_workers.c

@@ -82,6 +82,7 @@ void register_libuv_worker_jobs() {

82 worker_register_job_name(UV_EVENT_ALERT_START_STREAMING, "alert start streaming");

83 worker_register_job_name(UV_EVENT_ALERT_CHECKPOINT, "alert checkpoint");

84 worker_register_job_name(UV_EVENT_CREATE_NODE_INSTANCE, "create node instance");

85 + worker_register_job_name(UV_EVENT_UNREGISTER_NODE, "unregister node locally");

87 // netdatacli

88 worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");

src/daemon/libuv_workers.h

@@ -75,6 +75,7 @@ enum event_loop_job {

75 UV_EVENT_ALERT_START_STREAMING,

76 UV_EVENT_ALERT_CHECKPOINT,

77 UV_EVENT_CREATE_NODE_INSTANCE,

78 + UV_EVENT_UNREGISTER_NODE,

80 // netdatacli

81 UV_EVENT_SCHEDULE_CMD,

src/database/sqlite/sqlite_aclk.c

+35 -8

@@ -526,6 +526,30 @@ static void aclk_execute_batch(uv_work_t *req)

526 worker_is_idle();

527 }

528

529 +struct worker_data {

530 + uv_work_t request;

531 + void *payload;

532 + struct aclk_sync_config_s *config;

533 +};

534 +

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

536 +{

537 + struct worker_data *data = req->data;

538 + freez(data);

539 +}

540 +

541 +static void do_unregister_node(uv_work_t *req)

542 +{

543 + register_libuv_worker_jobs();

544 +

545 + struct worker_data *data = req->data;

546 +

547 + worker_is_busy(UV_EVENT_UNREGISTER_NODE);

548 +

549 + sql_unregister_node(data->payload);

550 +

551 + worker_is_idle();

552 +}

553

554 static void node_update_timer_cb(uv_timer_t *handle)

555 {

@@ -545,14 +569,9 @@ static void close_callback(uv_handle_t *handle, void *data __maybe_unused)

569 uv_close(handle, NULL); // Automatically close and free the handle

570 }

571

548 -struct alert_push_data {

549 - uv_work_t request;

550 - struct aclk_sync_config_s *config;

551 -};

552 -

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

573 {

555 - struct alert_push_data *data = req->data;

574 + struct worker_data *data = req->data;

575 struct aclk_sync_config_s *config = data->config;

576

577 config->alert_push_running = false;

@@ -633,7 +652,7 @@ static void aclk_synchronization(void *arg)

652 int query_thread_count = netdata_conf_cloud_query_threads();

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

654

636 - struct alert_push_data *data;

655 + struct worker_data *data;

656 aclk_query_t query;

657 struct judy_list_t *aclk_query_batch = NULL;

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

@@ -710,7 +729,15 @@ static void aclk_synchronization(void *arg)

729 break;

730

731 case ACLK_DATABASE_NODE_UNREGISTER:

713 - sql_unregister_node(cmd.param[0]);

732 + data = mallocz(sizeof(*data));

733 + data->request.data = data;

734 + data->config = config;

735 + data->payload = cmd.param[0];

736 +

737 + if (uv_queue_work(loop, &data->request, do_unregister_node, after_do_unregister_node)) {

738 + freez(data->payload);

739 + freez(data);

740 + }

741 break;

742 // ALERTS

743 case ACLK_DATABASE_PUSH_ALERT_CONFIG: