@cryptotaxi247 / netdata-1 / commits / e46080f64

Fix memory leaks and service thread corruption (#20159)

* fix debug log of used dictionaries * fix cleanup of rrdcontexts * use judy queues instead of dictionary queues * rework service thread cleanup to ensure no collisions with collectors * mark all charts obsolete if a child does not reconnect * mark all charts of exited plugins as obsolete * exclusive destroy lock for RRDDIM * cleanup rrdcontext queues on free * increase stack trace array to 100 * fix memory leak in ML MODELS when the models are ignored * make dictionary traversals cleanup if the dictionary that is traversed, is destroyed mid traversal * fix debug log of used dictionaries * fix cleanup of rrdcontexts * use judy queues instead of dictionary queues * rework service thread cleanup to ensure no collisions with collectors * mark all charts obsolete if a child does not reconnect * mark all charts of exited plugins as obsolete * exclusive destroy lock for RRDDIM * cleanup rrdcontext queues on free * increase stack trace array to 100 * fix memory leak in ML MODELS when the models are ignored * make dictionary traversals cleanup if the dictionary that is traversed, is destroyed mid traversal * make referenced directories a warning

Costa Tsaousis committed Apr 24, 2025 at 19:23 UTC e46080f64fef3444f1714f698cc6a5e5dcd06d92
45 files changed +711 -572
CMakeLists.txt
+8 -7
@@ -1598,15 +1598,15 @@ set(RRD_PLUGIN_FILES
1598 src/database/contexts/api_v2_contexts_alerts.h
1599 src/database/contexts/api_v2_contexts_alert_transitions.c
1600 src/database/contexts/api_v2_contexts_alert_config.c
1601 - src/database/contexts/context.c
1602 - src/database/contexts/instance.c
1603 - src/database/contexts/internal.h
1604 - src/database/contexts/metric.c
1601 + src/database/contexts/rrdcontext-context.c
1602 + src/database/contexts/rrdcontext-instance.c
1603 + src/database/contexts/rrdcontext-internal.h
1604 + src/database/contexts/rrdcontext-metric.c
1605 src/database/contexts/query_scope.c
1606 src/database/contexts/query_target.c
1607 src/database/contexts/rrdcontext.c
1608 src/database/contexts/rrdcontext.h
1609 - src/database/contexts/worker.c
1609 + src/database/contexts/rrdcontext-worker.c
1610 src/database/rrdcollector.c
1611 src/database/rrdcollector.h
1612 src/database/rrddim.c
@@ -1658,7 +1658,7 @@ set(RRD_PLUGIN_FILES
1658 src/database/rrd-database-mode.c
1659 src/database/rrdhost-system-info.c
1660 src/database/rrdhost-system-info.h
1661 - src/database/contexts/contexts-loading.c
1661 + src/database/contexts/rrdcontext-loading.c
1662 src/database/rrdset-index-id.c
1663 src/database/rrdset-index-id.h
1664 src/database/rrdset-index-name.c
@@ -1688,6 +1688,7 @@ set(RRD_PLUGIN_FILES
1688 src/database/rrdhost-collection.h
1689 src/database/pattern-array.c
1690 src/database/pattern-array.h
1691 + src/database/contexts/rrdcontext-queues.c
1692 )
1693
1694 if(ENABLE_DBENGINE)
@@ -2134,7 +2135,7 @@ set(ACLK_FILES
2135 src/aclk/schema-wrappers/context_stream.cc
2136 src/aclk/schema-wrappers/context_stream.h
2137 src/aclk/schema-wrappers/context.cc
2137 - src/aclk/schema-wrappers/context.h
2138 + src/aclk/schema-wrappers/rrdcontext-context.h
2139 src/aclk/schema-wrappers/schema_wrappers.h
2140 src/aclk/schema-wrappers/schema_wrapper_utils.cc
2141 src/aclk/schema-wrappers/schema_wrapper_utils.h
src/aclk/schema-wrappers/context.cc
+1 -1
@@ -6,7 +6,7 @@
6
7 #include "schema_wrapper_utils.h"
8
9 -#include "context.h"
9 +#include "rrdcontext-context.h"
10
11 using namespace context::v1;
12
src/aclk/schema-wrappers/rrdcontext-context.h renamed
src/aclk/schema-wrappers/schema_wrappers.h
+1 -1
@@ -13,7 +13,7 @@
13 #include "node_info.h"
14 #include "capability.h"
15 #include "context_stream.h"
16 -#include "context.h"
16 +#include "rrdcontext-context.h"
17 #include "agent_cmds.h"
18
19 #endif /* SCHEMA_WRAPPERS_H */
src/collectors/cgroups.plugin/cgroup-charts.c
-5
@@ -90,8 +90,6 @@ void update_cpu_utilization_limit_chart(struct cgroup *cg, NETDATA_DOUBLE cpu_li
90 cpu_usage = (NETDATA_DOUBLE)(cg->cpuacct_stat.user + cg->cpuacct_stat.system) * 100;
91 NETDATA_DOUBLE cpu_used = 100 * (cpu_usage - cg->prev_cpu_usage) / (cpu_limit * cgroup_update_every);
92
93 - rrdset_isnot_obsolete___safe_from_collector_thread(chart);
94 -
93 rrddim_set(chart, "used", (cpu_used > 0) ? (collected_number)cpu_used : 0);
94
95 cg->prev_cpu_usage = cpu_usage;
@@ -480,8 +478,6 @@ void update_mem_usage_limit_chart(struct cgroup *cg, unsigned long long memory_l
478 rrddim_add(chart, "used", NULL, 1, 1024 * 1024, RRD_ALGORITHM_ABSOLUTE);
479 }
480
483 - rrdset_isnot_obsolete___safe_from_collector_thread(chart);
484 -
481 rrddim_set(chart, "available", (collected_number)(memory_limit - cg->memory.usage_in_bytes));
482 rrddim_set(chart, "used", (collected_number)cg->memory.usage_in_bytes);
483 rrdset_done(chart);
@@ -518,7 +514,6 @@ void update_mem_utilization_chart(struct cgroup *cg, unsigned long long memory_l
514 rrddim_add(chart, "utilization", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
515 }
516
521 - rrdset_isnot_obsolete___safe_from_collector_thread(chart);
517 collected_number util = (collected_number)(cg->memory.usage_in_bytes * 100 / memory_limit);
518 rrddim_set(chart, "utilization", util);
519 rrdset_done(chart);
src/collectors/proc.plugin/proc_mdstat.c
-14
@@ -408,8 +408,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
408 NETDATA_CHART_PRIO_MDSTAT_HEALTH,
409 update_every,
410 RRDSET_TYPE_LINE);
411 -
412 - rrdset_isnot_obsolete___safe_from_collector_thread(st_mdstat_health);
411 }
412
413 if (!redundant_num) {
@@ -457,8 +455,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
455 update_every,
456 RRDSET_TYPE_STACKED);
457
460 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_disks);
461 -
458 add_labels_to_mdstat(raid, raid->st_disks);
459 }
460
@@ -492,8 +488,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
488 update_every,
489 RRDSET_TYPE_LINE);
490
495 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_mismatch_cnt);
496 -
491 add_labels_to_mdstat(raid, raid->st_mismatch_cnt);
492 }
493
@@ -524,8 +518,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
518 update_every,
519 RRDSET_TYPE_LINE);
520
527 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_operation);
528 -
521 add_labels_to_mdstat(raid, raid->st_operation);
522 }
523
@@ -561,8 +553,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
553 NETDATA_CHART_PRIO_MDSTAT_FINISH + raid_idx * 10,
554 update_every, RRDSET_TYPE_LINE);
555
564 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_finish);
565 -
556 add_labels_to_mdstat(raid, raid->st_finish);
557 }
558
@@ -590,8 +580,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
580 update_every,
581 RRDSET_TYPE_LINE);
582
593 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_speed);
594 -
583 add_labels_to_mdstat(raid, raid->st_speed);
584 }
585
@@ -622,8 +610,6 @@ int do_proc_mdstat(int update_every, usec_t dt)
610 update_every,
611 RRDSET_TYPE_LINE);
612
625 - rrdset_isnot_obsolete___safe_from_collector_thread(raid->st_nonredundant);
626 -
613 add_labels_to_mdstat(raid, raid->st_nonredundant);
614 }
615
src/collectors/statsd.plugin/statsd.c
+21 -19
@@ -483,10 +483,7 @@ static inline void metric_update_counters_and_obsoletion(STATSD_METRIC *m) {
483 m->events++;
484 m->count++;
485 m->last_collected = now_realtime_sec();
486 - if (m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE))) {
487 - rrdset_isnot_obsolete___safe_from_collector_thread(m->st);
488 - m->options &= ~STATSD_METRIC_OPTION_OBSOLETE;
489 - }
486 + m->options &= ~STATSD_METRIC_OPTION_OBSOLETE;
487 }
488
489 static inline void statsd_process_gauge(STATSD_METRIC *m, const char *value, const char *sampling) {
@@ -1637,8 +1634,8 @@ static inline RRDSET *statsd_private_rrdset_create(
1634 static inline void statsd_private_chart_gauge(STATSD_METRIC *m) {
1635 netdata_log_debug(D_STATSD, "updating private chart for gauge metric '%s'", m->name);
1636
1640 - if(m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE)))
1641 - return;
1637 + if(m->options & STATSD_METRIC_OPTION_OBSOLETE)
1638 + return;
1639
1640 if(unlikely(!m->st || m->options & STATSD_METRIC_OPTION_UPDATED_CHART_METADATA)) {
1641 m->options &= ~STATSD_METRIC_OPTION_UPDATED_CHART_METADATA;
@@ -1680,7 +1677,7 @@ static inline void statsd_private_chart_gauge(STATSD_METRIC *m) {
1677 static inline void statsd_private_chart_counter_or_meter(STATSD_METRIC *m, const char *dim, const char *family) {
1678 netdata_log_debug(D_STATSD, "updating private chart for %s metric '%s'", dim, m->name);
1679
1683 - if(m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE)))
1680 + if(m->options & STATSD_METRIC_OPTION_OBSOLETE)
1681 return;
1682
1683 if(unlikely(!m->st || m->options & STATSD_METRIC_OPTION_UPDATED_CHART_METADATA)) {
@@ -1723,8 +1720,8 @@ static inline void statsd_private_chart_counter_or_meter(STATSD_METRIC *m, const
1720 static inline void statsd_private_chart_set(STATSD_METRIC *m) {
1721 netdata_log_debug(D_STATSD, "updating private chart for set metric '%s'", m->name);
1722
1726 - if(m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE)))
1727 - return;
1723 + if(m->options & STATSD_METRIC_OPTION_OBSOLETE)
1724 + return;
1725
1726 if(unlikely(!m->st || m->options & STATSD_METRIC_OPTION_UPDATED_CHART_METADATA)) {
1727 m->options &= ~STATSD_METRIC_OPTION_UPDATED_CHART_METADATA;
@@ -1766,8 +1763,8 @@ static inline void statsd_private_chart_set(STATSD_METRIC *m) {
1763 static inline void statsd_private_chart_dictionary(STATSD_METRIC *m) {
1764 netdata_log_debug(D_STATSD, "updating private chart for dictionary metric '%s'", m->name);
1765
1769 - if(m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE)))
1770 - return;
1766 + if(m->options & STATSD_METRIC_OPTION_OBSOLETE)
1767 + return;
1768
1769 if(unlikely(!m->st || m->options & STATSD_METRIC_OPTION_UPDATED_CHART_METADATA)) {
1770 m->options &= ~STATSD_METRIC_OPTION_UPDATED_CHART_METADATA;
@@ -1812,8 +1809,8 @@ static inline void statsd_private_chart_dictionary(STATSD_METRIC *m) {
1809 static inline void statsd_private_chart_timer_or_histogram(STATSD_METRIC *m, const char *dim, const char *family, const char *units) {
1810 netdata_log_debug(D_STATSD, "updating private chart for %s metric '%s'", dim, m->name);
1811
1815 - if(m->st && unlikely(rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE)))
1816 - return;
1812 + if(m->options & STATSD_METRIC_OPTION_OBSOLETE)
1813 + return;
1814
1815 if(unlikely(!m->st || m->options & STATSD_METRIC_OPTION_UPDATED_CHART_METADATA)) {
1816 m->options &= ~STATSD_METRIC_OPTION_UPDATED_CHART_METADATA;
@@ -1868,11 +1865,16 @@ static inline void statsd_private_chart_timer_or_histogram(STATSD_METRIC *m, con
1865 // statsd flush metrics
1866
1867 static inline void metric_check_obsoletion(STATSD_METRIC *m) {
1871 - if(statsd.set_obsolete_after &&
1872 - !rrdset_flag_check(m->st, RRDSET_FLAG_OBSOLETE) &&
1873 - m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED &&
1874 - m->last_collected + (time_t)statsd.set_obsolete_after < now_realtime_sec()) {
1875 - rrdset_is_obsolete___safe_from_collector_thread(m->st);
1868 + if(!(m->options & STATSD_METRIC_OPTION_OBSOLETE) &&
1869 + statsd.set_obsolete_after &&
1870 + (m->options & STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED) &&
1871 + m->last_collected + (time_t)statsd.set_obsolete_after < now_realtime_sec()) {
1872 +
1873 + if(m->st) {
1874 + rrdset_is_obsolete___safe_from_collector_thread(m->st);
1875 + m->st = NULL;
1876 + }
1877 +
1878 m->options |= STATSD_METRIC_OPTION_OBSOLETE;
1879 }
1880 }
@@ -2365,7 +2367,7 @@ static inline void statsd_flush_index_metrics(STATSD_INDEX *index, void (*flush_
2367 }
2368 dfe_done(m);
2369
2368 - // flush all the useful metrics
2370 + // flush all the unuseful metrics
2371 STATSD_METRIC *m_prev;
2372 for(m_prev = m = index->first_useful; m ; m = m->next_useful) {
2373 flush_metric(m);
src/daemon/service.c
+38 -60
@@ -5,9 +5,6 @@
5 /* Run service jobs every X seconds */
6 #define SERVICE_HEARTBEAT 10
7
8 -#define TIME_TO_RUN_OBSOLETIONS_ON_CHILD_CONNECT (3600 / 2)
9 -#define ITERATIONS_TO_RUN_OBSOLETIONS_ON_CHILD_CONNECT 60
10 -
8 #define WORKER_JOB_CHILD_CHART_OBSOLETION_CHECK 1
9 #define WORKER_JOB_CLEANUP_OBSOLETE_CHARTS 2
10 #define WORKER_JOB_ARCHIVE_CHART 3
@@ -23,31 +20,30 @@
20 #define WORKER_JOB_PGC_OPEN_EVICT 18
21 #define WORKER_JOB_PGC_OPEN_FLUSH 19
22
26 -static void svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
23 +static bool svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
24 RRDSET *st = rd->rrdset;
25
29 - if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED) || !rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))
30 - return;
26 + if(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE) && spinlock_trylock(&rd->destroy_lock)) {
27 + if(!rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE)) {
28 + spinlock_unlock(&rd->destroy_lock);
29 + return false;
30 + }
31 + }
32 + else
33 + return false;
34
35 worker_is_busy(WORKER_JOB_ARCHIVE_DIMENSION);
36
34 - rrddim_flag_set(rd, RRDDIM_FLAG_ARCHIVED);
35 - rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE);
36 -
37 if (rd->rrd_memory_mode == RRD_DB_MODE_DBENGINE) {
38 - /* only a collector can mark a chart as obsolete, so we must remove the reference */
38 if (!rrddim_finalize_collection_and_check_retention(rd)) {
39 /* This metric has no data and no references */
40 metaqueue_delete_dimension_uuid(uuidmap_uuid_ptr(rd->uuid));
41 }
43 - else {
44 - /* Do not delete this dimension */
45 - return;
46 - }
42 }
43
44 worker_is_busy(WORKER_JOB_FREE_DIMENSION);
45 rrddim_free(st, rd);
46 + return true;
47 }
48
49 static inline bool svc_rrdset_archive_obsolete_dimensions(RRDSET *st, bool all_dimensions) {
@@ -75,8 +71,8 @@ static inline bool svc_rrdset_archive_obsolete_dimensions(RRDSET *st, bool all_d
71 if(references == 1) {
72 // netdata_log_info("Removing obsolete dimension 'host:%s/chart:%s/dim:%s'",
73 // rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
78 - svc_rrddim_obsolete_to_archive(rd);
79 - dim_archives++;
74 + if(svc_rrddim_obsolete_to_archive(rd))
75 + dim_archives++;
76 }
77 // else
78 // netdata_log_info("Cannot remove obsolete dimension 'host:%s/chart:%s/dim:%s'",
@@ -94,18 +90,19 @@ static inline bool svc_rrdset_archive_obsolete_dimensions(RRDSET *st, bool all_d
90 return true;
91 }
92
97 -static void svc_rrdset_obsolete_to_free(RRDSET *st) {
98 - if(!svc_rrdset_archive_obsolete_dimensions(st, true))
99 - return;
93 +static bool svc_rrdset_lock_for_deletion(RRDSET *st, time_t now) {
94 + if(st->last_accessed_time_s + rrdset_free_obsolete_time_s < now &&
95 + st->last_updated.tv_sec + rrdset_free_obsolete_time_s < now &&
96 + st->last_collected_time.tv_sec + rrdset_free_obsolete_time_s < now &&
97 + spinlock_trylock(&st->destroy_lock)) {
98
101 - worker_is_busy(WORKER_JOB_FREE_CHART);
99 + if(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE))
100 + return true;
101
103 - rrdcalc_unlink_and_delete_all_rrdset_alerts(st);
104 -
105 - // has to be run after all dimensions are archived - or use-after-free will occur
106 - rrdvar_delete_all(st->rrdvars);
102 + spinlock_unlock(&st->destroy_lock);
103 + }
104
108 - rrdset_free(st);
105 + return false;
106 }
107
108 static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {
@@ -128,25 +125,26 @@ static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {
125 continue;
126
127 RRDSET_FLAGS flags = rrdset_flag_get(st);
131 - bool obsolete_chart = flags & RRDSET_FLAG_OBSOLETE;
132 - bool obsolete_dims = flags & RRDSET_FLAG_OBSOLETE_DIMENSIONS;
128
134 - if(obsolete_dims) {
129 + if(flags & RRDSET_FLAG_OBSOLETE_DIMENSIONS) {
130 partial_candidates++;
131
132 if(svc_rrdset_archive_obsolete_dimensions(st, false))
133 partial_archives++;
134 }
135
141 - if(obsolete_chart) {
136 + if(flags & RRDSET_FLAG_OBSOLETE) {
137 full_candidates++;
138
144 - if(unlikely( st->last_accessed_time_s + rrdset_free_obsolete_time_s < now
145 - && st->last_updated.tv_sec + rrdset_free_obsolete_time_s < now
146 - && st->last_collected_time.tv_sec + rrdset_free_obsolete_time_s < now
147 - )) {
148 - svc_rrdset_obsolete_to_free(st);
149 - full_archives++;
139 + if(svc_rrdset_lock_for_deletion(st, now)) {
140 + if(svc_rrdset_archive_obsolete_dimensions(st, true)) {
141 + full_archives++;
142 +
143 + worker_is_busy(WORKER_JOB_FREE_CHART);
144 + rrdset_free(st);
145 + }
146 + else
147 + spinlock_unlock(&st->destroy_lock);
148 }
149 }
150 }
@@ -161,27 +159,10 @@ static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {
159 rrdhost_flag_set(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS);
160 }
161
164 -static void svc_rrdhost_detect_obsolete_charts(RRDHOST *host) {
165 - worker_is_busy(WORKER_JOB_CHILD_CHART_OBSOLETION_CHECK);
166 -
167 - time_t now = now_realtime_sec();
168 - time_t last_entry_t;
162 +void svc_rrdhost_obsolete_all_charts(RRDHOST *host) {
163 RRDSET *st;
170 -
171 - time_t child_connect_time = host->stream.rcv.status.last_connected;
172 -
164 rrdset_foreach_read(st, host) {
174 - if(rrdset_is_replicating(st))
175 - continue;
176 -
177 - last_entry_t = rrdset_last_entry_s(st);
178 -
179 - if (last_entry_t && last_entry_t < child_connect_time &&
180 - child_connect_time + TIME_TO_RUN_OBSOLETIONS_ON_CHILD_CONNECT +
181 - (ITERATIONS_TO_RUN_OBSOLETIONS_ON_CHILD_CONNECT * st->update_every) <
182 - now)
183 -
184 - rrdset_is_obsolete___safe_from_collector_thread(st);
165 + rrdset_is_obsolete___safe_from_collector_thread(st);
166 }
167 rrdset_foreach_done(st);
168 }
@@ -209,12 +190,9 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
190
191 time_t now = now_realtime_sec();
192
212 - if (host->stream.rcv.status.check_obsolete &&
213 - ((host->stream.rcv.status.last_chart &&
214 - host->stream.rcv.status.last_chart + host->health.delay_up_to < now) ||
215 - (host->stream.rcv.status.last_connected + TIME_TO_RUN_OBSOLETIONS_ON_CHILD_CONNECT < now))) {
216 - svc_rrdhost_detect_obsolete_charts(host);
217 - host->stream.rcv.status.check_obsolete = false;
193 + if (host->stream.rcv.status.last_connected == 0 &&
194 + (host->stream.rcv.status.last_disconnected + rrdset_free_obsolete_time_s < now)) {
195 + svc_rrdhost_obsolete_all_charts(host);
196 }
197
198 rrdhost_receiver_unlock(host);
src/database/contexts/api_v1_contexts.c
+1 -1
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 static void rrd_flags_to_buffer_json_array_items(RRD_FLAGS flags, BUFFER *wb) {
6 if(flags & RRD_FLAG_QUEUED_FOR_HUB)
src/database/contexts/api_v2_contexts.h
+1 -1
@@ -3,7 +3,7 @@
3 #ifndef NETDATA_API_V2_CONTEXTS_H
4 #define NETDATA_API_V2_CONTEXTS_H
5
6 -#include "internal.h"
6 +#include "rrdcontext-internal.h"
7
8 typedef enum __attribute__ ((__packed__)) {
9 FTS_MATCHED_NONE = 0,
src/database/contexts/api_v2_contexts_alerts.h
+1 -1
@@ -3,7 +3,7 @@
3 #ifndef NETDATA_API_V2_CONTEXTS_ALERTS_H
4 #define NETDATA_API_V2_CONTEXTS_ALERTS_H
5
6 -#include "internal.h"
6 +#include "rrdcontext-internal.h"
7 #include "api_v2_contexts.h"
8
9 struct alert_transitions_callback_data {
src/database/contexts/query_scope.c
+2 -2
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 ssize_t query_scope_foreach_host(SIMPLE_PATTERN *scope_hosts_sp, SIMPLE_PATTERN *hosts_sp,
6 foreach_host_cb_t cb, void *data,
@@ -50,7 +50,7 @@ ssize_t query_scope_foreach_host(SIMPLE_PATTERN *scope_hosts_sp, SIMPLE_PATTERN
50 bool queryable_host = (match == SP_MATCHED_POSITIVE);
51
52 v_hash += dictionary_version(host->rrdctx.contexts);
53 - h_hash += dictionary_version(host->rrdctx.hub_queue);
53 + h_hash += rrdcontext_queue_version(&host->rrdctx.hub_queue);
54 a_hash += dictionary_version(host->rrdcalc_root_index);
55 t_hash += __atomic_load_n(&host->health_transitions, __ATOMIC_RELAXED);
56 ssize_t ret = cb(data, host, queryable_host);
src/database/contexts/query_target.c
+2 -2
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 #define QUERY_TARGET_MAX_REALLOC_INCREASE 500
6 #define query_target_realloc_size(size, start) \
@@ -1128,7 +1128,7 @@ QUERY_TARGET *query_target_create(QUERY_TARGET_REQUEST *qtr) {
1128
1129 // single host query
1130 qt->versions.contexts_hard_hash = dictionary_version(host->rrdctx.contexts);
1131 - qt->versions.contexts_soft_hash = dictionary_version(host->rrdctx.hub_queue);
1131 + qt->versions.contexts_soft_hash = rrdcontext_queue_version(&host->rrdctx.hub_queue);
1132 qt->versions.alerts_hard_hash = dictionary_version(host->rrdcalc_root_index);
1133 qt->versions.alerts_soft_hash = __atomic_load_n(&host->health_transitions, __ATOMIC_RELAXED);
1134 query_node_add(&qtl, host, true);
src/database/contexts/rrdcontext-context.c renamed
+14 -96
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 inline const char *rrdcontext_acquired_id(RRDCONTEXT_ACQUIRED *rca) {
6 RRDCONTEXT *rc = rrdcontext_acquired_value(rca);
@@ -96,6 +96,9 @@ static void rrdcontext_delete_callback(const DICTIONARY_ITEM *item __maybe_unuse
96 // update the count of contexts
97 __atomic_sub_fetch(&rc->rrdhost->rrdctx.contexts_count, 1, __ATOMIC_RELAXED);
98
99 + rrdcontext_del_from_hub_queue(rc, false);
100 + rrdcontext_del_from_pp_queue(rc, false);
101 +
102 rrdinstances_destroy_from_rrdcontext(rc);
103 rrdcontext_freez(rc);
104 }
@@ -229,64 +232,6 @@ void rrdcontext_trigger_updates(RRDCONTEXT *rc, const char *function) {
232 rrdcontext_queue_for_post_processing(rc, function, rc->flags);
233 }
234
232 -static void rrdcontext_hub_queue_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
233 - RRDCONTEXT *rc = context;
234 - rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
235 - rc->queue.queued_ut = now_realtime_usec();
236 - rc->queue.queued_flags = rrd_flags_get(rc);
237 -}
238 -
239 -static void rrdcontext_hub_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
240 - RRDCONTEXT *rc = context;
241 - rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB);
242 -}
243 -
244 -static bool rrdcontext_hub_queue_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *new_context __maybe_unused, void *nothing __maybe_unused) {
245 - // context and new_context are the same
246 - // we just need to update the timings
247 - RRDCONTEXT *rc = context;
248 - rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
249 - rc->queue.queued_ut = now_realtime_usec();
250 - rc->queue.queued_flags |= rrd_flags_get(rc);
251 -
252 - return true;
253 -}
254 -
255 -static void rrdcontext_post_processing_queue_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
256 - RRDCONTEXT *rc = context;
257 - rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP);
258 - rc->pp.queued_flags = rc->flags;
259 - rc->pp.queued_ut = now_realtime_usec();
260 -}
261 -
262 -static void rrdcontext_post_processing_queue_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *nothing __maybe_unused) {
263 - RRDCONTEXT *rc = context;
264 -
265 - // IMPORTANT:
266 - // Do not rely on this flag being absent, because the dictionaries have delayed deletions (garbage collect)
267 - // so, this flag may not be deleted immediately from the context.
268 - rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_PP);
269 - rc->pp.dequeued_ut = now_realtime_usec();
270 -}
271 -
272 -static bool rrdcontext_post_processing_queue_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *context, void *new_context __maybe_unused, void *nothing __maybe_unused) {
273 - RRDCONTEXT *rc = context;
274 - bool changed = false;
275 -
276 - if(!(rc->flags & RRD_FLAG_QUEUED_FOR_PP)) {
277 - rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP);
278 - changed = true;
279 - }
280 -
281 - if(rc->pp.queued_flags != rc->flags) {
282 - rc->pp.queued_flags |= rc->flags;
283 - changed = true;
284 - }
285 -
286 - return changed;
287 -}
288 -
289 -
235 void rrdhost_create_rrdcontexts(RRDHOST *host) {
236 if(unlikely(!host)) return;
237 if(likely(host->rrdctx.contexts)) return;
@@ -300,49 +245,22 @@ void rrdhost_create_rrdcontexts(RRDHOST *host) {
245 dictionary_register_conflict_callback(host->rrdctx.contexts, rrdcontext_conflict_callback, host);
246 dictionary_register_react_callback(host->rrdctx.contexts, rrdcontext_react_callback, host);
247
303 - host->rrdctx.hub_queue = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE, &dictionary_stats_category_rrdcontext, 0);
304 - dictionary_register_insert_callback(host->rrdctx.hub_queue, rrdcontext_hub_queue_insert_callback, NULL);
305 - dictionary_register_delete_callback(host->rrdctx.hub_queue, rrdcontext_hub_queue_delete_callback, NULL);
306 - dictionary_register_conflict_callback(host->rrdctx.hub_queue, rrdcontext_hub_queue_conflict_callback, NULL);
248 + memset(&host->rrdctx.pp_queue, 0, sizeof(host->rrdctx.pp_queue));
249 + RRDCONTEXT_QUEUE_INIT(&host->rrdctx.pp_queue);
250 + spinlock_init(&host->rrdctx.pp_queue.spinlock);
251
308 - host->rrdctx.pp_queue = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_VALUE_LINK_DONT_CLONE, &dictionary_stats_category_rrdcontext, 0);
309 - dictionary_register_insert_callback(host->rrdctx.pp_queue, rrdcontext_post_processing_queue_insert_callback, NULL);
310 - dictionary_register_delete_callback(host->rrdctx.pp_queue, rrdcontext_post_processing_queue_delete_callback, NULL);
311 - dictionary_register_conflict_callback(host->rrdctx.pp_queue, rrdcontext_post_processing_queue_conflict_callback, NULL);
252 + memset(&host->rrdctx.hub_queue, 0, sizeof(host->rrdctx.hub_queue));
253 + RRDCONTEXT_QUEUE_INIT(&host->rrdctx.hub_queue);
254 + spinlock_init(&host->rrdctx.hub_queue.spinlock);
255 }
256
257 void rrdhost_destroy_rrdcontexts(RRDHOST *host) {
258 if(unlikely(!host)) return;
259 if(unlikely(!host->rrdctx.contexts)) return;
260
318 - DICTIONARY *old;
319 -
320 - if(host->rrdctx.hub_queue) {
321 - old = host->rrdctx.hub_queue;
322 - host->rrdctx.hub_queue = NULL;
323 -
324 - RRDCONTEXT *rc;
325 - dfe_start_write(old, rc) {
326 - dictionary_del(old, string2str(rc->id));
327 - }
328 - dfe_done(rc);
329 - dictionary_destroy(old);
330 - }
331 -
332 - if(host->rrdctx.pp_queue) {
333 - old = host->rrdctx.pp_queue;
334 - host->rrdctx.pp_queue = NULL;
335 -
336 - RRDCONTEXT *rc;
337 - dfe_start_write(old, rc) {
338 - dictionary_del(old, string2str(rc->id));
339 - }
340 - dfe_done(rc);
341 - dictionary_destroy(old);
342 - }
343 -
344 - old = host->rrdctx.contexts;
261 + dictionary_destroy(host->rrdctx.contexts);
262 host->rrdctx.contexts = NULL;
346 - dictionary_destroy(old);
347 -}
263
264 + RRDCONTEXT_QUEUE_FREE(&host->rrdctx.pp_queue, NULL, NULL);
265 + RRDCONTEXT_QUEUE_FREE(&host->rrdctx.hub_queue, NULL, NULL);
266 +}
src/database/contexts/rrdcontext-instance.c renamed
+1 -1
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 // ----------------------------------------------------------------------------
6 // helper one-liners for RRDINSTANCE
src/database/contexts/rrdcontext-internal.h renamed
+19 -1
@@ -5,7 +5,7 @@
5
6 #include "rrdcontext.h"
7 #include "../sqlite/sqlite_context.h"
8 -#include "../../aclk/schema-wrappers/context.h"
8 +#include "../../aclk/schema-wrappers/rrdcontext-context.h"
9 #include "../../aclk/aclk_contexts_api.h"
10 #include "../../aclk/aclk.h"
11 #include "../storage-engine.h"
@@ -279,6 +279,7 @@ typedef struct rrdcontext {
279 RRDHOST *rrdhost;
280
281 struct {
282 + Word_t idx;
283 RRD_FLAGS queued_flags; // the last flags that triggered the post-processing
284 size_t executions; // how many times this context has been processed
285 usec_t queued_ut; // the last time this was queued
@@ -286,6 +287,7 @@ typedef struct rrdcontext {
287 } pp;
288
289 struct {
290 + Word_t idx;
291 RRD_FLAGS queued_flags; // the last flags that triggered the queueing
292 size_t dispatches; // the number of times this has been dispatched to hub
293 usec_t queued_ut; // the last time this was queued
@@ -295,6 +297,11 @@ typedef struct rrdcontext {
297 } queue;
298 } RRDCONTEXT;
299
300 +void rrdcontext_add_to_pp_queue(RRDCONTEXT *rc);
301 +void rrdcontext_add_to_hub_queue(RRDCONTEXT *rc);
302 +void rrdcontext_del_from_hub_queue(RRDCONTEXT *rc, bool having_lock);
303 +void rrdcontext_del_from_pp_queue(RRDCONTEXT *rc, bool having_lock);
304 +
305 // ----------------------------------------------------------------------------
306 // helpers for counting collected metrics, instances and contexts
307
@@ -484,4 +491,15 @@ void rrdcontext_initial_processing_after_loading(RRDCONTEXT *rc);
491
492 RRDLABELS *rrdinstance_labels(RRDINSTANCE *ri);
493
494 +bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs);
495 +void rrdcontext_post_process_queued_contexts(RRDHOST *host);
496 +void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut);
497 +usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut);
498 +bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending);
499 +bool rrdcontext_should_be_deleted(RRDCONTEXT *rc);
500 +void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc);
501 +static inline void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc) {
502 + rrdcontext_del_from_pp_queue(rc, false);
503 +}
504 +
505 #endif //NETDATA_RRDCONTEXT_INTERNAL_H
src/database/contexts/rrdcontext-loading.c renamed
+1 -1
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 static __thread size_t th_ignored_metrics = 0, th_ignored_instances = 0, th_zero_retention_metrics = 0;
6
src/database/contexts/rrdcontext-metric.c renamed
+2 -2
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 void rrdmetric_trigger_updates(RRDMETRIC *rm, const char *function);
6
@@ -300,7 +300,7 @@ inline void rrdmetric_updated_rrddim_flags(RRDDIM *rd) {
300 RRDMETRIC *rm = rrddim_get_rrdmetric(rd);
301 if(unlikely(!rm)) return;
302
303 - if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED|RRDDIM_FLAG_OBSOLETE))) {
303 + if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))) {
304 if(unlikely(rrd_flag_is_collected(rm)))
305 rrdmetric_set_archived(rm);
306 }
src/database/contexts/rrdcontext-queues.c new
+272
@@ -0,0 +1,272 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "rrdcontext-internal.h"
4 +
5 +typedef enum {
6 + RRDCONTEXT_QUEUE_INVALID = 0,
7 + RRDCONTEXT_QUEUE_ADDED,
8 + RRDCONTEXT_QUEUE_FOUND,
9 +} RRDCONTEXT_QUEUE_STATUS;
10 +
11 +static inline RRDCONTEXT_QUEUE_STATUS rrdcontext_queue_add(RRDCONTEXT_QUEUE_JudyLSet *queue, RRDCONTEXT *rc, Word_t *idx, bool having_lock) {
12 + RRDCONTEXT_QUEUE_STATUS ret = RRDCONTEXT_QUEUE_INVALID;
13 + if(!queue || !rc || !idx) return ret;
14 +
15 + if(!having_lock)
16 + spinlock_lock(&queue->spinlock);
17 +
18 + if(*idx) {
19 + fatal_assert(RRDCONTEXT_QUEUE_GET(queue, *idx) == rc);
20 + ret = RRDCONTEXT_QUEUE_FOUND;
21 + }
22 + else {
23 + *idx = queue->id++;
24 + RRDCONTEXT_QUEUE_SET(queue, *idx, rc);
25 + __atomic_add_fetch(&queue->version, 1, __ATOMIC_RELAXED);
26 + __atomic_add_fetch(&queue->entries, 1, __ATOMIC_RELAXED);
27 + ret = RRDCONTEXT_QUEUE_ADDED;
28 + }
29 +
30 + if(!having_lock)
31 + spinlock_unlock(&queue->spinlock);
32 +
33 + return ret;
34 +}
35 +
36 +void rrdcontext_add_to_hub_queue(RRDCONTEXT *rc) {
37 + if(!rc || !rc->rrdhost) return;
38 +
39 + spinlock_lock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
40 +
41 + RRDCONTEXT_QUEUE_STATUS ret = rrdcontext_queue_add(&rc->rrdhost->rrdctx.hub_queue, rc, &rc->queue.idx, true);
42 +
43 + if(ret == RRDCONTEXT_QUEUE_ADDED) {
44 + rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
45 + rc->queue.queued_ut = now_realtime_usec();
46 + rc->queue.queued_flags = rrd_flags_get(rc);
47 + }
48 + else if(ret == RRDCONTEXT_QUEUE_FOUND) {
49 + rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_HUB);
50 + rc->queue.queued_ut = now_realtime_usec();
51 + rc->queue.queued_flags |= rrd_flags_get(rc);
52 + }
53 +
54 + spinlock_unlock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
55 +}
56 +
57 +void rrdcontext_add_to_pp_queue(RRDCONTEXT *rc) {
58 + if(!rc || !rc->rrdhost) return;
59 +
60 + spinlock_lock(&rc->rrdhost->rrdctx.pp_queue.spinlock);
61 +
62 + RRDCONTEXT_QUEUE_STATUS ret = rrdcontext_queue_add(&rc->rrdhost->rrdctx.pp_queue, rc, &rc->pp.idx, true);
63 +
64 + if(ret == RRDCONTEXT_QUEUE_ADDED) {
65 + rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP);
66 + rc->pp.queued_flags = rc->flags;
67 + rc->pp.queued_ut = now_realtime_usec();
68 + }
69 + else if(ret == RRDCONTEXT_QUEUE_FOUND) {
70 + rrd_flag_set(rc, RRD_FLAG_QUEUED_FOR_PP);
71 + rc->pp.queued_flags |= rc->flags;
72 + }
73 +
74 + spinlock_unlock(&rc->rrdhost->rrdctx.pp_queue.spinlock);
75 +}
76 +
77 +static inline bool rrdcontext_queue_del(RRDCONTEXT_QUEUE_JudyLSet *queue, RRDCONTEXT *rc, Word_t *idx, bool having_lock) {
78 + bool ret = false;
79 + if(!queue || !rc || !idx) return ret;
80 +
81 + if(!having_lock)
82 + spinlock_lock(&queue->spinlock);
83 +
84 + RRDCONTEXT *rc_found = RRDCONTEXT_QUEUE_GET(queue, *idx);
85 +
86 + if(rc_found == rc) {
87 + RRDCONTEXT_QUEUE_DEL(queue, *idx);
88 + __atomic_add_fetch(&queue->version, 1, __ATOMIC_RELAXED);
89 + __atomic_sub_fetch(&queue->entries, 1, __ATOMIC_RELAXED);
90 + ret = true;
91 + }
92 + *idx = 0;
93 +
94 + if(!having_lock)
95 + spinlock_unlock(&queue->spinlock);
96 +
97 + return ret;
98 +}
99 +
100 +void rrdcontext_del_from_hub_queue(RRDCONTEXT *rc, bool having_lock) {
101 + if(!rc || !rc->rrdhost) return;
102 + if(!having_lock)
103 + spinlock_lock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
104 +
105 + if(rrdcontext_queue_del(&rc->rrdhost->rrdctx.hub_queue, rc, &rc->queue.idx, true)) {
106 + rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_HUB);
107 + }
108 +
109 + if(!having_lock)
110 + spinlock_unlock(&rc->rrdhost->rrdctx.hub_queue.spinlock);
111 +}
112 +
113 +void rrdcontext_del_from_pp_queue(RRDCONTEXT *rc, bool having_lock) {
114 + if(!rc || !rc->rrdhost) return;
115 +
116 + if(!having_lock)
117 + spinlock_lock(&rc->rrdhost->rrdctx.pp_queue.spinlock);
118 +
119 + if(rrdcontext_queue_del(&rc->rrdhost->rrdctx.pp_queue, rc, &rc->pp.idx, true)) {
120 + rrd_flag_clear(rc, RRD_FLAG_QUEUED_FOR_PP);
121 + rc->pp.dequeued_ut = now_realtime_usec();
122 + }
123 +
124 + if(!having_lock)
125 + spinlock_unlock(&rc->rrdhost->rrdctx.pp_queue.spinlock);
126 +}
127 +
128 +
129 +uint32_t rrdcontext_queue_version(RRDCONTEXT_QUEUE_JudyLSet *queue) {
130 + return __atomic_load_n(&queue->version, __ATOMIC_RELAXED);
131 +}
132 +
133 +int32_t rrdcontext_queue_entries(RRDCONTEXT_QUEUE_JudyLSet *queue) {
134 + return __atomic_load_n(&queue->entries, __ATOMIC_RELAXED);
135 +}
136 +
137 +void rrdcontext_post_process_queued_contexts(RRDHOST *host) {
138 +
139 + spinlock_lock(&host->rrdctx.pp_queue.spinlock);
140 + Word_t idx = 0;
141 + for(RRDCONTEXT *rc = RRDCONTEXT_QUEUE_FIRST(&host->rrdctx.pp_queue, &idx);
142 + rc;
143 + rc = RRDCONTEXT_QUEUE_NEXT(&host->rrdctx.pp_queue, &idx)) {
144 + if(unlikely(!service_running(SERVICE_CONTEXT))) break;
145 +
146 + const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(rc->id));
147 + bool do_it = dictionary_acquired_item_value(item) == rc;
148 +
149 + if(do_it)
150 + rrdcontext_del_from_pp_queue(rc, true);
151 +
152 + spinlock_unlock(&host->rrdctx.pp_queue.spinlock);
153 +
154 + if(item) {
155 + if (do_it)
156 + rrdcontext_post_process_updates(rc, false, RRD_FLAG_NONE, true);
157 +
158 + dictionary_acquired_item_release(host->rrdctx.contexts, item);
159 + }
160 +
161 + spinlock_lock(&host->rrdctx.pp_queue.spinlock);
162 + }
163 +
164 + spinlock_unlock(&host->rrdctx.pp_queue.spinlock);
165 +}
166 +
167 +void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
168 + // check if we have received a streaming command for this host
169 + if(UUIDiszero(host->node_id) || !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_online_for_contexts())
170 + return;
171 +
172 + // check if there are queued items to send
173 + if(!rrdcontext_queue_entries(&host->rrdctx.hub_queue))
174 + return;
175 +
176 + size_t messages_added = 0;
177 + contexts_updated_t bundle = NULL;
178 +
179 + spinlock_lock(&host->rrdctx.hub_queue.spinlock);
180 + Word_t idx = 0;
181 + for(RRDCONTEXT *rc = RRDCONTEXT_QUEUE_FIRST(&host->rrdctx.hub_queue, &idx);
182 + rc;
183 + rc = RRDCONTEXT_QUEUE_NEXT(&host->rrdctx.hub_queue, &idx)) {
184 + if(unlikely(!service_running(SERVICE_CONTEXT))) break;
185 +
186 + if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST))
187 + break;
188 +
189 + const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(host->rrdctx.contexts, string2str(rc->id));
190 + bool do_it = dictionary_acquired_item_value(item) == rc;
191 +
192 + if(do_it) {
193 + worker_is_busy(WORKER_JOB_DEQUEUE);
194 + rrdcontext_del_from_hub_queue(rc, true);
195 + }
196 +
197 + spinlock_unlock(&host->rrdctx.hub_queue.spinlock);
198 +
199 + if(item) {
200 + if (do_it) {
201 + worker_is_busy(WORKER_JOB_QUEUED);
202 + usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
203 + CLAIM_ID claim_id = claim_id_get();
204 +
205 + if(unlikely(now_ut >= dispatch_ut) && claim_id_is_set(claim_id)) {
206 + worker_is_busy(WORKER_JOB_CHECK);
207 +
208 + rrdcontext_lock(rc);
209 +
210 + if(check_if_cloud_version_changed_unsafe(rc, true)) {
211 + worker_is_busy(WORKER_JOB_SEND);
212 +
213 + if(!bundle) {
214 + // prepare the bundle to send the messages
215 + char uuid_str[UUID_STR_LEN];
216 + uuid_unparse_lower(host->node_id.uuid, uuid_str);
217 +
218 + bundle = contexts_updated_new(claim_id.str, uuid_str, 0, now_ut);
219 + }
220 + // update the hub data of the context, give a new version, pack the message
221 + // and save an update to SQL
222 + rrdcontext_message_send_unsafe(rc, false, bundle);
223 + messages_added++;
224 +
225 + rc->queue.dispatches++;
226 + rc->queue.dequeued_ut = now_ut;
227 + }
228 + else
229 + rc->version = rc->hub.version;
230 +
231 + if(unlikely(rrdcontext_should_be_deleted(rc))) {
232 + // this is a deleted context - delete it forever...
233 +
234 + worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
235 +
236 + rrdcontext_dequeue_from_post_processing(rc);
237 + rrdcontext_delete_from_sql_unsafe(rc);
238 +
239 + STRING *id = string_dup(rc->id);
240 + rrdcontext_unlock(rc);
241 +
242 + // delete it from the master dictionary
243 + if(!dictionary_del(host->rrdctx.contexts, string2str(rc->id)))
244 + netdata_log_error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
245 + string2str(id), rrdhost_hostname(host));
246 +
247 + string_freez(id);
248 + }
249 + else
250 + rrdcontext_unlock(rc);
251 + }
252 + }
253 +
254 + dictionary_acquired_item_release(host->rrdctx.contexts, item);
255 + }
256 +
257 + spinlock_lock(&host->rrdctx.hub_queue.spinlock);
258 + }
259 + spinlock_unlock(&host->rrdctx.hub_queue.spinlock);
260 +
261 + if(service_running(SERVICE_CONTEXT) && bundle) {
262 + // we have a bundle to send messages
263 +
264 + // update the version hash
265 + contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host));
266 +
267 + // send it
268 + aclk_send_contexts_updated(bundle);
269 + }
270 + else if(bundle)
271 + contexts_updated_delete(bundle);
272 +}
src/database/contexts/rrdcontext-worker.c renamed
+13 -148
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 static struct {
6 bool enabled;
@@ -14,17 +14,8 @@ static struct {
14 .active_vs_archived_percentage = 50,
15 };
16
17 -static void rrdcontext_dequeue_from_hub_queue(RRDCONTEXT *rc);
18 -
17 static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc);
18
21 -static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused);
22 -
23 -static void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc);
24 -
25 -static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc);
26 -static bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs);
27 -
19 static void rrdcontext_garbage_collect_for_all_hosts(void);
20
21 extern usec_t rrdcontext_next_db_rotation_ut;
@@ -260,7 +251,7 @@ static inline bool rrdinstance_should_be_deleted(RRDINSTANCE *ri) {
251 return true;
252 }
253
263 -static inline bool rrdcontext_should_be_deleted(RRDCONTEXT *rc) {
254 +bool rrdcontext_should_be_deleted(RRDCONTEXT *rc) {
255 if(likely(!rrd_flag_check(rc, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
256 return false;
257
@@ -620,7 +611,7 @@ static bool rrdinstance_forcefully_clear_retention(RRDCONTEXT *rc, size_t count,
611 return false;
612 }
613
623 -static bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs) {
614 +bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs) {
615 bool ret = false;
616
617 if(reason != RRD_FLAG_NONE)
@@ -804,11 +795,10 @@ static bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
795 }
796 }
797
807 - if(unlikely(rrd_flag_is_updated(rc) && rc->rrdhost->rrdctx.hub_queue)) {
798 + if(unlikely(rrd_flag_is_updated(rc))) {
799 if(check_if_cloud_version_changed_unsafe(rc, false)) {
800 rc->version = rrdcontext_get_next_version(rc);
810 - dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.hub_queue,
811 - string2str(rc->id), rc, sizeof(*rc));
801 + rrdcontext_add_to_hub_queue(rc);
802 }
803 }
804
@@ -819,8 +809,6 @@ static bool rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
809 }
810
811 void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function __maybe_unused, RRD_FLAGS flags __maybe_unused) {
822 - if(unlikely(!rc->rrdhost->rrdctx.pp_queue)) return;
823 -
812 #if 0
813 if(string_strcmp(rc->id, "system.cpu") == 0) {
814 CLEAN_BUFFER *wb = buffer_create(0, NULL);
@@ -837,15 +825,7 @@ void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function _
825 }
826 #endif
827
840 - dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.pp_queue,
841 - string2str(rc->id),
842 - rc,
843 - sizeof(*rc));
844 -}
845 -
846 -static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc) {
847 - if(unlikely(!rc->rrdhost->rrdctx.pp_queue)) return;
848 - dictionary_del(rc->rrdhost->rrdctx.pp_queue, string2str(rc->id));
828 + rrdcontext_add_to_pp_queue(rc);
829 }
830
831 void rrdcontext_initial_processing_after_loading(RRDCONTEXT *rc) {
@@ -854,24 +834,11 @@ void rrdcontext_initial_processing_after_loading(RRDCONTEXT *rc) {
834 }
835
836 void rrdcontext_delete_after_loading(RRDHOST *host, RRDCONTEXT *rc) {
857 - rrdcontext_dequeue_from_hub_queue(rc);
837 + rrdcontext_del_from_hub_queue(rc, false);
838 rrdcontext_dequeue_from_post_processing(rc);
839 dictionary_del(host->rrdctx.contexts, string2str(rc->id));
840 }
841
862 -static void rrdcontext_post_process_queued_contexts(RRDHOST *host) {
863 - if(unlikely(!host->rrdctx.pp_queue)) return;
864 -
865 - RRDCONTEXT *rc;
866 - dfe_start_reentrant(host->rrdctx.pp_queue, rc) {
867 - if(unlikely(!service_running(SERVICE_CONTEXT))) break;
868 -
869 - rrdcontext_dequeue_from_post_processing(rc);
870 - rrdcontext_post_process_updates(rc, false, RRD_FLAG_NONE, true);
871 - }
872 - dfe_done(rc);
873 -}
874 -
842 // ----------------------------------------------------------------------------
843 // dispatching contexts to cloud
844
@@ -933,7 +900,7 @@ void rrdcontext_message_send_unsafe(RRDCONTEXT *rc, bool snapshot __maybe_unused
900 }
901 }
902
936 -static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused) {
903 +bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused) {
904 bool id_changed = false,
905 title_changed = false,
906 units_changed = false,
@@ -1004,7 +971,7 @@ static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending _
971 return false;
972 }
973
1007 -static inline usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut) {
974 +usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut) {
975
976 if(likely(rc->queue.delay_calc_ut >= rc->queue.queued_ut))
977 return rc->queue.scheduled_dispatch_ut;
@@ -1031,102 +998,6 @@ static inline usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc
998 return dispatch_ut;
999 }
1000
1034 -static void rrdcontext_dequeue_from_hub_queue(RRDCONTEXT *rc) {
1035 - dictionary_del(rc->rrdhost->rrdctx.hub_queue, string2str(rc->id));
1036 -}
1037 -
1038 -static void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {
1039 -
1040 - // check if we have received a streaming command for this host
1041 - if(UUIDiszero(host->node_id) || !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_online_for_contexts() || !host->rrdctx.hub_queue)
1042 - return;
1043 -
1044 - // check if there are queued items to send
1045 - if(!dictionary_entries(host->rrdctx.hub_queue))
1046 - return;
1047 -
1048 - size_t messages_added = 0;
1049 - contexts_updated_t bundle = NULL;
1050 -
1051 - RRDCONTEXT *rc;
1052 - dfe_start_reentrant(host->rrdctx.hub_queue, rc) {
1053 - if(unlikely(!service_running(SERVICE_CONTEXT))) break;
1054 -
1055 - if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST))
1056 - break;
1057 -
1058 - worker_is_busy(WORKER_JOB_QUEUED);
1059 - usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
1060 - CLAIM_ID claim_id = claim_id_get();
1061 -
1062 - if(unlikely(now_ut >= dispatch_ut) && claim_id_is_set(claim_id)) {
1063 - worker_is_busy(WORKER_JOB_CHECK);
1064 -
1065 - rrdcontext_lock(rc);
1066 -
1067 - if(check_if_cloud_version_changed_unsafe(rc, true)) {
1068 - worker_is_busy(WORKER_JOB_SEND);
1069 -
1070 - if(!bundle) {
1071 - // prepare the bundle to send the messages
1072 - char uuid_str[UUID_STR_LEN];
1073 - uuid_unparse_lower(host->node_id.uuid, uuid_str);
1074 -
1075 - bundle = contexts_updated_new(claim_id.str, uuid_str, 0, now_ut);
1076 - }
1077 - // update the hub data of the context, give a new version, pack the message
1078 - // and save an update to SQL
1079 - rrdcontext_message_send_unsafe(rc, false, bundle);
1080 - messages_added++;
1081 -
1082 - rc->queue.dispatches++;
1083 - rc->queue.dequeued_ut = now_ut;
1084 - }
1085 - else
1086 - rc->version = rc->hub.version;
1087 -
1088 - // remove it from the queue
1089 - worker_is_busy(WORKER_JOB_DEQUEUE);
1090 - rrdcontext_dequeue_from_hub_queue(rc);
1091 -
1092 - if(unlikely(rrdcontext_should_be_deleted(rc))) {
1093 - // this is a deleted context - delete it forever...
1094 -
1095 - worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
1096 -
1097 - rrdcontext_dequeue_from_post_processing(rc);
1098 - rrdcontext_delete_from_sql_unsafe(rc);
1099 -
1100 - STRING *id = string_dup(rc->id);
1101 - rrdcontext_unlock(rc);
1102 -
1103 - // delete it from the master dictionary
1104 - if(!dictionary_del(host->rrdctx.contexts, string2str(rc->id)))
1105 - netdata_log_error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
1106 - string2str(id), rrdhost_hostname(host));
1107 -
1108 - string_freez(id);
1109 - }
1110 - else
1111 - rrdcontext_unlock(rc);
1112 - }
1113 - }
1114 - dfe_done(rc);
1115 -
1116 - if(service_running(SERVICE_CONTEXT) && bundle) {
1117 - // we have a bundle to send messages
1118 -
1119 - // update the version hash
1120 - contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host));
1121 -
1122 - // send it
1123 - aclk_send_contexts_updated(bundle);
1124 - }
1125 - else if(bundle)
1126 - contexts_updated_delete(bundle);
1127 -
1128 -}
1129 -
1001 // ----------------------------------------------------------------------------
1002 // worker thread
1003
@@ -1209,17 +1080,11 @@ void *rrdcontext_main(void *ptr) {
1080
1081 worker_is_busy(WORKER_JOB_HOSTS);
1082
1212 - if(host->rrdctx.pp_queue) {
1213 - pp_queued_contexts_for_all_hosts += dictionary_entries(host->rrdctx.pp_queue);
1214 - rrdcontext_post_process_queued_contexts(host);
1215 - dictionary_garbage_collect(host->rrdctx.pp_queue);
1216 - }
1083 + pp_queued_contexts_for_all_hosts += rrdcontext_queue_entries(&host->rrdctx.pp_queue);
1084 + rrdcontext_post_process_queued_contexts(host);
1085
1218 - if(host->rrdctx.hub_queue) {
1219 - hub_queued_contexts_for_all_hosts += dictionary_entries(host->rrdctx.hub_queue);
1220 - rrdcontext_dispatch_queued_contexts_to_hub(host, now_ut);
1221 - dictionary_garbage_collect(host->rrdctx.hub_queue);
1222 - }
1086 + hub_queued_contexts_for_all_hosts += rrdcontext_queue_entries(&host->rrdctx.hub_queue);
1087 + rrdcontext_dispatch_queued_contexts_to_hub(host, now_ut);
1088
1089 if (host->rrdctx.contexts)
1090 dictionary_garbage_collect(host->rrdctx.contexts);
src/database/contexts/rrdcontext.c
+1 -1
@@ -1,6 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -#include "internal.h"
3 +#include "rrdcontext-internal.h"
4
5 // ----------------------------------------------------------------------------
6 // visualizing flags
src/database/contexts/rrdcontext.h
+3
@@ -757,5 +757,8 @@ static inline bool query_target_has_percentage_units(QUERY_TARGET *qt) {
757 return query_target_has_percentage_of_group(qt);
758 }
759
760 +uint32_t rrdcontext_queue_version(RRDCONTEXT_QUEUE_JudyLSet *queue);
761 +int32_t rrdcontext_queue_entries(RRDCONTEXT_QUEUE_JudyLSet *queue);
762 +
763 #endif // NETDATA_RRDCONTEXT_H
764
src/database/rrddim.c
+36 -19
@@ -51,6 +51,8 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
51 RRDSET *st = ctr->st;
52 RRDHOST *host = st->rrdhost;
53
54 + spinlock_init(&rd->destroy_lock);
55 +
56 rd->flags = RRDDIM_FLAG_NONE;
57
58 rd->id = string_strdupz(ctr->id);
@@ -255,6 +257,9 @@ static void rrddim_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, v
257 string_freez(rd->name);
258 uuidmap_free(rd->uuid);
259
260 + if(rd->destroy_lock.locked)
261 + spinlock_unlock(&rd->destroy_lock);
262 +
263 memset(rd, 0, sizeof(RRDDIM));
264 }
265
@@ -267,8 +272,6 @@ static bool rrddim_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused,
272
273 ctr->react_action = RRDDIM_REACT_NONE;
274
270 - rrddim_flag_clear(rd, RRDDIM_FLAG_ARCHIVED | RRDDIM_FLAG_OBSOLETE);
271 -
275 int rc = rrddim_reset_name(st, rd, ctr->name);
276 rc += rrddim_set_algorithm(st, rd, ctr->algorithm);
277 rc += rrddim_set_multiplier(st, rd, ctr->multiplier);
@@ -359,8 +362,8 @@ void rrddim_acquired_release(RRDDIM_ACQUIRED *rda) {
362 RRDDIM *rrddim_find_active(RRDSET *st, const char *id) {
363 RRDDIM *rd = rrddim_find(st, id);
364
362 - if (unlikely(rd && rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)))
363 - return NULL;
365 + if(rd && rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))
366 + rd = NULL;
367
368 return rd;
369 }
@@ -485,17 +488,35 @@ RRDDIM *rrddim_add_custom(RRDSET *st
488 ,
489 RRD_DB_MODE memory_mode
490 ) {
488 - struct rrddim_constructor tmp = {
489 - .st = st,
490 - .id = id,
491 - .name = name,
492 - .multiplier = multiplier,
493 - .divisor = divisor,
494 - .algorithm = algorithm,
495 - .memory_mode = memory_mode,
496 - };
491
498 - RRDDIM *rd = dictionary_set_advanced(st->rrddim_root_index, tmp.id, -1, NULL, rrddim_size(), &tmp);
492 + RRDDIM *rd = NULL;
493 + while(!rd) {
494 + rd = rrddim_index_find(st, id);
495 + if(rd) {
496 + if(spinlock_trylock(&rd->destroy_lock)) {
497 + rrddim_isnot_obsolete___safe_from_collector_thread(st, rd);
498 + spinlock_unlock(&rd->destroy_lock);
499 + }
500 + else {
501 + rd = NULL;
502 + microsleep(1 * USEC_PER_MS);
503 + continue;
504 + }
505 + }
506 +
507 + struct rrddim_constructor tmp = {
508 + .st = st,
509 + .id = id,
510 + .name = name,
511 + .multiplier = multiplier,
512 + .divisor = divisor,
513 + .algorithm = algorithm,
514 + .memory_mode = memory_mode,
515 + };
516 +
517 + rd = dictionary_set_advanced(st->rrddim_root_index, tmp.id, -1, NULL, rrddim_size(), &tmp);
518 + }
519 +
520 return(rd);
521 }
522
@@ -553,10 +574,6 @@ int rrddim_unhide(RRDSET *st, const char *id) {
574 inline void rrddim_is_obsolete___safe_from_collector_thread(RRDSET *st, RRDDIM *rd) {
575 netdata_log_debug(D_RRD_CALLS, "rrddim_is_obsolete___safe_from_collector_thread() for chart %s, dimension %s", rrdset_name(st), rrddim_name(rd));
576
556 - if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED))) {
557 - netdata_log_info("Cannot obsolete already archived dimension %s from chart %s", rrddim_name(rd), rrdset_name(st));
558 - return;
559 - }
577 rrddim_flag_set(rd, RRDDIM_FLAG_OBSOLETE);
578 rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS);
579 rrdhost_flag_set(st->rrdhost, RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);
@@ -566,7 +583,7 @@ inline void rrddim_is_obsolete___safe_from_collector_thread(RRDSET *st, RRDDIM *
583 inline void rrddim_isnot_obsolete___safe_from_collector_thread(RRDSET *st __maybe_unused, RRDDIM *rd) {
584 netdata_log_debug(D_RRD_CALLS, "rrddim_isnot_obsolete___safe_from_collector_thread() for chart %s, dimension %s", rrdset_name(st), rrddim_name(rd));
585
569 - rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE|RRDDIM_FLAG_ARCHIVED);
586 + rrddim_flag_clear(rd, RRDDIM_FLAG_OBSOLETE);
587 rrddim_reinitialize_collection(rd);
588 rrdcontext_updated_rrddim_flags(rd);
589 }
src/database/rrddim.h
+5 -5
@@ -36,11 +36,10 @@ typedef enum __attribute__ ((__packed__)) rrddim_flags {
36 // No new values have been collected for this dimension since agent start, or it was marked RRDDIM_FLAG_OBSOLETE at
37 // least rrdset_free_obsolete_time seconds ago.
38
39 - RRDDIM_FLAG_ARCHIVED = (1 << 1),
40 - RRDDIM_FLAG_METADATA_UPDATE = (1 << 2), // Metadata needs to go to the database
39 + RRDDIM_FLAG_METADATA_UPDATE = (1 << 1), // Metadata needs to go to the database
40
42 - RRDDIM_FLAG_META_HIDDEN = (1 << 3), // Status of hidden option in the metadata database
43 - RRDDIM_FLAG_ML_MODEL_LOAD = (1 << 4), // Do ML LOAD for this dimension
41 + RRDDIM_FLAG_META_HIDDEN = (1 << 2), // Status of hidden option in the metadata database
42 + RRDDIM_FLAG_ML_MODEL_LOAD = (1 << 3), // Do ML LOAD for this dimension
43
44 // this is 8 bit
45 } RRDDIM_FLAGS;
@@ -61,9 +60,10 @@ struct rrddim {
60 STRING *name; // the name of this dimension (as presented to user)
61
62 RRD_ALGORITHM algorithm; // the algorithm that is applied to add new collected values
64 - RRD_DB_MODE rrd_memory_mode; // the memory mode for this dimension
63 + RRD_DB_MODE rrd_memory_mode; // the memory mode for this dimension
64 RRDDIM_FLAGS flags; // run time changing status flags
65
66 + SPINLOCK destroy_lock;
67 int32_t multiplier; // the multiplier of the collected values
68 int32_t divisor; // the divider of the collected values
69
src/database/rrdhost.h
+10 -5
@@ -25,6 +25,14 @@ typedef struct rrdhost_acquired RRDHOST_ACQUIRED;
25 #include "rrdlabels.h"
26 #include "health/health-alert-log.h"
27
28 +struct rrdcontext;
29 +DEFINE_JUDYL_TYPED_ADVANCED(RRDCONTEXT_QUEUE, struct rrdcontext *, JUDYL_TYPED_NO_CONVERSION, JUDYL_TYPED_NO_CONVERSION, \
30 + SPINLOCK spinlock; \
31 + Word_t id; \
32 + uint32_t version; \
33 + int32_t entries; \
34 + );
35 +
36 // ----------------------------------------------------------------------------
37 // RRDHOST flags
38 // use this for configuration flags, not for state control
@@ -217,9 +225,6 @@ struct rrdhost {
225
226 time_t last_connected; // the time the last sender was connected
227 time_t last_disconnected; // the time the last sender was disconnected
220 - time_t last_chart; // the time of the last CHART streaming command
221 - bool check_obsolete; // set when child connects, will instruct parent to
222 - // trigger a check for obsoleted charts since previous connect
228
229 uint32_t connections; // the number of times this receiver has connected
230 STREAM_HANDSHAKE reason; // the last receiver exit reason
@@ -304,8 +309,8 @@ struct rrdhost {
309
310 struct {
311 DICTIONARY *contexts;
307 - DICTIONARY *hub_queue;
308 - DICTIONARY *pp_queue;
312 + RRDCONTEXT_QUEUE_JudyLSet pp_queue;
313 + RRDCONTEXT_QUEUE_JudyLSet hub_queue;
314 uint32_t metrics_count; // atomic
315 uint32_t instances_count; // atomic
316 uint32_t contexts_count; // atomic
src/database/rrdset-collection.c
+11 -12
@@ -81,10 +81,8 @@ static void rrdset_collection_reset(RRDSET *st) {
81 rd->collector.last_collected_time.tv_usec = 0;
82 rd->collector.counter = 0;
83
84 - if(!rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
85 - for(size_t tier = 0; tier < nd_profile.storage_tiers;tier++)
86 - storage_engine_store_flush(rd->tiers[tier].sch);
87 - }
84 + for(size_t tier = 0; tier < nd_profile.storage_tiers;tier++)
85 + storage_engine_store_flush(rd->tiers[tier].sch);
86 }
87 rrddim_foreach_done(rd);
88 }
@@ -542,7 +540,11 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
540
541 if (unlikely(rrdset_flags & RRDSET_FLAG_OBSOLETE)) {
542 netdata_log_error("Chart '%s' has the OBSOLETE flag set, but it is collected.", rrdset_id(st));
543 + if(!spinlock_trylock(&st->destroy_lock))
544 + fatal("RRDSET: chart '%s' of host '%s' is being collected while is being destroyed.", rrdset_id(st), rrdhost_hostname(st->rrdhost));
545 +
546 rrdset_isnot_obsolete___safe_from_collector_thread(st);
547 + spinlock_unlock(&st->destroy_lock);
548 }
549
550 // check if the chart has a long time to be updated
@@ -653,13 +655,6 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
655
656 rda = &rda_base[dimensions++];
657
656 - if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
657 - rda->item = NULL;
658 - rda->rd = NULL;
659 - rda->reset_or_overflow = false;
660 - continue;
661 - }
662 -
658 // store the dimension in the array
659 rda->item = dictionary_acquired_item_dup(st->rrddim_root_index, rd_dfe.item);
660 rda->rd = dictionary_acquired_item_value(rda->item);
@@ -686,9 +681,13 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
681 last_collected_total += rd->collector.last_collected_value;
682 collected_total += rd->collector.collected_value;
683
689 - if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE | RRDDIM_FLAG_ARCHIVED))) {
684 + if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))) {
685 netdata_log_error("Dimension %s in chart '%s' has the OBSOLETE or ARCHIVED flag set, but it is collected.", rrddim_name(rd), rrdset_id(st));
686 + if(!spinlock_trylock(&rd->destroy_lock))
687 + fatal("RRDSET: dimension '%s' of chart '%s' of host '%s' is being collected while is being destroyed.", rrddim_id(rd), rrdset_id(st), rrdhost_hostname(st->rrdhost));
688 +
689 rrddim_isnot_obsolete___safe_from_collector_thread(st, rd);
690 + spinlock_unlock(&rd->destroy_lock);
691 }
692 }
693 }
src/database/rrdset-index-id.c
+47 -24
@@ -63,6 +63,8 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
63 RRDHOST *host = ctr->host;
64 RRDSET *st = rrdset;
65
66 + spinlock_init(&st->destroy_lock);
67 +
68 const char *chart_full_id = dictionary_acquired_item_name(item);
69
70 st->id = string_strdupz(chart_full_id);
@@ -180,6 +182,9 @@ static void rrdset_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, v
182
183 freez(st->exporting_flags);
184
185 + if(st->destroy_lock.locked)
186 + spinlock_unlock(&st->destroy_lock);
187 +
188 memset(st, 0, sizeof(RRDSET));
189 }
190
@@ -192,8 +197,6 @@ static bool rrdset_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused,
197 struct rrdset_constructor *ctr = constructor_data;
198 RRDSET *st = rrdset;
199
195 - rrdset_isnot_obsolete___safe_from_collector_thread(st);
196 -
200 ctr->react_action = RRDSET_REACT_NONE;
201
202 if (unlikely(st->priority != ctr->priority)) {
@@ -275,6 +278,7 @@ static void rrdset_react_callback(const DICTIONARY_ITEM *item __maybe_unused, vo
278 RRDSET *st = rrdset;
279 RRDHOST *host = st->rrdhost;
280
281 + st->collector_tid = gettid_cached();
282 st->last_accessed_time_s = now_realtime_sec();
283
284 if(ctr->react_action & (RRDSET_REACT_NEW | RRDSET_REACT_PLUGIN_UPDATED | RRDSET_REACT_MODULE_UPDATED)) {
@@ -390,9 +394,6 @@ RRDSET *rrdset_create_custom(
394 RRD_DB_MODE memory_mode
395 , long history_entries
396 ) {
393 - if (host != localhost)
394 - host->stream.rcv.status.last_chart = now_realtime_sec();
395 -
397 if(!type || !type[0])
398 fatal("Cannot create rrd stats without a type: id '%s', name '%s', family '%s', context '%s', title '%s', units '%s', plugin '%s', module '%s'."
399 , (id && *id)?id:"<unset>"
@@ -428,25 +429,47 @@ RRDSET *rrdset_create_custom(
429
430 netdata_log_debug(D_RRD_CALLS, "Creating RRD_STATS for '%s.%s'.", type, id);
431
431 - struct rrdset_constructor ctr = {
432 - .host = host,
433 - .type = type,
434 - .id = id,
435 - .name = name,
436 - .family = family,
437 - .context = context,
438 - .title = title,
439 - .units = units,
440 - .plugin = plugin,
441 - .module = module,
442 - .priority = priority,
443 - .update_every = update_every,
444 - .chart_type = chart_type,
445 - .memory_mode = memory_mode,
446 - .history_entries = history_entries,
447 - };
448 -
449 - RRDSET *st = rrdset_index_add(host, chart_full_id, &ctr);
432 + struct rrdset_constructor ctr;
433 +
434 + RRDSET *st = NULL;
435 + while(!st) {
436 + st = rrdset_index_find(host, chart_full_id);
437 + if(st) {
438 + if(spinlock_trylock(&st->destroy_lock)) {
439 + rrdset_isnot_obsolete___safe_from_collector_thread(st);
440 + spinlock_unlock(&st->destroy_lock);
441 + }
442 + else {
443 +#ifdef FSANITIZE_ADDRESS
444 + fprintf(stderr, "rrdset_create_custom() - chart '%s' of host '%s' is being deleted but we need it. Retrying...\n",
445 + chart_full_id, rrdhost_hostname(host));
446 +#endif
447 + st = NULL;
448 + microsleep(1 * USEC_PER_MS);
449 + continue;
450 + }
451 + }
452 +
453 + ctr = (struct rrdset_constructor){
454 + .host = host,
455 + .type = type,
456 + .id = id,
457 + .name = name,
458 + .family = family,
459 + .context = context,
460 + .title = title,
461 + .units = units,
462 + .plugin = plugin,
463 + .module = module,
464 + .priority = priority,
465 + .update_every = update_every,
466 + .chart_type = chart_type,
467 + .memory_mode = memory_mode,
468 + .history_entries = history_entries,
469 + };
470 +
471 + st = rrdset_index_add(host, chart_full_id, &ctr);
472 + }
473
474 bool name_updated = false;
475 if(!st->name) {
src/database/rrdset.h
+5 -1
@@ -111,9 +111,11 @@ struct rrdset {
111 // ------------------------------------------------------------------------
112 // operational state members
113
114 - RRDSET_FLAGS flags; // flags
114 RRD_DB_MODE rrd_memory_mode; // the db mode of this rrdset
115 uint16_t collection_modulo; // tier1/2 spread over time
116 + RRDSET_FLAGS flags; // flags
117 +
118 + pid_t collector_tid;
119
120 DICTIONARY *rrddim_root_index; // dimensions index
121
@@ -223,6 +225,8 @@ struct rrdset {
225 time_t before;
226 } replay;
227 #endif // NETDATA_LOG_REPLICATION_REQUESTS
228 +
229 + SPINLOCK destroy_lock;
230 };
231
232 // --------------------------------------------------------------------------------------------------------------------
src/database/sqlite/sqlite_aclk_node.c
+1 -1
@@ -154,7 +154,7 @@ void aclk_check_node_info_and_collectors(void)
154 if(replicating_rcv)
155 continue;
156
157 - bool pp_queue_empty = !(host->rrdctx.pp_queue && dictionary_entries(host->rrdctx.pp_queue));
157 + bool pp_queue_empty = !rrdcontext_queue_entries(&host->rrdctx.pp_queue);
158
159 if (!pp_queue_empty && (aclk_host_config->node_info_send_time || aclk_host_config->node_collectors_send)) {
160 context_pp++;
src/database/sqlite/sqlite_context.c
+1 -1
@@ -3,7 +3,7 @@
3 #include "sqlite_functions.h"
4 #include "sqlite_context.h"
5 #include "sqlite_db_migration.h"
6 -#include "database/contexts/internal.h"
6 +#include "database/contexts/rrdcontext-internal.h"
7
8 #define DB_CONTEXT_METADATA_VERSION 1
9
src/health/health_event_loop.c
-10
@@ -271,16 +271,6 @@ static void health_event_loop_for_host(RRDHOST *host, bool apply_hibernation_del
271 host->health.delay_up_to = 0;
272 }
273
274 - // wait until cleanup of obsolete charts on children is complete
275 - if (host != localhost) {
276 - if (unlikely(host->stream.rcv.status.check_obsolete)) {
277 - nd_log(NDLS_DAEMON, NDLP_DEBUG,
278 - "[%s]: Waiting for chart obsoletion check.",
279 - rrdhost_hostname(host));
280 - return;
281 - }
282 - }
283 -
274 worker_is_busy(WORKER_HEALTH_JOB_HOST_LOCK);
275 {
276 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_RELAXED);
src/libnetdata/dictionary/dictionary-debug.c
+6 -6
@@ -93,8 +93,8 @@ static size_t report_allocated_dictionaries(void) {
93 Pvoid_t PDictList;
94 PDictList = JudyLGet(stacktrace_dictionaries, st_key, PJE0);
95 if (!PDictList) {
96 - // Create a new array (starting with size 16)
97 - DICTIONARY **dict_list = (DICTIONARY **)calloc(16, sizeof(DICTIONARY*));
96 + // Create a new array (starting with size 1024)
97 + DICTIONARY **dict_list = (DICTIONARY **)callocz(1024, sizeof(DICTIONARY*));
98 if (dict_list) {
99 dict_list[0] = dict;
100 JudyLIns(&stacktrace_dictionaries, st_key, PJE0);
@@ -110,7 +110,7 @@ static size_t report_allocated_dictionaries(void) {
110 size_t i = 0;
111
112 // Check if dictionary is already in this list
113 - while (dict_list[i] && i < 1024) {
113 + while (i < 1024 && dict_list[i]) {
114 if (dict_list[i] == dict) {
115 already_added = true;
116 break;
@@ -147,7 +147,7 @@ static size_t report_allocated_dictionaries(void) {
147 }
148
149 if (num_stacktraces > 0) {
150 - StacktraceInfo *stacktraces = (StacktraceInfo *)calloc(num_stacktraces, sizeof(StacktraceInfo));
150 + StacktraceInfo *stacktraces = (StacktraceInfo *)callocz(num_stacktraces, sizeof(StacktraceInfo));
151 if (stacktraces) {
152 // Fill array
153 Word_t i = 0;
@@ -199,14 +199,14 @@ static size_t report_allocated_dictionaries(void) {
199 }
200 }
201
202 - free(stacktraces);
202 + freez(stacktraces);
203 }
204
205 // Clean up
206 index = 0;
207 PValue = JudyLFirst(stacktrace_dictionaries, &index, PJE0);
208 while (PValue != NULL) {
209 - free(*(void**)PValue);
209 + freez(*(void**)PValue);
210 PValue = JudyLNext(stacktrace_dictionaries, &index, PJE0);
211 }
212
src/libnetdata/dictionary/dictionary-internals.h
+2
@@ -247,6 +247,8 @@ static inline void pointer_del(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM
247 extern ARAL *dict_items_aral;
248 extern ARAL *dict_shared_items_aral;
249
250 +size_t dictionary_destroy_delayed_count(void);
251 +
252 #include "dictionary-debug.h"
253 #include "dictionary-statistics.h"
254 #include "dictionary-locks.h"
src/libnetdata/dictionary/dictionary-traversal.c
+10 -18
@@ -6,31 +6,30 @@
6 // ----------------------------------------------------------------------------
7 // traversal with loop
8
9 -void *dictionary_foreach_start_rw(DICTFE *dfe, DICTIONARY *dict, char rw) {
10 - if(unlikely(!dfe || !dict)) return NULL;
9 +void *dictionary_foreach_start_rw(DICTFE *dfe) {
10 + if(unlikely(!dfe || !dfe->dict)) return NULL;
11
12 - DICTIONARY_STATS_TRAVERSALS_PLUS1(dict);
12 + DICTIONARY_STATS_TRAVERSALS_PLUS1(dfe->dict);
13
14 - if(unlikely(is_dictionary_destroyed(dict))) {
14 + if(unlikely(is_dictionary_destroyed(dfe->dict))) {
15 internal_error(true, "DICTIONARY: attempted to dictionary_foreach_start_rw() on a destroyed dictionary");
16 - dfe->counter = 0;
16 + dfe->dict = NULL;
17 dfe->item = NULL;
18 dfe->name = NULL;
19 dfe->value = NULL;
20 + dfe->counter = 0;
21 return NULL;
22 }
23
24 dfe->counter = 0;
24 - dfe->dict = dict;
25 - dfe->rw = rw;
25 dfe->locked = true;
27 - ll_recursive_lock(dict, dfe->rw);
26 + ll_recursive_lock(dfe->dict, dfe->rw);
27
28 // get the first item from the list
30 - DICTIONARY_ITEM *item = dict->items.list;
29 + DICTIONARY_ITEM *item = dfe->dict->items.list;
30
31 // skip all the deleted items
33 - while(item && !item_check_and_acquire(dict, item))
32 + while(item && !item_check_and_acquire(dfe->dict, item))
33 item = item->next;
34
35 if(likely(item)) {
@@ -57,9 +56,7 @@ ALWAYS_INLINE void *dictionary_foreach_next(DICTFE *dfe) {
56
57 if(unlikely(is_dictionary_destroyed(dfe->dict))) {
58 internal_error(true, "DICTIONARY: attempted to dictionary_foreach_next() on a destroyed dictionary");
60 - dfe->item = NULL;
61 - dfe->name = NULL;
62 - dfe->value = NULL;
59 + dictionary_foreach_done(dfe);
60 return NULL;
61 }
62
@@ -114,11 +111,6 @@ void dictionary_foreach_unlock(DICTFE *dfe) {
111 void dictionary_foreach_done(DICTFE *dfe) {
112 if(unlikely(!dfe || !dfe->dict)) return;
113
117 - if(unlikely(is_dictionary_destroyed(dfe->dict))) {
118 - internal_error(true, "DICTIONARY: attempted to dictionary_foreach_next() on a destroyed dictionary");
119 - return;
120 - }
121 -
114 // the item we just did
115 DICTIONARY_ITEM *item = dfe->item;
116
src/libnetdata/dictionary/dictionary-unittest.c
+33
@@ -1027,6 +1027,25 @@ size_t dictionary_unittest_views(void) {
1027 return errors;
1028 }
1029
1030 +bool dictionary_traverse_or_destroy_unittest(void) {
1031 + DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);
1032 + dictionary_set(dict, "KEY 1", "VALUE1", strlen("VALUE1") + 1);
1033 + dictionary_set(dict, "KEY 2", "VALUE2", strlen("VALUE2") + 1);
1034 + dictionary_set(dict, "KEY 3", "VALUE3", strlen("VALUE3") + 1);
1035 +
1036 + size_t counted = 0;
1037 + const char *s;
1038 + dfe_start_read(dict, s) {
1039 + if(!counted)
1040 + dictionary_destroy(dict);
1041 +
1042 + counted++;
1043 + }
1044 + dfe_done(s);
1045 +
1046 + return counted == 1;
1047 +}
1048 +
1049 /*
1050 * FIXME: a dictionary-related leak is reported when running the address
1051 * sanitizer. Need to investigate if it's introduced by the unit-test itself,
@@ -1183,8 +1202,22 @@ int dictionary_unittest(size_t entries) {
1202 errors += dictionary_unittest_threads();
1203 errors += dictionary_unittest_view_threads();
1204
1205 + if(!dictionary_traverse_or_destroy_unittest()) {
1206 + fprintf(stderr, "Destroy on traversal test failed\n");
1207 + errors++;
1208 + }
1209 + else
1210 + fprintf(stderr, "Destroy on traversal test OK\n");
1211 +
1212 cleanup_destroyed_dictionaries(false);
1213
1214 + size_t delayed = dictionary_destroy_delayed_count();
1215 + if(delayed != 0) {
1216 + fprintf(stderr, "WARNING: There are %zu dictionaries that cannot be destroyed\n", delayed);
1217 + }
1218 + else
1219 + fprintf(stderr, "All dictionaries have been freed: OK\n");
1220 +
1221 fprintf(stderr, "\n%zu errors found\n", errors);
1222 return errors ? 1 : 0;
1223 }
src/libnetdata/dictionary/dictionary.c
+13
@@ -333,6 +333,19 @@ static void dictionary_queue_for_destruction(DICTIONARY *dict) {
333 netdata_mutex_unlock(&dictionaries_waiting_to_be_destroyed_mutex);
334 }
335
336 +size_t dictionary_destroy_delayed_count(void) {
337 + netdata_mutex_lock(&dictionaries_waiting_to_be_destroyed_mutex);
338 +
339 + size_t count = 0;
340 + for(DICTIONARY *dict = dictionaries_waiting_to_be_destroyed, *next; dict ; dict = next) {
341 + next = dict->next;
342 + count++;
343 + }
344 +
345 + netdata_mutex_unlock(&dictionaries_waiting_to_be_destroyed_mutex);
346 + return count;
347 +}
348 +
349 size_t cleanup_destroyed_dictionaries(bool shutdown __maybe_unused) {
350 netdata_mutex_lock(&dictionaries_waiting_to_be_destroyed_mutex);
351 if (!dictionaries_waiting_to_be_destroyed) {
src/libnetdata/dictionary/dictionary.h
+15 -7
@@ -289,14 +289,22 @@ typedef DICTFE_CONST struct dictionary_foreach {
289 #define dfe_start_write(dict, value) dfe_start_rw(dict, value, DICTIONARY_LOCK_WRITE)
290 #define dfe_start_reentrant(dict, value) dfe_start_rw(dict, value, DICTIONARY_LOCK_REENTRANT)
291
292 -#define dfe_start_rw(dict, value, mode) \
292 +#define dfe_start_rw(dictionary, ptr, mode) \
293 do { \
294 /* automatically cleanup DFE, to allow using return from within the loop */ \
295 - DICTFE _cleanup_(dictionary_foreach_done) value ## _dfe = {}; \
296 - (void)(value); /* needed to avoid warning when looping without using this */ \
297 - for((value) = dictionary_foreach_start_rw(&value ## _dfe, (dict), (mode)); \
298 - (value ## _dfe.item) || (value) ; \
299 - (value) = dictionary_foreach_next(&value ## _dfe)) \
295 + DICTFE _cleanup_(dictionary_foreach_done) ptr ## _dfe = (DICTFE){ \
296 + .dict = (dictionary), \
297 + .item = NULL, \
298 + .name = NULL, \
299 + .value = NULL, \
300 + .counter = 0, \
301 + .rw = (mode), \
302 + .locked = false, \
303 + }; \
304 + (void)(ptr); /* needed to avoid warning when looping without using this */ \
305 + for((ptr) = dictionary_foreach_start_rw(&ptr ## _dfe); \
306 + (ptr ## _dfe.item) || (ptr) ; \
307 + (ptr) = dictionary_foreach_next(&ptr ## _dfe)) \
308 {
309
310 #define dfe_done(value) \
@@ -305,7 +313,7 @@ typedef DICTFE_CONST struct dictionary_foreach {
313
314 #define dfe_unlock(value) dictionary_foreach_unlock(&value ## _dfe)
315
308 -void *dictionary_foreach_start_rw(DICTFE *dfe, DICTIONARY *dict, char rw);
316 +void *dictionary_foreach_start_rw(DICTFE *dfe);
317 void *dictionary_foreach_next(DICTFE *dfe);
318 void dictionary_foreach_done(DICTFE *dfe);
319 void dictionary_foreach_unlock(DICTFE *dfe);
src/libnetdata/libjudy/judyl-typed.h
+75 -68
@@ -5,80 +5,87 @@
5
6 #include <Judy.h>
7
8 +#ifdef __cplusplus
9 +#define DEFINED_JUDYL_CHECK_SIZE(TYPE, NAME)
10 +#else
11 +#define DEFINED_JUDYL_CHECK_SIZE(TYPE, NAME) _Static_assert(sizeof(TYPE) <= sizeof(Word_t), #NAME "_type_must_have_same_size_as_Word_t")
12 +#endif
13 +
14 +
15 // Advanced macro for types requiring conversion
9 -#define DEFINE_JUDYL_TYPED_ADVANCED(NAME, TYPE, PACK_MACRO, UNPACK_MACRO) \
10 - _Static_assert(sizeof(TYPE) <= sizeof(Word_t), \
11 - #NAME "_type_must_have_same_size_as_Word_t"); \
12 - typedef struct { \
13 - Pvoid_t judyl; \
14 - } NAME##_JudyLSet; \
15 - \
16 - ALWAYS_INLINE \
17 - static __attribute__((unused)) void NAME##_INIT(NAME##_JudyLSet *set) { \
18 - set->judyl = NULL; \
19 - } \
20 - \
21 - ALWAYS_INLINE \
22 - static bool __attribute__((unused)) NAME##_SET(NAME##_JudyLSet *set, Word_t index, TYPE value) { \
23 - Pvoid_t *pValue = JudyLIns(&set->judyl, index, PJE0); \
24 - if (pValue == PJERR) return false; \
25 - *pValue = (void *)PACK_MACRO(value); \
26 - return true; \
27 - } \
28 - \
29 - ALWAYS_INLINE \
30 - static TYPE __attribute__((unused)) NAME##_GET(NAME##_JudyLSet *set, Word_t index) { \
31 - Pvoid_t *pValue = JudyLGet(set->judyl, index, PJE0); \
32 - return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
33 - } \
34 - \
35 - ALWAYS_INLINE \
36 - static bool __attribute__((unused)) NAME##_DEL(NAME##_JudyLSet *set, Word_t index) { \
37 - return JudyLDel(&set->judyl, index, PJE0) == 1; \
38 - } \
39 - \
40 - ALWAYS_INLINE \
41 - static TYPE __attribute__((unused)) NAME##_FIRST(NAME##_JudyLSet *set, Word_t *index) { \
42 - Pvoid_t *pValue = JudyLFirst(set->judyl, index, PJE0); \
43 - return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
44 - } \
45 - \
46 - ALWAYS_INLINE \
47 - static TYPE __attribute__((unused)) NAME##_NEXT(NAME##_JudyLSet *set, Word_t *index) { \
48 - Pvoid_t *pValue = JudyLNext(set->judyl, index, PJE0); \
49 - return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
50 - } \
51 - \
52 - ALWAYS_INLINE \
53 - static TYPE __attribute__((unused)) NAME##_LAST(NAME##_JudyLSet *set, Word_t *index) { \
54 - Pvoid_t *pValue = JudyLLast(set->judyl, index, PJE0); \
55 - return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
56 - } \
57 - \
58 - ALWAYS_INLINE \
59 - static TYPE __attribute__((unused)) NAME##_PREV(NAME##_JudyLSet *set, Word_t *index) { \
60 - Pvoid_t *pValue = JudyLPrev(set->judyl, index, PJE0); \
61 - return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
62 - } \
63 - \
64 - ALWAYS_INLINE \
16 +#define DEFINE_JUDYL_TYPED_ADVANCED(NAME, TYPE, PACK_MACRO, UNPACK_MACRO, EXTRA_MEMBERS) \
17 + DEFINED_JUDYL_CHECK_SIZE(TYPE, NAME); \
18 + \
19 + typedef struct { \
20 + Pvoid_t judyl; \
21 + EXTRA_MEMBERS \
22 + } NAME##_JudyLSet; \
23 + ALWAYS_INLINE \
24 + static __attribute__((unused)) void NAME##_INIT(NAME##_JudyLSet *set) { \
25 + set->judyl = NULL; \
26 + } \
27 + \
28 + ALWAYS_INLINE \
29 + static bool __attribute__((unused)) NAME##_SET(NAME##_JudyLSet *set, Word_t index, TYPE value) { \
30 + Pvoid_t *pValue = JudyLIns(&set->judyl, index, PJE0); \
31 + if (pValue == PJERR) return false; \
32 + *pValue = (void *)PACK_MACRO(value); \
33 + return true; \
34 + } \
35 + \
36 + ALWAYS_INLINE \
37 + static TYPE __attribute__((unused)) NAME##_GET(NAME##_JudyLSet *set, Word_t index) { \
38 + Pvoid_t *pValue = JudyLGet(set->judyl, index, PJE0); \
39 + return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
40 + } \
41 + \
42 + ALWAYS_INLINE \
43 + static bool __attribute__((unused)) NAME##_DEL(NAME##_JudyLSet *set, Word_t index) { \
44 + return JudyLDel(&set->judyl, index, PJE0) == 1; \
45 + } \
46 + \
47 + ALWAYS_INLINE \
48 + static TYPE __attribute__((unused)) NAME##_FIRST(NAME##_JudyLSet *set, Word_t *index) { \
49 + Pvoid_t *pValue = JudyLFirst(set->judyl, index, PJE0); \
50 + return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
51 + } \
52 + \
53 + ALWAYS_INLINE \
54 + static TYPE __attribute__((unused)) NAME##_NEXT(NAME##_JudyLSet *set, Word_t *index) { \
55 + Pvoid_t *pValue = JudyLNext(set->judyl, index, PJE0); \
56 + return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
57 + } \
58 + \
59 + ALWAYS_INLINE \
60 + static TYPE __attribute__((unused)) NAME##_LAST(NAME##_JudyLSet *set, Word_t *index) { \
61 + Pvoid_t *pValue = JudyLLast(set->judyl, index, PJE0); \
62 + return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
63 + } \
64 + \
65 + ALWAYS_INLINE \
66 + static TYPE __attribute__((unused)) NAME##_PREV(NAME##_JudyLSet *set, Word_t *index) { \
67 + Pvoid_t *pValue = JudyLPrev(set->judyl, index, PJE0); \
68 + return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
69 + } \
70 + \
71 + ALWAYS_INLINE \
72 static void __attribute__((unused)) NAME##_FREE(NAME##_JudyLSet *set, void (*callback)(Word_t, TYPE, void *), void *data) { \
66 - Word_t index = 0; \
67 - Pvoid_t *pValue; \
68 - if (callback) { \
69 - for (pValue = JudyLFirst(set->judyl, &index, PJE0); \
70 - pValue != NULL; \
71 - pValue = JudyLNext(set->judyl, &index, PJE0)) { \
72 - callback(index, (TYPE)UNPACK_MACRO(*pValue), data); \
73 - } \
74 - } \
75 - JudyLFreeArray(&set->judyl, PJE0); \
73 + Word_t index = 0; \
74 + Pvoid_t *pValue; \
75 + if (callback) { \
76 + for (pValue = JudyLFirst(set->judyl, &index, PJE0); \
77 + pValue != NULL; \
78 + pValue = JudyLNext(set->judyl, &index, PJE0)) { \
79 + callback(index, (TYPE)UNPACK_MACRO(*pValue), data); \
80 + } \
81 + } \
82 + JudyLFreeArray(&set->judyl, PJE0); \
83 }
84
85 // Basic macro for types with no conversion
86 #define JUDYL_TYPED_NO_CONVERSION(value) (uintptr_t)(value)
87
81 -#define DEFINE_JUDYL_TYPED(NAME, TYPE) \
82 - DEFINE_JUDYL_TYPED_ADVANCED(NAME, TYPE, JUDYL_TYPED_NO_CONVERSION, JUDYL_TYPED_NO_CONVERSION)
88 +#define DEFINE_JUDYL_TYPED(NAME, TYPE) \
89 + DEFINE_JUDYL_TYPED_ADVANCED(NAME, TYPE, JUDYL_TYPED_NO_CONVERSION, JUDYL_TYPED_NO_CONVERSION,)
90
91 #endif //NETDATA_JUDYL_TYPED_H
src/libnetdata/stacktrace/stacktrace-array.h
+1 -1
@@ -8,7 +8,7 @@
8
9 // Default maximum number of stacktraces to track per array
10 #ifndef STACKTRACE_ARRAY_MAX_TRACES
11 -#define STACKTRACE_ARRAY_MAX_TRACES 20
11 +#define STACKTRACE_ARRAY_MAX_TRACES 100
12 #endif
13
14 // Structure to track multiple stacktraces
src/ml/ml.cc
+1
@@ -544,6 +544,7 @@ ml_dimension_deserialize_kmeans(const char *json_str)
544 ml_dimension_t *Dim = reinterpret_cast<ml_dimension_t *>(AcqDim.dimension());
545 if (!Dim) {
546 pulse_ml_models_ignored();
547 + json_object_put(root);
548 return true;
549 }
550
src/plugins.d/pluginsd_parser.c
+21 -5
@@ -382,10 +382,8 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p
382 else
383 rrdset_flag_clear(st, RRDSET_FLAG_STORE_FIRST);
384 }
385 - else {
386 - rrdset_isnot_obsolete___safe_from_collector_thread(st);
385 + else
386 rrdset_flag_clear(st, RRDSET_FLAG_STORE_FIRST);
388 - }
387
388 if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_CHART))
389 return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
@@ -735,8 +733,13 @@ static ALWAYS_INLINE PARSER_RC pluginsd_begin_v2(char **words, size_t num_words,
733 if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_BEGIN_V2))
734 return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
735
738 - if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)))
736 + if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE))) {
737 + if(!spinlock_trylock(&st->destroy_lock))
738 + fatal("PLUGINSD: chart '%s' of host '%s' is being collected while is being destroyed.", rrdset_id(st), rrdhost_hostname(st->rrdhost));
739 +
740 rrdset_isnot_obsolete___safe_from_collector_thread(st);
741 + spinlock_unlock(&st->destroy_lock);
742 + }
743
744 timing_step(TIMING_STEP_BEGIN2_FIND_CHART);
745
@@ -876,8 +879,13 @@ static ALWAYS_INLINE PARSER_RC pluginsd_set_v2(char **words, size_t num_words, P
879
880 st->pluginsd.set = true;
881
879 - if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE | RRDDIM_FLAG_ARCHIVED)))
882 + if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))) {
883 + if(!spinlock_trylock(&rd->destroy_lock))
884 + fatal("PLUGINSD: dimension '%s' of chart '%s' is being collected while is being destroyed.", rrddim_id(rd), rrdset_id(st));
885 +
886 rrddim_isnot_obsolete___safe_from_collector_thread(st, rd);
887 + spinlock_unlock(&rd->destroy_lock);
888 + }
889
890 timing_step(TIMING_STEP_SET2_LOOKUP_DIMENSION);
891
@@ -1230,6 +1238,14 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, int fd_input,
1238 else
1239 cd->serial_failures++;
1240
1241 + // mark all charts of this plugin as obsolete
1242 + RRDSET *st;
1243 + rrdset_foreach_read(st, localhost) {
1244 + if(st->collector_tid == gettid_cached())
1245 + rrdset_is_obsolete___safe_from_collector_thread(st);
1246 + }
1247 + rrdset_foreach_done(st);
1248 +
1249 pluginsd_process_cleanup(parser);
1250 rrd_collector_finished();
1251
src/plugins.d/pluginsd_replication.c
+10 -21
@@ -262,29 +262,18 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARS
262 flags_str = "";
263
264 if (likely(value_str)) {
265 - RRDDIM_FLAGS rd_flags = rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE | RRDDIM_FLAG_ARCHIVED);
265 + NETDATA_DOUBLE value = str2ndd_encoded(value_str, NULL);
266 + SN_FLAGS flags = pluginsd_parse_storage_number_flags(flags_str);
267
267 - if(!(rd_flags & RRDDIM_FLAG_ARCHIVED)) {
268 - NETDATA_DOUBLE value = str2ndd_encoded(value_str, NULL);
269 - SN_FLAGS flags = pluginsd_parse_storage_number_flags(flags_str);
270 -
271 - if (!netdata_double_isnumber(value) || (flags == SN_EMPTY_SLOT)) {
272 - value = NAN;
273 - flags = SN_EMPTY_SLOT;
274 - }
275 -
276 - rrddim_store_metric(rd, parser->user.replay.end_time_ut, value, flags);
277 - rd->collector.last_collected_time.tv_sec = parser->user.replay.end_time;
278 - rd->collector.last_collected_time.tv_usec = 0;
279 - rd->collector.counter++;
280 - }
281 - else {
282 - nd_log_limit_static_global_var(erl, 1, 0);
283 - nd_log_limit(&erl, NDLS_COLLECTORS, NDLP_WARNING,
284 - "PLUGINSD REPLAY ERROR: 'host:%s/chart:%s/dim:%s' has the ARCHIVED flag set, but it is replicated. "
285 - "Ignoring data.",
286 - rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_name(rd));
268 + if (!netdata_double_isnumber(value) || (flags == SN_EMPTY_SLOT)) {
269 + value = NAN;
270 + flags = SN_EMPTY_SLOT;
271 }
272 +
273 + rrddim_store_metric(rd, parser->user.replay.end_time_ut, value, flags);
274 + rd->collector.last_collected_time.tv_sec = parser->user.replay.end_time;
275 + rd->collector.last_collected_time.tv_usec = 0;
276 + rd->collector.counter++;
277 }
278
279 return PARSER_RC_OK;
src/streaming/stream-receiver-connection.c
+5
@@ -6,6 +6,8 @@
6 #include "web/server/h2o/http_server.h"
7 #include "stream-replication-sender.h"
8
9 +void svc_rrdhost_obsolete_all_charts(RRDHOST *host);
10 +
11 // --------------------------------------------------------------------------------------------------------------------
12
13 static void stream_receiver_connected_msg(RRDHOST *host, char *dst, size_t len) {
@@ -703,6 +705,9 @@ int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_
705 if(stream_receiver_send_first_response(rpt)) {
706 // we are the receiver of the node
707
708 + // mark all charts as obsolete
709 + svc_rrdhost_obsolete_all_charts(rpt->host);
710 +
711 char msg[256];
712 stream_receiver_connected_msg(rpt->host, msg, sizeof(msg));
713 stream_receiver_log_status(rpt, msg, 0, NDLP_INFO);
src/streaming/stream-receiver.c
-3
@@ -1125,8 +1125,6 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
1125 __atomic_store_n(&rpt->exit.shutdown, false, __ATOMIC_RELEASE);
1126 host->stream.rcv.status.last_connected = now_realtime_sec();
1127 host->stream.rcv.status.last_disconnected = 0;
1128 - host->stream.rcv.status.last_chart = 0;
1129 - host->stream.rcv.status.check_obsolete = true;
1128
1129 if (rpt->config.health.enabled != CONFIG_BOOLEAN_NO) {
1130 if (rpt->config.health.delay > 0) {
@@ -1206,7 +1204,6 @@ void rrdhost_clear_receiver(struct receiver_state *rpt, STREAM_HANDSHAKE reason)
1204 host->stream.rcv.status.reason = rpt->exit.reason;
1205 rpt->exit.reason = 0;
1206 __atomic_store_n(&rpt->exit.shutdown, false, __ATOMIC_RELEASE);
1209 - host->stream.rcv.status.check_obsolete = false;
1207 host->stream.rcv.status.last_connected = 0;
1208 host->stream.rcv.status.last_disconnected = now_realtime_sec();
1209 host->health.enabled = false;
src/web/api/functions/function-metrics-cardinality.c
+1 -1
@@ -1,7 +1,7 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "function-metrics-cardinality.h"
4 -#include "database/contexts/internal.h"
4 +#include "database/contexts/rrdcontext-internal.h"
5
6 struct counts {
7 size_t nodes;