Fix metric retention check and cleanup (#19278)
fixed bug in mrg cleanup, not deleting metrics that do not have retention fix for mrg acquired and referenced going negative Co-authored-by: Costa Tsaousis <costa@netdata.cloud>
Stelios Fragkakis committed
Dec 23, 2024 at 16:39 UTC
b985604c88a8a3c52738d2fafc1c0adc657409be
3 files changed
+19
-11
src/daemon/pulse/pulse-db-dbengine.c
+1
-1
@@ -836,7 +836,7 @@ void pulse_dbengine_do(bool extended) {
836
priority++;
837
838
rrddim_set_by_pointer(st_mrg_metrics, rd_mrg_metrics, (collected_number)mrg_stats.entries);
839
- rrddim_set_by_pointer(st_mrg_metrics, rd_mrg_acquired, (collected_number)mrg_stats.entries_referenced);
839
+ rrddim_set_by_pointer(st_mrg_metrics, rd_mrg_acquired, (collected_number)mrg_stats.entries_acquired);
840
rrddim_set_by_pointer(st_mrg_metrics, rd_mrg_collected, (collected_number)mrg_stats.writers);
841
rrddim_set_by_pointer(st_mrg_metrics, rd_mrg_multiple_writers, (collected_number)mrg_stats.writers_conflicts);
842
src/database/engine/metric.c
+16
-8
@@ -154,7 +154,12 @@ static void metric_log(MRG *mrg __maybe_unused, METRIC *metric, const char *msg)
154
static inline bool acquired_metric_has_retention(MRG *mrg, METRIC *metric) {
155
time_t first, last;
156
mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
157
- return (!first || !last || first > last);
157
+ bool rc = (first != 0 && last != 0 && first <= last);
158
+
159
+ if(!rc && __atomic_load_n(&mrg->index[metric->partition].stats.writers, __ATOMIC_RELAXED) > 0)
160
+ rc = true;
161
+
162
+ return rc;
163
}
164
165
static inline void acquired_for_deletion_metric_delete(MRG *mrg, METRIC *metric) {
@@ -217,14 +222,14 @@ static inline bool metric_acquire(MRG *mrg, METRIC *metric) {
222
size_t partition = metric->partition;
223
224
if(desired == 1)
220
- __atomic_add_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
225
+ __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
226
227
__atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
228
229
return true;
230
}
231
227
-static inline bool metric_release(MRG *mrg, METRIC *metric, bool delete_if_last_without_retention) {
232
+static inline bool metric_release(MRG *mrg, METRIC *metric) {
233
size_t partition = metric->partition;
234
REFCOUNT expected, desired;
235
@@ -236,7 +241,7 @@ static inline bool metric_release(MRG *mrg, METRIC *metric, bool delete_if_last_
241
fatal("METRIC: refcount is %d (zero or negative) during release", expected);
242
}
243
239
- if(expected == 1 && delete_if_last_without_retention && !acquired_metric_has_retention(mrg, metric))
244
+ if(expected == 1 && !acquired_metric_has_retention(mrg, metric))
245
desired = REFCOUNT_DELETING;
246
else
247
desired = expected - 1;
@@ -244,7 +249,7 @@ static inline bool metric_release(MRG *mrg, METRIC *metric, bool delete_if_last_
249
} while(!__atomic_compare_exchange_n(&metric->refcount, &expected, desired, false, __ATOMIC_RELEASE, __ATOMIC_RELAXED));
250
251
if(desired == 0 || desired == REFCOUNT_DELETING) {
247
- __atomic_sub_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
252
+ __atomic_sub_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
253
254
if(desired == REFCOUNT_DELETING)
255
acquired_for_deletion_metric_delete(mrg, metric);
@@ -318,6 +323,9 @@ static inline METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *r
323
metric->partition = partition;
324
*PValue = metric;
325
326
+ __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
327
+ __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
328
+
329
MRG_STATS_ADDED_METRIC(mrg, partition);
330
331
mrg_index_write_unlock(mrg, partition);
@@ -411,7 +419,7 @@ inline METRIC *mrg_metric_get_and_acquire(MRG *mrg, nd_uuid_t *uuid, Word_t sect
419
}
420
421
inline bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
414
- return metric_release(mrg, metric, true);
422
+ return metric_release(mrg, metric);
423
}
424
425
inline METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
@@ -420,7 +428,7 @@ inline METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
428
}
429
430
inline void mrg_metric_release(MRG *mrg, METRIC *metric) {
423
- metric_release(mrg, metric, false);
431
+ metric_release(mrg, metric);
432
}
433
434
inline Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
@@ -717,7 +725,7 @@ inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
725
726
for(size_t i = 0; i < mrg->partitions ;i++) {
727
s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
720
- s->entries_referenced += __atomic_load_n(&mrg->index[i].stats.entries_referenced, __ATOMIC_RELAXED);
728
+ s->entries_acquired += __atomic_load_n(&mrg->index[i].stats.entries_acquired, __ATOMIC_RELAXED);
729
s->size += __atomic_load_n(&mrg->index[i].stats.size, __ATOMIC_RELAXED);
730
s->current_references += __atomic_load_n(&mrg->index[i].stats.current_references, __ATOMIC_RELAXED);
731
s->additions += __atomic_load_n(&mrg->index[i].stats.additions, __ATOMIC_RELAXED);
src/database/engine/metric.h
+2
-2
@@ -31,10 +31,10 @@ struct mrg_statistics {
31
// --- atomic --- multiple readers / writers
32
33
CACHE_LINE_PADDING();
34
- size_t entries_referenced;
34
+ ssize_t entries_acquired;
35
36
CACHE_LINE_PADDING();
37
- size_t current_references;
37
+ ssize_t current_references;
38
39
CACHE_LINE_PADDING();
40
size_t search_hits;