Send node info when node is switched to ephemeral (#21456)
* Send node info when node is switched to ephemeral * Fix type for `completion` parameter in `aclk_host_state_update` and adjust syntax inconsistencies * Handle `completion` in early returns for `aclk_update_node_instance_job` and `aclk_host_state_update` * Refactor `completion` usage to replace with `aclk_sync_completion` for improved safety and timed waits * Add null check for UUID in `aclk_update_node_instance_job` and `aclk_host_state_update` to prevent unnecessary updates * Fix potential memory leak, re-check host after cloud notification * Restart iteration of host list after releasing lock to handle potential modifications
Stelios Fragkakis committed
Jan 19, 2026 at 18:47 UTC
b54ea527f4d5387d4cbbca726bf454027017abbc
10 files changed
+169
-23
src/aclk/aclk.c
+22
-9
@@ -945,7 +945,7 @@ bool aclk_host_state_update_auto(RRDHOST *host) {
945
live = 1;
946
break;
947
}
948
- aclk_host_state_update(host, live, 1);
948
+ aclk_host_state_update(host, live, 1, NULL);
949
return true;
950
}
951
@@ -976,16 +976,23 @@ void aclk_create_node_instance_job(RRDHOST *host)
976
aclk_add_job(query);
977
}
978
979
-void aclk_update_node_instance_job(RRDHOST *host, int live, int queryable)
979
+void aclk_update_node_instance_job(RRDHOST *host, int live, int queryable, struct aclk_sync_completion *sync_completion)
980
{
981
- if (unlikely(!host))
981
+ if (unlikely(!host)) {
982
+ if (sync_completion)
983
+ aclk_sync_completion_signal(sync_completion);
984
return;
985
+ }
986
987
CLAIM_ID claim_id = claim_id_get();
985
- if (!claim_id_is_set(claim_id))
988
+ if (!claim_id_is_set(claim_id)) {
989
+ if (sync_completion)
990
+ aclk_sync_completion_signal(sync_completion);
991
return;
992
+ }
993
994
aclk_query_t *query = aclk_query_new(NODE_STATE_UPDATE);
995
+ query->sync_completion = sync_completion;
996
997
int32_t hops = rrdhost_ingestion_hops(host);
998
node_instance_connection_t node_state_update = {
@@ -1017,15 +1024,21 @@ void aclk_update_node_instance_job(RRDHOST *host, int live, int queryable)
1024
aclk_add_job(query);
1025
}
1026
1020
-void aclk_host_state_update(RRDHOST *host, int live, int queryable)
1027
+void aclk_host_state_update(RRDHOST *host, int live, int queryable, struct aclk_sync_completion *sync_completion)
1028
{
1022
- if (!aclk_online())
1029
+ if (!aclk_online()) {
1030
+ if (sync_completion)
1031
+ aclk_sync_completion_signal(sync_completion);
1032
return;
1033
+ }
1034
1025
- if (uuid_is_null(host->node_id.uuid))
1035
+ if (uuid_is_null(host->node_id.uuid)) {
1036
aclk_create_node_instance_job(host);
1037
+ if (sync_completion)
1038
+ aclk_sync_completion_signal(sync_completion);
1039
+ }
1040
else
1028
- aclk_update_node_instance_job(host, live, queryable);
1041
+ aclk_update_node_instance_job(host, live, queryable, sync_completion);
1042
}
1043
1044
void aclk_send_node_instances()
@@ -1034,7 +1047,7 @@ void aclk_send_node_instances()
1047
dfe_start_reentrant(rrdhost_root_index, host)
1048
{
1049
int live = rrdhost_ingestion_status(host) == RRDHOST_INGEST_STATUS_ONLINE ? 1 : 0;
1037
- aclk_host_state_update(host, live, 1);
1050
+ aclk_host_state_update(host, live, 1, NULL);
1051
}
1052
dfe_done(host);
1053
}
src/aclk/aclk.h
+4
-1
@@ -5,6 +5,9 @@
5
#include "database/rrd.h"
6
7
#include "aclk_util.h"
8
+
9
+// Forward declaration - defined in aclk_query_queue.h
10
+struct aclk_sync_completion;
11
//#include "aclk_rrdhost_state.h"
12
13
#include "https_client.h"
@@ -91,7 +94,7 @@ extern struct aclk_shared_state {
94
int mqtt_shutdown_msg_rcvd;
95
} aclk_shared_state;
96
94
-void aclk_host_state_update(RRDHOST *host, int live, int queryable);
97
+void aclk_host_state_update(RRDHOST *host, int live, int queryable, struct aclk_sync_completion *sync_completion);
98
bool aclk_host_state_update_auto(RRDHOST *host);
99
100
void aclk_send_node_instances();
src/aclk/aclk_contexts_api.c
+2
-1
@@ -31,9 +31,10 @@ void aclk_update_node_collectors(struct update_node_collectors *collectors)
31
QUEUE_IF_PAYLOAD_PRESENT(query);
32
}
33
34
-void aclk_update_node_info(struct update_node_info *info)
34
+void aclk_update_node_info(struct update_node_info *info, struct aclk_sync_completion *sync_completion)
35
{
36
aclk_query_t *query = aclk_query_new(UPDATE_NODE_INFO);
37
+ query->sync_completion = sync_completion;
38
query->data.bin_payload.topic = ACLK_TOPICID_NODE_INFO;
39
query->data.bin_payload.payload = generate_update_node_info_message(&query->data.bin_payload.size, info);
40
query->data.bin_payload.msg_name = "UpdateNodeInfo";
src/aclk/aclk_contexts_api.h
+3
-1
@@ -4,11 +4,13 @@
4
5
#include "schema-wrappers/schema_wrappers.h"
6
7
+// Forward declaration - defined in aclk_query_queue.h
8
+struct aclk_sync_completion;
9
10
void aclk_send_contexts_snapshot(contexts_snapshot_t data);
11
void aclk_send_contexts_updated(contexts_updated_t data);
12
void aclk_update_node_collectors(struct update_node_collectors *collectors);
11
-void aclk_update_node_info(struct update_node_info *info);
13
+void aclk_update_node_info(struct update_node_info *info, struct aclk_sync_completion *sync_completion);
14
15
#endif /* ACLK_CONTEXTS_API_H */
16
src/aclk/aclk_query_queue.c
+3
@@ -98,5 +98,8 @@ void aclk_query_free(aclk_query_t *query)
98
freez(query->dedup_id);
99
freez(query->callback_topic);
100
freez(query->msg_id);
101
+
102
+ if (query->sync_completion)
103
+ aclk_sync_completion_signal(query->sync_completion);
104
return_query(query);
105
}
src/aclk/aclk_query_queue.h
+40
@@ -40,6 +40,45 @@ struct aclk_bin_payload {
40
const char *msg_name;
41
};
42
43
+// ----------------------------------------------------------------------------
44
+// Reference-counted completion for safe timed waits
45
+// Both waiter and query hold a reference; last one to release frees the structure
46
+
47
+struct aclk_sync_completion {
48
+ struct completion compl;
49
+ int32_t refcount;
50
+};
51
+
52
+static inline struct aclk_sync_completion *aclk_sync_completion_create(void) {
53
+ struct aclk_sync_completion *sc = callocz(1, sizeof(*sc));
54
+ completion_init(&sc->compl);
55
+ sc->refcount = 2; // One for waiter, one for query
56
+ return sc;
57
+}
58
+
59
+static inline void aclk_sync_completion_release(struct aclk_sync_completion *sc) {
60
+ if (__atomic_sub_fetch(&sc->refcount, 1, __ATOMIC_ACQ_REL) == 0) {
61
+ completion_destroy(&sc->compl);
62
+ freez(sc);
63
+ }
64
+}
65
+
66
+// Called by query processing to signal completion and release query's reference
67
+static inline void aclk_sync_completion_signal(struct aclk_sync_completion *sc) {
68
+ completion_mark_complete(&sc->compl);
69
+ aclk_sync_completion_release(sc);
70
+}
71
+
72
+// Called by waiter - waits with timeout, then releases waiter's reference
73
+// Returns true if completed within timeout, false if timed out
74
+static inline bool aclk_sync_completion_timedwait(struct aclk_sync_completion *sc, uint64_t timeout_s) {
75
+ bool result = completion_timedwait_for(&sc->compl, timeout_s);
76
+ aclk_sync_completion_release(sc);
77
+ return result;
78
+}
79
+
80
+// ----------------------------------------------------------------------------
81
+
82
typedef struct {
83
aclk_query_type_t type;
84
bool allocated;
@@ -68,6 +107,7 @@ typedef struct {
107
void *payload;
108
char *node_id;
109
} data;
110
+ struct aclk_sync_completion *sync_completion;
111
} aclk_query_t;
112
113
aclk_query_t *aclk_query_new(aclk_query_type_t type);
src/daemon/commands.c
+10
-4
@@ -371,8 +371,12 @@ static int remove_ephemeral_host(BUFFER *wb, RRDHOST *host, bool report_error, b
371
sql_set_host_label(&host->host_id.uuid, "_is_ephemeral", "true");
372
pulse_host_status(host, 0, 0);
373
374
- if(unregister) {
375
- aclk_host_state_update(host, 0, 0);
374
+ if (marked)
375
+ send_node_info_with_wait(host);
376
+
377
+ if (unregister) {
378
+ send_node_update_with_wait(host, 0, 0);
379
+
380
unregister_node(host->machine_guid);
381
host->node_id = UUID_ZERO;
382
buffer_sprintf(wb, "Node '%s' (machine guid: %s) has been unregistered",
@@ -382,12 +386,14 @@ static int remove_ephemeral_host(BUFFER *wb, RRDHOST *host, bool report_error, b
386
rrd_wrunlock();
387
return 1;
388
}
385
- else if(marked) {
389
+
390
+ if (marked) {
391
buffer_sprintf(wb, "Node '%s' (machine guid: %s) has been marked ephemeral",
392
rrdhost_hostname(host), host->machine_guid);
393
return 1;
394
}
390
- else if (report_error) {
395
+
396
+ if (report_error) {
397
buffer_sprintf(wb, "Node '%s' (machine guid: %s) is already ephemeral - not changing it",
398
rrdhost_hostname(host), host->machine_guid);
399
}
src/daemon/service.c
+29
-4
@@ -210,10 +210,35 @@ static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
210
211
if (delete) {
212
netdata_log_info("Host '%s' with machine guid '%s' is archived, ephemeral clean up.", rrdhost_hostname(host), host->machine_guid);
213
- // we inform cloud a child has been removed
214
- aclk_host_state_update(host, 0, 0);
215
- unregister_node(host->machine_guid);
216
- rrdhost_free___while_having_rrd_wrlock(host);
213
+
214
+ // Save machine_guid before releasing lock - we'll use it to look up fresh pointers
215
+ char machine_guid[UUID_STR_LEN];
216
+ strncpyz(machine_guid, host->machine_guid, GUID_LEN);
217
+
218
+ // Release lock before synchronous cloud operations to avoid deadlock
219
+ // (build_node_info acquires rrd_rdlock which would deadlock with our wrlock)
220
+ rrd_wrunlock();
221
+
222
+ // Look up fresh pointer for cloud operations (don't use stale 'host' pointer)
223
+ RRDHOST *cloud_host = rrdhost_find_by_guid(machine_guid);
224
+ if (cloud_host) {
225
+ send_node_info_with_wait(cloud_host);
226
+ send_node_update_with_wait(cloud_host, 0, 0);
227
+ }
228
+
229
+ // Re-acquire lock for cleanup
230
+ rrd_wrlock();
231
+
232
+ // Re-validate host still exists for cleanup
233
+ RRDHOST *host_check = rrdhost_find_by_guid(machine_guid);
234
+ if (host_check) {
235
+ unregister_node(host_check->machine_guid);
236
+ rrdhost_free___while_having_rrd_wrlock(host_check);
237
+ }
238
+
239
+ // Restart iteration - the list may have changed while lock was released
240
+ next = localhost;
241
+ now = now_realtime_sec();
242
}
243
else
244
rrdhost_cleanup_data_collection_and_health(host);
src/database/sqlite/sqlite_aclk_node.c
+54
-3
@@ -5,6 +5,7 @@
5
6
#include "../../aclk/aclk_contexts_api.h"
7
#include "../../aclk/aclk_capas.h"
8
+#include "../../aclk/aclk_query_queue.h"
9
10
DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) {
11
RRDSET *st;
@@ -44,7 +45,7 @@ static void build_node_collectors(RRDHOST *host)
45
aclk_host_config->node_id, rrdhost_hostname(host));
46
}
47
47
-static void build_node_info(RRDHOST *host)
48
+static void build_node_info(RRDHOST *host, struct aclk_sync_completion *sync_completion)
49
{
50
struct update_node_info node_info;
51
@@ -82,7 +83,7 @@ static void build_node_info(RRDHOST *host)
83
84
rrdhost_system_info_to_node_info(host->system_info, &node_info);
85
85
- aclk_update_node_info(&node_info);
86
+ aclk_update_node_info(&node_info, sync_completion);
87
nd_log(
88
NDLS_ACCESS,
89
NDLP_DEBUG,
@@ -99,6 +100,56 @@ static void build_node_info(RRDHOST *host)
100
aclk_host_config->node_collectors_send = now_realtime_sec();
101
}
102
103
+void send_node_info_with_wait(RRDHOST *host)
104
+{
105
+ if (unlikely(!host || !__atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED)))
106
+ return;
107
+
108
+ // No node_id means cloud doesn't know about this node - nothing to update
109
+ if (uuid_is_null(host->node_id.uuid))
110
+ return;
111
+
112
+ if (!aclk_online())
113
+ return;
114
+
115
+ struct aclk_sync_completion *sc = aclk_sync_completion_create();
116
+
117
+ build_node_info(host, sc);
118
+
119
+ bool success = aclk_sync_completion_timedwait(sc, 30);
120
+ if (!success) {
121
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
122
+ "Timed out waiting for node info update for host '%s'",
123
+ rrdhost_hostname(host));
124
+ }
125
+ // sc is automatically freed when both waiter and query release their references
126
+}
127
+
128
+void send_node_update_with_wait(RRDHOST *host, int live, int queryable)
129
+{
130
+ if (unlikely(!host || !__atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED)))
131
+ return;
132
+
133
+ // No node_id means cloud doesn't know about this node - nothing to update
134
+ if (uuid_is_null(host->node_id.uuid))
135
+ return;
136
+
137
+ if (!aclk_online())
138
+ return;
139
+
140
+ struct aclk_sync_completion *sc = aclk_sync_completion_create();
141
+
142
+ aclk_host_state_update(host, live, queryable, sc);
143
+
144
+ bool success = aclk_sync_completion_timedwait(sc, 30);
145
+ if (!success) {
146
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
147
+ "Timed out waiting for node state update for host '%s'",
148
+ rrdhost_hostname(host));
149
+ }
150
+ // sc is automatically freed when both waiter and query release their references
151
+}
152
+
153
void aclk_check_node_info_and_collectors(void)
154
{
155
RRDHOST *host;
@@ -166,7 +217,7 @@ void aclk_check_node_info_and_collectors(void)
217
if (pp_queue_empty && aclk_host_config->node_info_send_time &&
218
aclk_host_config->node_info_send_time + 30 < now) {
219
aclk_host_config->node_info_send_time = 0;
169
- build_node_info(host);
220
+ build_node_info(host, NULL);
221
schedule_node_state_update(host, 10000);
222
internal_error(true, "ACLK SYNC: Sending node info for %s", rrdhost_hostname(host));
223
}
src/database/sqlite/sqlite_aclk_node.h
+2
@@ -4,4 +4,6 @@
4
#define NETDATA_SQLITE_ACLK_NODE_H
5
6
void aclk_check_node_info_and_collectors(void);
7
+void send_node_info_with_wait(RRDHOST *host);
8
+void send_node_update_with_wait(RRDHOST *host, int live, int queryable);
9
#endif //NETDATA_SQLITE_ACLK_NODE_H