Handle virtual host disconnection (#20860)
* Handle virtual host disconnection * Clean correct host * Add some logging * Maintain a list of virtual host to process on cleanup * Keep track of vnode activity so that we can eventually switch it to stale Add additional checks in the service thread * Add simple check for node being stale If host resumes, set online * Configure stale after seconds on host define end * Add more logging * Properly reschedule check time * Store last connection time for vnode on creation * Reduce logging * Stop sender when vnode goes stale
Stelios Fragkakis committed
Sep 19, 2025 at 18:02 UTC
28a557cecca908f287352967be7d96aca2d21e6a
7 files changed
+108
-25
src/daemon/service.c
+1
-1
@@ -185,7 +185,7 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
185
186
svc_rrdhost_cleanup_charts_marked_obsolete(host);
187
188
- if (rrdhost_is_local(host))
188
+ if (rrdhost_is_local(host) || IS_VIRTUAL_HOST_OS(host))
189
continue;
190
191
rrdhost_receiver_lock(host);
src/database/rrdhost.h
+4
@@ -8,6 +8,8 @@
8
#define HOST_LABEL_IS_EPHEMERAL "_is_ephemeral"
9
#define NETDATA_VIRTUAL_HOST "Netdata Virtual Host 1.0"
10
11
+#define IS_VIRTUAL_HOST_OS(host) (strcmp(string2str(host->os), NETDATA_VIRTUAL_HOST) == 0)
12
+
13
struct stream_thread;
14
struct rrdset;
15
@@ -139,6 +141,8 @@ struct rrdhost {
141
STRING *program_name; // the program name that collects metrics for this host
142
STRING *program_version; // the program version that collects metrics for this host
143
144
+ uint32_t node_stale_after_seconds; // vnode stale timeout
145
+
146
OBJECT_STATE state_id; // every time data collection (stream receiver) (dis)connects,
147
// this gets incremented - it is used to detect stale functions,
148
// stale backfilling requests, etc.
src/database/sqlite/sqlite_aclk_node.c
+1
-2
@@ -65,8 +65,7 @@ static void build_node_info(RRDHOST *host)
65
now_realtime_timeval(&node_info.updated_at);
66
67
char *host_version = NULL;
68
- bool is_virtual_host = (rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) ||
69
- strcmp(string2str(host->os), NETDATA_VIRTUAL_HOST) == 0);
68
+ bool is_virtual_host = (rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) || IS_VIRTUAL_HOST_OS(host));
69
70
if (host != localhost && !is_virtual_host)
71
host_version = stream_receiver_program_version_strdupz(host);
src/plugins.d/pluginsd_parser.c
+75
-3
@@ -164,7 +164,10 @@ static inline PARSER_RC pluginsd_host_dictionary(char **words, size_t num_words,
164
return PLUGINSD_DISABLE_PLUGIN(parser, keyword, "host is not defined, send " PLUGINSD_KEYWORD_HOST_DEFINE " before this");
165
166
rrdlabels_add(labels, name, value, RRDLABEL_SRC_CONFIG);
167
-
167
+ if (strcmp(name, "_node_stale_after_seconds") == 0) {
168
+ uint32_t seconds = str2u(value);
169
+ parser->user.host_define.node_stale_after_seconds = seconds;
170
+ }
171
return PARSER_RC_OK;
172
}
173
@@ -190,6 +193,8 @@ static inline void pluginsd_update_host_ephemerality(RRDHOST *host) {
193
rrdlabels_add(host->rrdlabels, HOST_LABEL_IS_EPHEMERAL, value, RRDLABEL_SRC_CONFIG);
194
}
195
196
+#define VNODE_BASE_EPOCH (1704067200L) // Jan 1, 2024 00:00:00 UTC
197
+
198
static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
199
if(!parser->user.host_define.parsing_host)
200
return PLUGINSD_DISABLE_PLUGIN(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END, "missing initialization, send " PLUGINSD_KEYWORD_HOST_DEFINE " before this");
@@ -251,14 +256,55 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si
256
schedule_node_state_update(host, 100);
257
258
rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_LABELS | RRDHOST_FLAG_METADATA_UPDATE);
259
+ uint32_t *Pvalue = (uint32_t *) JudyLIns(&parser->user.vnodes.JudyL, (Word_t) host, PJE0);
260
+ if (Pvalue != PJERR)
261
+ *Pvalue = (uint32_t) (now_realtime_sec() - VNODE_BASE_EPOCH);
262
+ else
263
+ nd_log_daemon(NDLP_ERR, "VNODE: Cannot track virtual host \"%s\" for staleness - JudyLIns error", rrdhost_hostname(host));
264
+ host->node_stale_after_seconds = parser->user.host_define.node_stale_after_seconds;
265
+ nd_log_daemon(NDLP_INFO, "VNODE: Configuring node stale after %u seconds for host \"%s\"", host->node_stale_after_seconds, rrdhost_hostname(host));
266
return PARSER_RC_OK;
267
}
268
257
-static inline PARSER_RC pluginsd_host(char **words, size_t num_words, PARSER *parser) {
269
+static inline PARSER_RC pluginsd_host(char **words, size_t num_words, PARSER *parser)
270
+{
271
+ static time_t last_host_stale_check = 0;
272
char *guid = get_word(words, num_words, 1);
273
274
if(!guid || !*guid || strcmp(guid, "localhost") == 0) {
275
parser->user.host = localhost;
276
+ // Check if we need to switch any nodes to stale
277
+ uint32_t min_check_interval = UINT_MAX;
278
+ time_t now = now_realtime_sec();
279
+ if (last_host_stale_check < now) {
280
+ Word_t Index = 0;
281
+ bool first_then_next = true;
282
+ uint32_t *Pvalue;
283
+ while ((Pvalue = (uint32_t *) JudyLFirstThenNext(parser->user.vnodes.JudyL, &Index, &first_then_next))) {
284
+ RRDHOST *virtual_host = (RRDHOST *) Index;
285
+ uint32_t stale_after_seconds = virtual_host->node_stale_after_seconds;
286
+ if (!stale_after_seconds)
287
+ continue;
288
+
289
+ min_check_interval = MIN(min_check_interval, stale_after_seconds);
290
+ if (rrdhost_option_check(virtual_host, RRDHOST_OPTION_VIRTUAL_HOST)) {
291
+ time_t last_seen = (*Pvalue + VNODE_BASE_EPOCH);
292
+ uint32_t seen_seconds_ago = (uint32_t) (now - last_seen);
293
+
294
+ if (seen_seconds_ago >= stale_after_seconds) {
295
+ rrdhost_option_clear(virtual_host, RRDHOST_OPTION_VIRTUAL_HOST);
296
+ rrdhost_flag_clear(virtual_host, RRDHOST_FLAG_COLLECTOR_ONLINE);
297
+ nd_log_daemon(NDLP_INFO, "VNODE: Marking node \"%s\" as STALE, last seen %u seconds ago", rrdhost_hostname(virtual_host), seen_seconds_ago);
298
+ schedule_node_state_update(virtual_host, 1000);
299
+ stream_sender_signal_to_stop_and_wait(virtual_host, STREAM_HANDSHAKE_SND_VNODE_IS_STALE, false);
300
+ }
301
+ }
302
+ }
303
+ if (min_check_interval == UINT_MAX)
304
+ min_check_interval = 60;
305
+
306
+ last_host_stale_check = now_realtime_sec() + min_check_interval;
307
+ }
308
return PARSER_RC_OK;
309
}
310
@@ -272,7 +318,17 @@ static inline PARSER_RC pluginsd_host(char **words, size_t num_words, PARSER *pa
318
return PLUGINSD_DISABLE_PLUGIN(parser, PLUGINSD_KEYWORD_HOST, "cannot find a host with this machine guid - have you created it?");
319
320
parser->user.host = host;
275
-
321
+ uint32_t *Pvalue = (uint32_t *) JudyLGet(parser->user.vnodes.JudyL, (Word_t) host, PJE0);
322
+ if (Pvalue) {
323
+ *Pvalue = (uint32_t) (now_realtime_sec() - VNODE_BASE_EPOCH);
324
+ // Check if we need to enable
325
+ if (!rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST)) {
326
+ rrdhost_option_set(host, RRDHOST_OPTION_VIRTUAL_HOST);
327
+ rrdhost_flag_set(host, RRDHOST_FLAG_COLLECTOR_ONLINE);
328
+ nd_log_daemon(NDLP_INFO, "VNODE: Re-enabling virtual host \"%s\"", rrdhost_hostname(host));
329
+ schedule_node_state_update(host, 1000);
330
+ }
331
+ }
332
return PARSER_RC_OK;
333
}
334
@@ -1260,6 +1316,22 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, int fd_input,
1316
else
1317
cd->serial_failures++;
1318
1319
+ {
1320
+ Word_t Index = 0;
1321
+ bool first_then_next = true;
1322
+ while (JudyLFirstThenNext(parser->user.vnodes.JudyL, &Index, &first_then_next)) {
1323
+ RRDHOST *virtual_host = (RRDHOST *) Index;
1324
+ nd_log_daemon(NDLP_INFO, "PLUGINSD: Checking virtual status for %s", rrdhost_hostname(virtual_host));
1325
+ if (rrdhost_option_check(virtual_host, RRDHOST_OPTION_VIRTUAL_HOST)) {
1326
+ nd_log_daemon(NDLP_INFO, "PLUGINSD: Reseting virtual host status for %s", rrdhost_hostname(virtual_host));
1327
+ rrdhost_option_clear(virtual_host, RRDHOST_OPTION_VIRTUAL_HOST);
1328
+ rrdhost_flag_clear(virtual_host, RRDHOST_FLAG_COLLECTOR_ONLINE);
1329
+ schedule_node_state_update(virtual_host, 1000);
1330
+ }
1331
+ }
1332
+ (void) JudyLFreeArray(&parser->user.vnodes.JudyL, PJE0);
1333
+ }
1334
+
1335
// mark all charts of this plugin as obsolete
1336
RRDSET *st;
1337
rrdset_foreach_read(st, localhost) {
src/plugins.d/pluginsd_parser.h
+6
@@ -69,6 +69,7 @@ typedef struct parser_user_object {
69
70
struct {
71
bool parsing_host;
72
+ uint32_t node_stale_after_seconds;
73
nd_uuid_t machine_guid;
74
char machine_guid_str[UUID_STR_LEN];
75
STRING *hostname;
@@ -95,6 +96,11 @@ typedef struct parser_user_object {
96
time_t wall_clock_time;
97
bool ml_locked;
98
} v2;
99
+
100
+ struct {
101
+ Pvoid_t JudyL;
102
+ } vnodes;
103
+
104
} PARSER_USER_OBJECT;
105
106
typedef void (*parser_deferred_action_t)(struct parser *parser, void *action_data);
src/streaming/stream-handshake.c
+1
@@ -34,6 +34,7 @@ static struct {
34
{STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED, "DISCONNECTED SND COMPRESSION FAILED", 415}, // Unsupported Media Type
35
{STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED, "HTTP UPGRADE ERROR", 426}, // Upgrade Required
36
{STREAM_HANDSHAKE_SND_DISCONNECT_RECEIVER_LEFT, "RECEIVER LEFT", 498},
37
+ {STREAM_HANDSHAKE_SND_VNODE_IS_STALE, "VNODE STALE", 498},
38
39
// receiver and sender codes
40
{STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP, "DISCONNECTED SIGNALED TO STOP", 499}, // Client Closed Request
src/streaming/stream-handshake.h
+20
-19
@@ -49,31 +49,32 @@ typedef enum {
49
STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED = -17,
50
STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED = -18,
51
STREAM_HANDSHAKE_SND_DISCONNECT_RECEIVER_LEFT = -19, // used only in pulse
52
+ STREAM_HANDSHAKE_SND_VNODE_IS_STALE = -20, // used only in pulse
53
54
// receiver and sender codes
54
- STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP = -20, // a fallback when (s|rpt)->exit.reason is not set
55
- STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN = -21,
56
- STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED = -22,
57
- STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED = -23,
58
- STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR = -24,
59
- STREAM_HANDSHAKE_DISCONNECT_TIMEOUT = -25,
60
- STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE = -26,
61
- STREAM_HANDSHAKE_DISCONNECT_BUFFER_OVERFLOW = -27,
62
- STREAM_HANDSHAKE_DISCONNECT_REPLICATION_STALLED = -28,
55
+ STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP = -21, // a fallback when (s|rpt)->exit.reason is not set
56
+ STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN = -22,
57
+ STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED = -23,
58
+ STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED = -24,
59
+ STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR = -25,
60
+ STREAM_HANDSHAKE_DISCONNECT_TIMEOUT = -26,
61
+ STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE = -27,
62
+ STREAM_HANDSHAKE_DISCONNECT_BUFFER_OVERFLOW = -28,
63
+ STREAM_HANDSHAKE_DISCONNECT_REPLICATION_STALLED = -29,
64
65
// sender (stream parents - SP) failures to connect
65
- STREAM_HANDSHAKE_SP_PREPARING = -29,
66
- STREAM_HANDSHAKE_SP_NO_HOST_IN_DESTINATION = -30,
67
- STREAM_HANDSHAKE_SP_CONNECT_TIMEOUT = -31,
68
- STREAM_HANDSHAKE_SP_CONNECTION_REFUSED = -32,
69
- STREAM_HANDSHAKE_SP_CANT_RESOLVE_HOSTNAME = -33,
70
- STREAM_HANDSHAKE_SP_CONNECTING = -34,
71
- STREAM_HANDSHAKE_SP_CONNECTED = -35,
72
- STREAM_HANDSHAKE_SP_NO_STREAM_INFO = -36,
73
- STREAM_HANDSHAKE_SP_NO_DESTINATION = -37,
66
+ STREAM_HANDSHAKE_SP_PREPARING = -30,
67
+ STREAM_HANDSHAKE_SP_NO_HOST_IN_DESTINATION = -31,
68
+ STREAM_HANDSHAKE_SP_CONNECT_TIMEOUT = -32,
69
+ STREAM_HANDSHAKE_SP_CONNECTION_REFUSED = -33,
70
+ STREAM_HANDSHAKE_SP_CANT_RESOLVE_HOSTNAME = -34,
71
+ STREAM_HANDSHAKE_SP_CONNECTING = -35,
72
+ STREAM_HANDSHAKE_SP_CONNECTED = -36,
73
+ STREAM_HANDSHAKE_SP_NO_STREAM_INFO = -37,
74
+ STREAM_HANDSHAKE_SP_NO_DESTINATION = -38,
75
76
// terminator - keep this positive, bigger than all negative values
76
- STREAM_HANDSHAKE_NEGATIVE_MAX = 38,
77
+ STREAM_HANDSHAKE_NEGATIVE_MAX = 39,
78
} STREAM_HANDSHAKE;
79
80
const char *stream_handshake_error_to_string(STREAM_HANDSHAKE reason);