@cryptotaxi247 / netdata-1 / commits / 7eaf10d19

Improve error handling and resource cleanup in ACLK query execution (#21479)

* Improve error handling and resource cleanup in ACLK query execution - Refine logic to handle resource cleanup when queueing commands fails. - Fix potential resource leaks by ensuring proper memory deallocation in failure paths. - Simplify and align conditional checks for better readability and maintainability. - Minor adjustments to redundant logging and error reporting. * Fix query batch count decrement during ACLK query cleanup

Stelios Fragkakis committed Dec 18, 2025 at 19:56 UTC 7eaf10d19077abfa9e1fca276750c649e3baec4c

1 file changed +49 -26

src/database/sqlite/sqlite_aclk.c

+49 -26

@@ -167,18 +167,19 @@ static int create_host_callback(void *data, int argc, char **argv, char **column

167

168 rrdhost_system_info_free(system_info);

169

170 - if (likely(host)) {

171 - if (is_ephemeral)

172 - rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);

170 + if (unlikely(!host))

171 + return 0;

172

174 - if (is_ephemeral)

175 - host->stream.rcv.status.last_disconnected = now_realtime_sec();

173 + if (is_ephemeral)

174 + rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);

175

177 - host->rrdlabels = sql_load_host_labels((nd_uuid_t *)argv[IDX_HOST_ID]);

178 - host->stream.snd.status.last_connected = last_connected;

176 + if (is_ephemeral)

177 + host->stream.rcv.status.last_disconnected = now_realtime_sec();

178

180 - pulse_host_status(host, 0, 0); // this will detect the receiver status

181 - }

179 + host->rrdlabels = sql_load_host_labels((nd_uuid_t *)argv[IDX_HOST_ID]);

180 + host->stream.snd.status.last_connected = last_connected;

181 +

182 + pulse_host_status(host, 0, 0); // this will detect the receiver status

183

184 if (IS_VIRTUAL_HOST_OS(host))

185 node_data->vnodes++;

@@ -320,6 +321,7 @@ static void aclk_run_query(struct aclk_sync_config_s *config, aclk_query_t *quer

321 {

322 if (query->type == UNKNOWN || query->type >= ACLK_QUERY_TYPE_COUNT) {

323 error_report("Unknown query in query queue. %u", query->type);

324 + aclk_query_free(query);

325 return;

326 }

327

@@ -440,9 +442,6 @@ static void aclk_execute_batch(uv_work_t *req)

442 if (!aclk_query_batch)

443 return;

444

443 - usec_t started_ut = now_monotonic_usec(); (void)started_ut;

444 -

445 - size_t entries = aclk_query_batch->count;

445 Word_t Index = 0;

446 bool first = true;

447 Pvoid_t *Pvalue;

@@ -457,11 +456,6 @@ static void aclk_execute_batch(uv_work_t *req)

456 (void) JudyLFreeArray(&aclk_query_batch->JudyL, PJE0);

457 freez(aclk_query_batch);

458

460 - usec_t ended_ut = now_monotonic_usec();

461 - (void)ended_ut;

462 - nd_log_daemon(

463 - NDLP_DEBUG, "Processed %zu ACLK commands in %0.2f ms", entries, (double)(ended_ut - started_ut) / USEC_PER_MS);

464 -

459 worker_is_idle();

460 }

461

@@ -729,7 +723,7 @@ static void aclk_synchronization_event_loop(void *arg)

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

724 aclk_host_config = host->aclk_host_config;

725 }

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

726 + aclk_host_config->node_info_send_time = (host == localhost || immediate) ? 1 : now_realtime_sec();

727 break;

728 case ACLK_CANCEL_NODE_UPDATE_TIMER:

729 host = cmd.param[0];

@@ -786,8 +780,10 @@ static void aclk_synchronization_event_loop(void *arg)

780 if (Pvalue != PJERR) {

781 *Pvalue = query;

782 pending_queries++;

789 - } else

783 + } else {

784 nd_log_daemon(NDLP_ERR, "Failed to add ACLK command to the pending commands Judy");

785 + aclk_query_free(query);

786 + }

787 break;

788 }

789

@@ -831,8 +827,10 @@ static void aclk_synchronization_event_loop(void *arg)

827 *Pvalue = query;

828 pending_queries++;

829 }

834 - else

