Fix context hub cleanup (#21832)
* Add hub-queue pruning logic and improve resource management - Introduce `rrdcontext_prune_hub_queue` for better queue management, including handling stale or over-limit entries. - Add queue bounding and cleanup during unclaimed or non-dispatching states. - Improve RRDCONTEXT hub-queue handling to prevent indefinite retention of entries. - Enhance pluginsd resource cleanup by releasing and reacquiring `rrddim` resources appropriately. * Add queue revalidation under lock to ensure integrity during stale entry removal * - Prevent queue processing when the service is not running. - Log an error and return on failure to acquire dimensions during pluginsd slot cache updates.
Stelios Fragkakis committed
Mar 1, 2026 at 17:10 UTC
a70ea4c306f3a11fbbfa6168fc4a7a01179b374e
2 files changed
+106
-2
src/database/contexts/rrdcontext-queues.c
+95
-2
@@ -2,6 +2,9 @@
2
3
#include "rrdcontext-internal.h"
4
5
+#define RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_AGE_UT (15 * 60 * USEC_PER_SEC)
6
+#define RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_ENTRIES 10000
7
+
8
typedef enum {
9
RRDCONTEXT_QUEUE_INVALID = 0,
10
RRDCONTEXT_QUEUE_ADDED,
@@ -36,6 +39,10 @@ static inline RRDCONTEXT_QUEUE_STATUS rrdcontext_queue_add(RRDCONTEXT_QUEUE_Judy
39
void rrdcontext_add_to_hub_queue(RRDCONTEXT *rc) {
40
if(!rc || !rc->rrdhost) return;
41
42
+ CLAIM_ID claim_id = claim_id_get();
43
+ if(unlikely(!claim_id_is_set(claim_id)))
44
+ return;
45
+
46
spinlock_lock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
47
48
RRDCONTEXT_QUEUE_STATUS ret = rrdcontext_queue_add(&rc->rrdhost->rrdctx.hub_queue, rc, &rc->queue.idx, true);
@@ -54,6 +61,82 @@ void rrdcontext_add_to_hub_queue(RRDCONTEXT *rc) {
61
spinlock_unlock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
62
}
63
64
+static void rrdcontext_prune_hub_queue(RRDHOST *host, usec_t now_ut, bool force_drop_all, bool apply_bounds) {
65
+ if(!rrdcontext_queue_entries(&host->rrdctx.hub_queue))
66
+ return;
67
+
68
+ size_t dropped = 0;
69
+ int32_t queued = rrdcontext_queue_entries(&host->rrdctx.hub_queue);
70
+
71
+ spinlock_lock(&host->rrdctx.hub_queue.spinlock);
72
+ Word_t idx = 0;
73
+ for(RRDCONTEXT *rc = RRDCONTEXT_QUEUE_FIRST(&host->rrdctx.hub_queue, &idx);
74
+ rc;
75
+ rc = RRDCONTEXT_QUEUE_NEXT(&host->rrdctx.hub_queue, &idx)) {
76
+ if(unlikely(!service_running(SERVICE_CONTEXT)))
77
+ break;
78
+
79
+ const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(rc->id));
80
+ bool context_matches = item && (dictionary_acquired_item_value(item) == rc);
81
+ bool stale = !item || !context_matches;
82
+ bool do_it = context_matches;
83
+
84
+ spinlock_unlock(&host->rrdctx.hub_queue.spinlock);
85
+
86
+ if(context_matches) {
87
+ bool drop = force_drop_all;
88
+ if(!drop && apply_bounds) {
89
+ bool queue_is_over_limit = queued > RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_ENTRIES;
90
+ bool queue_item_expired = (now_ut > rc->queue.queued_ut) &&
91
+ ((now_ut - rc->queue.queued_ut) > RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_AGE_UT);
92
+ drop = queue_is_over_limit || queue_item_expired;
93
+ }
94
+
95
+ do_it = drop;
96
+ }
97
+
98
+ if(item)
99
+ dictionary_acquired_item_release(host->rrdctx.contexts, item);
100
+
101
+ spinlock_lock(&host->rrdctx.hub_queue.spinlock);
102
+ if(unlikely(!service_running(SERVICE_CONTEXT)))
103
+ break;
104
+
105
+ if(stale) {
106
+ // Revalidate after re-lock: queue may have changed while lock was dropped.
107
+ RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.hub_queue, idx);
108
+ if(rc_at_idx == rc) {
109
+ RRDCONTEXT_QUEUE_DEL(&host->rrdctx.hub_queue, idx);
110
+ __atomic_add_fetch(&host->rrdctx.hub_queue.version, 1, __ATOMIC_RELAXED);
111
+ __atomic_sub_fetch(&host->rrdctx.hub_queue.entries, 1, __ATOMIC_RELAXED);
112
+
113
+ rc->queue.idx = 0;
114
+ rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB);
115
+
116
+ dropped++;
117
+ if(queued > 0)
118
+ queued--;
119
+ }
120
+ }
121
+ else if(do_it) {
122
+ rrdcontext_del_from_hub_queue(rc, true);
123
+ dropped++;
124
+ if(queued > 0)
125
+ queued--;
126
+ }
127
+ }
128
+ spinlock_unlock(&host->rrdctx.hub_queue.spinlock);
129
+
130
+ if(unlikely(dropped)) {
131
+ nd_log_limit_static_global_var(erl, 1, 0);
132
+ nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE,
133
+ "RRDCONTEXT: host '%s' pruned %zu queued context updates (force_drop_all=%s, bounded=%s)",
134
+ rrdhost_hostname(host), dropped,
135
+ force_drop_all ? "true" : "false",
136
+ apply_bounds ? "true" : "false");
137
+ }
138
+}
139
+
140
void rrdcontext_add_to_pp_queue(RRDCONTEXT *rc) {
141
if(!rc || !rc->rrdhost) return;
142
@@ -165,9 +248,21 @@ void rrdcontext_post_process_queued_contexts(RRDHOST *host) {
248
}
249
250
void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
251
+ CLAIM_ID claim_id = claim_id_get();
252
+
253
+ if(unlikely(!claim_id_is_set(claim_id))) {
254
+ // unclaimed agents should not retain hub-queue state, otherwise queued flags block local GC indefinitely
255
+ rrdcontext_prune_hub_queue(host, now_ut, true, false);
256
+ return;
257
+ }
258
+
259
// check if we have received a streaming command for this host
260
if(UUIDiszero(host->node_id) || !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_online_for_contexts())
261
+ {
262
+ // claimed but currently not dispatching: keep queue bounded to avoid indefinite retention
263
+ rrdcontext_prune_hub_queue(host, now_ut, false, true);
264
return;
265
+ }
266
267
// check if there are queued items to send
268
if(!rrdcontext_queue_entries(&host->rrdctx.hub_queue))
@@ -195,8 +290,6 @@ void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
290
if (do_it) {
291
worker_is_busy(WORKER_JOB_QUEUED);
292
usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
198
- CLAIM_ID claim_id = claim_id_get();
199
-
293
if(unlikely(now_ut >= dispatch_ut) && claim_id_is_set(claim_id)) {
294
worker_is_busy(WORKER_JOB_CHECK);
295
src/plugins.d/pluginsd_internals.h
+11
@@ -151,7 +151,18 @@ static inline void pluginsd_rrddim_put_to_slot(PARSER *parser, RRDSET *st, RRDDI
151
struct pluginsd_rrddim *prd = &st->pluginsd.prd_array[slot - 1];
152
153
if(prd->rd != rd) {
154
+ rrddim_acquired_release(prd->rda);
155
+ prd->rda = NULL;
156
+ prd->rd = NULL;
157
+ prd->id = NULL;
158
+
159
prd->rda = rrddim_find_and_acquire(st, string2str(rd->id), true);
160
+ if(unlikely(!prd->rda)) {
161
+ netdata_log_error("PLUGINSD: failed to acquire dimension '%s' for chart '%s' while updating slot cache",
162
+ rrddim_id(rd), rrdset_id(st));
163
+ return;
164
+ }
165
+
166
prd->rd = rrddim_acquired_to_rrddim(prd->rda);
167
prd->id = string2str(prd->rd->id);
168
}