do not rely on the queued flag to queue a context (#18198)
Costa Tsaousis committed
Jul 18, 2024 at 21:53 UTC
e422c8dace5ef33abfb5b26874360e208f4c688d
2 files changed
+24
-27
src/database/contexts/context.c
+4
@@ -255,6 +255,10 @@ static void rrdcontext_post_processing_queue_insert_callback(const DICTIONARY_IT
255
256
static void rrdcontext_post_processing_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
257
RRDCONTEXT *rc = context;
258
+
259
+ // IMPORTANT:
260
+ // Do not rely on this flag being absent, because the dictionaries have delayed deletions (garbage collect)
261
+ // so, this flag may not be deleted immediately from the context.
262
rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_PP);
263
rc->pp.dequeued_ut = now_realtime_usec();
264
}
src/database/contexts/worker.c
+20
-27
@@ -753,30 +753,26 @@ static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
753
void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function __maybe_unused, RRD_FLAGS flags __maybe_unused) {
754
if(unlikely(!rc->rrdhost->rrdctx.pp_queue)) return;
755
756
- if(!rrd_flag_check(rc, RRD_FLAG_QUEUED_FOR_PP)) {
757
- dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.pp_queue,
758
- string2str(rc->id),
759
- rc,
760
- sizeof(*rc));
761
-
762
-#if(defined(NETDATA_INTERNAL_CHECKS) && defined(LOG_POST_PROCESSING_QUEUE_INSERTIONS))
763
- {
764
- BUFFER *wb_flags = buffer_create(1000);
765
- rrd_flags_to_buffer(flags, wb_flags);
766
-
767
- BUFFER *wb_reasons = buffer_create(1000);
768
- rrd_reasons_to_buffer(flags, wb_reasons);
769
-
770
- internal_error(true, "RRDCONTEXT: '%s' update triggered by function %s(), due to flags: %s, reasons: %s",
771
- string2str(rc->id), function,
772
- buffer_tostring(wb_flags),
773
- buffer_tostring(wb_reasons));
774
-
775
- buffer_free(wb_reasons);
776
- buffer_free(wb_flags);
777
- }
778
-#endif
756
+#if 0
757
+ if(string_strcmp(rc->id, "system.cpu") == 0) {
758
+ CLEAN_BUFFER *wb = buffer_create(0, NULL);
759
+ buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_MINIFY);
760
+ buffer_json_member_add_array(wb, "flags");
761
+ rrd_flags_to_buffer_json_array_items(rc->flags, wb);
762
+ buffer_json_array_close(wb);
763
+ buffer_json_member_add_array(wb, "reasons");
764
+ rrd_reasons_to_buffer_json_array_items(rc->flags, wb);
765
+ buffer_json_array_close(wb);
766
+ buffer_json_finalize(wb);
767
+ nd_log(NDLS_DAEMON, NDLP_EMERG, "%s() context '%s', triggered: %s",
768
+ function, string2str(rc->id), buffer_tostring(wb));
769
}
770
+#endif
771
+
772
+ dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.pp_queue,
773
+ string2str(rc->id),
774
+ rc,
775
+ sizeof(*rc));
776
}
777
778
static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc) {
@@ -960,16 +956,13 @@ static void rrdcontext_dequeue_from_hub_queue(RRDCONTEXT *rc) {
956
static void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
957
958
// check if we have received a streaming command for this host
963
- if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_connected || !host->rrdctx.hub_queue)
959
+ if(!host->node_id || !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_connected || !host->rrdctx.hub_queue)
960
return;
961
962
// check if there are queued items to send
963
if(!dictionary_entries(host->rrdctx.hub_queue))
964
return;
965
970
- if(!host->node_id)
971
- return;
972
-
966
size_t messages_added = 0;
967
contexts_updated_t bundle = NULL;
968