| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 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, |
| 11 | RRDCONTEXT_QUEUE_FOUND, |
| 12 | } RRDCONTEXT_QUEUE_STATUS; |
| 13 | |
| 14 | static inline RRDCONTEXT_QUEUE_STATUS rrdcontext_queue_add(RRDCONTEXT_QUEUE_JudyLSet *queue, RRDCONTEXT *rc, Word_t *idx, bool having_lock) { |
| 15 | RRDCONTEXT_QUEUE_STATUS ret = RRDCONTEXT_QUEUE_INVALID; |
| 16 | if(!queue || !rc || !idx) return ret; |
| 17 | |
| 18 | if(!having_lock) |
| 19 | spinlock_lock(&queue->spinlock); |
| 20 | |
| 21 | if(*idx) { |
| 22 | fatal_assert(RRDCONTEXT_QUEUE_GET(queue, *idx) == rc); |
| 23 | ret = RRDCONTEXT_QUEUE_FOUND; |
| 24 | } |
| 25 | else { |
| 26 | *idx = queue->id++; |
| 27 | RRDCONTEXT_QUEUE_SET(queue, *idx, rc); |
| 28 | __atomic_add_fetch(&queue->version, 1, __ATOMIC_RELAXED); |
| 29 | __atomic_add_fetch(&queue->entries, 1, __ATOMIC_RELAXED); |
| 30 | ret = RRDCONTEXT_QUEUE_ADDED; |
| 31 | } |
| 32 | |
| 33 | if(!having_lock) |
| 34 | spinlock_unlock(&queue->spinlock); |
| 35 | |
| 36 | return ret; |
| 37 | } |
| 38 | |
| 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); |
| 49 | |
| 50 | if(ret == RRDCONTEXT_QUEUE_ADDED) { |
| 51 | rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB); |
| 52 | rc->queue.queued_ut = now_realtime_usec(); |
| 53 | rc->queue.queued_flags = rrd_flags_get(rc); |
| 54 | } |
| 55 | else if(ret == RRDCONTEXT_QUEUE_FOUND) { |
| 56 | rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB); |
| 57 | rc->queue.queued_ut = now_realtime_usec(); |
| 58 | rc->queue.queued_flags |= rrd_flags_get(rc); |
| 59 | } |
| 60 | |
| 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 | STRING *lookup_id = string_dup(rc->id); |
| 80 | usec_t queued_ut = rc->queue.queued_ut; |
| 81 | spinlock_unlock(&host->rrdctx.hub_queue.spinlock); |
| 82 | |
| 83 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(lookup_id)); |
| 84 | bool context_matches = item && (dictionary_acquired_item_value(item) == rc); |
| 85 | bool stale = !item || !context_matches; |
| 86 | bool do_it = context_matches; |
| 87 | |
| 88 | if(context_matches) { |
| 89 | bool drop = force_drop_all; |
| 90 | if(!drop && apply_bounds) { |
| 91 | bool queue_is_over_limit = queued > RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_ENTRIES; |
| 92 | bool queue_item_expired = (now_ut > queued_ut) && |
| 93 | ((now_ut - queued_ut) > RRDCONTEXT_HUB_QUEUE_MAX_OFFLINE_AGE_UT); |
| 94 | drop = queue_is_over_limit || queue_item_expired; |
| 95 | } |
| 96 | |
| 97 | do_it = drop; |
| 98 | } |
| 99 | |
| 100 | if(item) |
| 101 | dictionary_acquired_item_release(host->rrdctx.contexts, item); |
| 102 | string_freez(lookup_id); |
| 103 | |
| 104 | spinlock_lock(&host->rrdctx.hub_queue.spinlock); |
| 105 | if(unlikely(!service_running(SERVICE_CONTEXT))) |
| 106 | break; |
| 107 | |
| 108 | if(stale) { |
| 109 | // Revalidate after re-lock: queue may have changed while lock was dropped. |
| 110 | RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.hub_queue, idx); |
| 111 | if(rc_at_idx == rc) { |
| 112 | RRDCONTEXT_QUEUE_DEL(&host->rrdctx.hub_queue, idx); |
| 113 | __atomic_add_fetch(&host->rrdctx.hub_queue.version, 1, __ATOMIC_RELAXED); |
| 114 | __atomic_sub_fetch(&host->rrdctx.hub_queue.entries, 1, __ATOMIC_RELAXED); |
| 115 | |
| 116 | rc->queue.idx = 0; |
| 117 | rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB); |
| 118 | |
| 119 | dropped++; |
| 120 | if(queued > 0) |
| 121 | queued--; |
| 122 | } |
| 123 | } |
| 124 | else if(do_it) { |
| 125 | RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.hub_queue, idx); |
| 126 | if(rc_at_idx == rc) { |
| 127 | rrdcontext_del_from_hub_queue(rc, true); |
| 128 | dropped++; |
| 129 | if(queued > 0) |
| 130 | queued--; |
| 131 | } |
| 132 | } |
| 133 | } |
| 134 | spinlock_unlock(&host->rrdctx.hub_queue.spinlock); |
| 135 | |
| 136 | if(unlikely(dropped)) { |
| 137 | nd_log_limit_static_global_var(erl, 1, 0); |
| 138 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE, |
| 139 | "RRDCONTEXT: host '%s' pruned %zu queued context updates (force_drop_all=%s, bounded=%s)", |
| 140 | rrdhost_hostname(host), dropped, |
| 141 | force_drop_all ? "true" : "false", |
| 142 | apply_bounds ? "true" : "false"); |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | void rrdcontext_add_to_pp_queue(RRDCONTEXT *rc) { |
| 147 | if(!rc || !rc->rrdhost) return; |
| 148 | |
| 149 | spinlock_lock(&rc->rrdhost->rrdctx.pp_queue.spinlock); |
| 150 | |
| 151 | RRDCONTEXT_QUEUE_STATUS ret = rrdcontext_queue_add(&rc->rrdhost->rrdctx.pp_queue, rc, &rc->pp.idx, true); |
| 152 | |
| 153 | if(ret == RRDCONTEXT_QUEUE_ADDED) { |
| 154 | rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP); |
| 155 | rc->pp.queued_flags = rc->flags; |
| 156 | rc->pp.queued_ut = now_realtime_usec(); |
| 157 | } |
| 158 | else if(ret == RRDCONTEXT_QUEUE_FOUND) { |
| 159 | rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP); |
| 160 | rc->pp.queued_flags |= rc->flags; |
| 161 | } |
| 162 | |
| 163 | spinlock_unlock(&rc->rrdhost->rrdctx.pp_queue.spinlock); |
| 164 | } |
| 165 | |
| 166 | static inline bool rrdcontext_queue_del(RRDCONTEXT_QUEUE_JudyLSet *queue, RRDCONTEXT *rc, Word_t *idx, bool having_lock) { |
| 167 | bool ret = false; |
| 168 | if(!queue || !rc || !idx) return ret; |
| 169 | |
| 170 | if(!having_lock) |
| 171 | spinlock_lock(&queue->spinlock); |
| 172 | |
| 173 | RRDCONTEXT *rc_found = RRDCONTEXT_QUEUE_GET(queue, *idx); |
| 174 | |
| 175 | if(rc_found == rc) { |
| 176 | RRDCONTEXT_QUEUE_DEL(queue, *idx); |
| 177 | __atomic_add_fetch(&queue->version, 1, __ATOMIC_RELAXED); |
| 178 | __atomic_sub_fetch(&queue->entries, 1, __ATOMIC_RELAXED); |
| 179 | ret = true; |
| 180 | } |
| 181 | *idx = 0; |
| 182 | |
| 183 | if(!having_lock) |
| 184 | spinlock_unlock(&queue->spinlock); |
| 185 | |
| 186 | return ret; |
| 187 | } |
| 188 | |
| 189 | void rrdcontext_del_from_hub_queue(RRDCONTEXT *rc, bool having_lock) { |
| 190 | if(!rc || !rc->rrdhost) return; |
| 191 | if(!having_lock) |
| 192 | spinlock_lock(&rc->rrdhost->rrdctx.hub_queue.spinlock); |
| 193 | |
| 194 | if(rrdcontext_queue_del(&rc->rrdhost->rrdctx.hub_queue, rc, &rc->queue.idx, true)) { |
| 195 | rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB); |
| 196 | } |
| 197 | |
| 198 | if(!having_lock) |
| 199 | spinlock_unlock(&rc->rrdhost->rrdctx.hub_queue.spinlock); |
| 200 | } |
| 201 | |
| 202 | void rrdcontext_del_from_pp_queue(RRDCONTEXT *rc, bool having_lock) { |
| 203 | if(!rc || !rc->rrdhost) return; |
| 204 | |
| 205 | if(!having_lock) |
| 206 | spinlock_lock(&rc->rrdhost->rrdctx.pp_queue.spinlock); |
| 207 | |
| 208 | if(rrdcontext_queue_del(&rc->rrdhost->rrdctx.pp_queue, rc, &rc->pp.idx, true)) { |
| 209 | rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_PP); |
| 210 | rc->pp.dequeued_ut = now_realtime_usec(); |
| 211 | } |
| 212 | |
| 213 | if(!having_lock) |
| 214 | spinlock_unlock(&rc->rrdhost->rrdctx.pp_queue.spinlock); |
| 215 | } |
| 216 | |
| 217 | |
| 218 | uint32_t rrdcontext_queue_version(RRDCONTEXT_QUEUE_JudyLSet *queue) { |
| 219 | return __atomic_load_n(&queue->version, __ATOMIC_RELAXED); |
| 220 | } |
| 221 | |
| 222 | int32_t rrdcontext_queue_entries(RRDCONTEXT_QUEUE_JudyLSet *queue) { |
| 223 | return __atomic_load_n(&queue->entries, __ATOMIC_RELAXED); |
| 224 | } |
| 225 | |
| 226 | void rrdcontext_post_process_queued_contexts(RRDHOST *host) { |
| 227 | |
| 228 | spinlock_lock(&host->rrdctx.pp_queue.spinlock); |
| 229 | Word_t idx = 0; |
| 230 | for(RRDCONTEXT *rc = RRDCONTEXT_QUEUE_FIRST(&host->rrdctx.pp_queue, &idx); |
| 231 | rc; |
| 232 | rc = RRDCONTEXT_QUEUE_NEXT(&host->rrdctx.pp_queue, &idx)) { |
| 233 | if(unlikely(!service_running(SERVICE_CONTEXT))) break; |
| 234 | |
| 235 | STRING *lookup_id = string_dup(rc->id); |
| 236 | spinlock_unlock(&host->rrdctx.pp_queue.spinlock); |
| 237 | |
| 238 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(lookup_id)); |
| 239 | bool do_it = item && (dictionary_acquired_item_value(item) == rc); |
| 240 | bool process_it = false; |
| 241 | |
| 242 | spinlock_lock(&host->rrdctx.pp_queue.spinlock); |
| 243 | if(unlikely(!service_running(SERVICE_CONTEXT))) { |
| 244 | spinlock_unlock(&host->rrdctx.pp_queue.spinlock); |
| 245 | if(item) |
| 246 | dictionary_acquired_item_release(host->rrdctx.contexts, item); |
| 247 | string_freez(lookup_id); |
| 248 | return; |
| 249 | } |
| 250 | |
| 251 | if(do_it) { |
| 252 | RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.pp_queue, idx); |
| 253 | if(rc_at_idx == rc) { |
| 254 | rrdcontext_del_from_pp_queue(rc, true); |
| 255 | process_it = true; |
| 256 | } |
| 257 | else |
| 258 | do_it = false; |
| 259 | } |
| 260 | |
| 261 | spinlock_unlock(&host->rrdctx.pp_queue.spinlock); |
| 262 | |
| 263 | if(item) { |
| 264 | if (process_it) |
| 265 | rrdcontext_post_process_updates(rc, false, RRD_FLAG_NONE, true); |
| 266 | dictionary_acquired_item_release(host->rrdctx.contexts, item); |
| 267 | } |
| 268 | string_freez(lookup_id); |
| 269 | |
| 270 | spinlock_lock(&host->rrdctx.pp_queue.spinlock); |
| 271 | } |
| 272 | |
| 273 | spinlock_unlock(&host->rrdctx.pp_queue.spinlock); |
| 274 | } |
| 275 | |
| 276 | void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) { |
| 277 | CLAIM_ID claim_id = claim_id_get(); |
| 278 | |
| 279 | if(unlikely(!claim_id_is_set(claim_id))) { |
| 280 | // unclaimed agents should not retain hub-queue state, otherwise queued flags block local GC indefinitely |
| 281 | rrdcontext_prune_hub_queue(host, now_ut, true, false); |
| 282 | return; |
| 283 | } |
| 284 | |
| 285 | // check if we have received a streaming command for this host |
| 286 | if(UUIDiszero(host->node_id) || !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_online_for_contexts()) |
| 287 | { |
| 288 | // claimed but currently not dispatching: keep queue bounded to avoid indefinite retention |
| 289 | rrdcontext_prune_hub_queue(host, now_ut, false, true); |
| 290 | return; |
| 291 | } |
| 292 | |
| 293 | // check if there are queued items to send |
| 294 | if(!rrdcontext_queue_entries(&host->rrdctx.hub_queue)) |
| 295 | return; |
| 296 | |
| 297 | size_t messages_added = 0; |
| 298 | contexts_updated_t bundle = NULL; |
| 299 | |
| 300 | spinlock_lock(&host->rrdctx.hub_queue.spinlock); |
| 301 | Word_t idx = 0; |
| 302 | for(RRDCONTEXT *rc = RRDCONTEXT_QUEUE_FIRST(&host->rrdctx.hub_queue, &idx); |
| 303 | rc; |
| 304 | rc = RRDCONTEXT_QUEUE_NEXT(&host->rrdctx.hub_queue, &idx)) { |
| 305 | if(unlikely(!service_running(SERVICE_CONTEXT))) break; |
| 306 | |
| 307 | if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST)) |
| 308 | break; |
| 309 | |
| 310 | STRING *lookup_id = string_dup(rc->id); |
| 311 | spinlock_unlock(&host->rrdctx.hub_queue.spinlock); |
| 312 | |
| 313 | const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(lookup_id)); |
| 314 | bool do_it = item && (dictionary_acquired_item_value(item) == rc); |
| 315 | bool dispatch_ready = false; |
| 316 | |
| 317 | if(item) { |
| 318 | if (do_it) { |
| 319 | spinlock_lock(&host->rrdctx.hub_queue.spinlock); |
| 320 | |
| 321 | if(likely(service_running(SERVICE_CONTEXT))) { |
| 322 | RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.hub_queue, idx); |
| 323 | if(rc_at_idx == rc) { |
| 324 | worker_is_busy(WORKER_JOB_QUEUED); |
| 325 | usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut); |
| 326 | dispatch_ready = unlikely(now_ut >= dispatch_ut) && claim_id_is_set(claim_id); |
| 327 | } |
| 328 | else |
| 329 | do_it = false; |
| 330 | } |
| 331 | else |
| 332 | do_it = false; |
| 333 | |
| 334 | spinlock_unlock(&host->rrdctx.hub_queue.spinlock); |
| 335 | |
| 336 | if(dispatch_ready) { |
| 337 | worker_is_busy(WORKER_JOB_CHECK); |
| 338 | |
| 339 | rrdcontext_lock(rc); |
| 340 | |
| 341 | if(check_if_cloud_version_changed_unsafe(rc, true)) { |
| 342 | worker_is_busy(WORKER_JOB_SEND); |
| 343 | |
| 344 | if(!bundle) { |
| 345 | // prepare the bundle to send the messages |
| 346 | char uuid_str[UUID_STR_LEN]; |
| 347 | uuid_unparse_lower(host->node_id.uuid, uuid_str); |
| 348 | |
| 349 | bundle = contexts_updated_new(claim_id.str, uuid_str, 0, now_ut); |
| 350 | } |
| 351 | // update the hub data of the context, give a new version, pack the message |
| 352 | // and save an update to SQL |
| 353 | rrdcontext_message_send_unsafe(rc, false, bundle); |
| 354 | messages_added++; |
| 355 | |
| 356 | rc->queue.dispatches++; |
| 357 | rc->queue.dequeued_ut = now_ut; |
| 358 | } |
| 359 | else |
| 360 | rc->version = rc->hub.version; |
| 361 | |
| 362 | if(unlikely(rrdcontext_should_be_deleted(rc))) { |
| 363 | // this is a deleted context - delete it forever... |
| 364 | |
| 365 | worker_is_busy(WORKER_JOB_CLEANUP_DELETE); |
| 366 | |
| 367 | rrdcontext_dequeue_from_post_processing(rc); |
| 368 | rrdcontext_delete_from_sql_unsafe(rc); |
| 369 | |
| 370 | STRING *delete_id = string_dup(rc->id); |
| 371 | rrdcontext_unlock(rc); |
| 372 | |
| 373 | // delete it from the master dictionary |
| 374 | if(!dictionary_del(host->rrdctx.contexts, string2str(delete_id))) |
| 375 | netdata_log_error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.", |
| 376 | string2str(delete_id), rrdhost_hostname(host)); |
| 377 | |
| 378 | string_freez(delete_id); |
| 379 | } |
| 380 | else |
| 381 | rrdcontext_unlock(rc); |
| 382 | } |
| 383 | else |
| 384 | do_it = false; |
| 385 | } |
| 386 | |
| 387 | dictionary_acquired_item_release(host->rrdctx.contexts, item); |
| 388 | } |
| 389 | string_freez(lookup_id); |
| 390 | |
| 391 | spinlock_lock(&host->rrdctx.hub_queue.spinlock); |
| 392 | if(unlikely(!service_running(SERVICE_CONTEXT))) |
| 393 | break; |
| 394 | |
| 395 | if(do_it) { |
| 396 | worker_is_busy(WORKER_JOB_DEQUEUE); |
| 397 | RRDCONTEXT *rc_at_idx = RRDCONTEXT_QUEUE_GET(&host->rrdctx.hub_queue, idx); |
| 398 | if(rc_at_idx == rc) |
| 399 | rrdcontext_del_from_hub_queue(rc, true); |
| 400 | } |
| 401 | } |
| 402 | spinlock_unlock(&host->rrdctx.hub_queue.spinlock); |
| 403 | |
| 404 | if(service_running(SERVICE_CONTEXT) && bundle) { |
| 405 | // we have a bundle to send messages |
| 406 | |
| 407 | // update the version hash |
| 408 | contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host)); |
| 409 | |
| 410 | // send it |
| 411 | aclk_send_contexts_updated(bundle); |
| 412 | } |
| 413 | else if(bundle) |
| 414 | contexts_updated_delete(bundle); |
| 415 | } |