830 + else {

831 nd_log_daemon(NDLP_ERR, "Failed to add ACLK command to the pending commands Judy");

832 + aclk_query_free(query);

833 + }

834 }

835 break;

836

@@ -846,8 +844,13 @@ static void aclk_synchronization_event_loop(void *arg)

844 aclk_query_batch = callocz(1, sizeof(*aclk_query_batch));

845

846 Pvalue = JudyLIns(&aclk_query_batch->JudyL, ++aclk_query_batch->count, PJE0);

849 - if (Pvalue)

847 + if (Pvalue != PJERR)

848 *Pvalue = query;

849 + else {

850 + aclk_query_free(query);

851 + aclk_query_batch->count--;

852 + break;

853 + }

854

855 config->aclk_jobs_pending++;

856 if (aclk_query_batch->count < MAX_ACLK_BATCH_JOBS_IN_QUEUE || config->aclk_batch_job_is_running)

@@ -891,7 +894,7 @@ static void aclk_synchronization_event_loop(void *arg)

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

895

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

894 - uv_walk(loop, libuv_close_callback, notify_timer_close_callback);

897 + uv_walk(loop, libuv_close_callback, NULL);

898

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

900

@@ -1084,7 +1087,14 @@ void aclk_push_alert_config(const char *node_id, const char *config_hash)

1087 if (unlikely(!node_id || !config_hash))

1088 return;

1089

1087 - queue_aclk_sync_cmd(ACLK_DATABASE_PUSH_ALERT_CONFIG, strdupz(node_id), strdupz(config_hash));

1090 + char *node_id_dup = strdupz(node_id);

1091 + char *config_hash_dup = strdupz(config_hash);

1092 + bool queued = queue_aclk_sync_cmd(ACLK_DATABASE_PUSH_ALERT_CONFIG, node_id_dup, config_hash_dup);

1093 + if (unlikely(!queued)) {

1094 + nd_log_daemon(NDLP_WARNING, "ACLK: Failed to queue alert config push for node %s (config hash %s)", node_id, config_hash);

1095 + freez(node_id_dup);

1096 + freez(config_hash_dup);

1097 + }

1098 }

1099

1100 void aclk_execute_query(aclk_query_t *query)

@@ -1092,7 +1102,11 @@ void aclk_execute_query(aclk_query_t *query)

1102 if (unlikely(!query))

1103 return;

1104

1095 - (void) queue_aclk_sync_cmd(ACLK_QUERY_EXECUTE, query, NULL);

1105 + bool queued = queue_aclk_sync_cmd(ACLK_QUERY_EXECUTE, query, NULL);

1106 + if (unlikely(!queued)) {

1107 + nd_log_daemon(NDLP_WARNING, "ACLK: Failed to queue query execution");

1108 + aclk_query_free(query);

1109 + }

1110 }

1111

1112 void aclk_add_job(aclk_query_t *query)

@@ -1100,7 +1114,11 @@ void aclk_add_job(aclk_query_t *query)

1114 if (unlikely(!query))

1115 return;

1116

1103 - (void) queue_aclk_sync_cmd(ACLK_QUERY_BATCH_ADD, query, NULL);

1117 + bool queued = queue_aclk_sync_cmd(ACLK_QUERY_BATCH_ADD, query, NULL);

1118 + if (unlikely(!queued)) {

1119 + nd_log_daemon(NDLP_WARNING, "ACLK: Failed to queue query job");

1120 + aclk_query_free(query);

1121 + }

1122 }

1123

1124 void aclk_mqtt_client_set(mqtt_wss_client client)

@@ -1133,7 +1151,12 @@ void unregister_node(const char *machine_guid)

1151 if (unlikely(!machine_guid))

1152 return;

1153

1136 - (void) queue_aclk_sync_cmd(ACLK_DATABASE_NODE_UNREGISTER, strdupz(machine_guid), NULL);

1154 + char *machine_guid_dup = strdupz(machine_guid);

1155 + bool queued = queue_aclk_sync_cmd(ACLK_DATABASE_NODE_UNREGISTER, machine_guid_dup, NULL);

1156 + if (unlikely(!queued)) {

1157 + nd_log_daemon(NDLP_WARNING, "ACLK: Failed to queue unregister node command for %s", machine_guid);

1158 + freez(machine_guid_dup);

1159 + }

1160 }

1161

1162 void destroy_aclk_config(RRDHOST *host)