@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 498 lines 17.7 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "mrg-internals.h"
4
5 struct aral_statistics mrg_aral_statistics;
6
7 // ----------------------------------------------------------------------------
8 // private helpers
9
10 static MRG *mrg_create_internal(bool load_from_db) {
11 MRG *mrg = callocz(1, sizeof(MRG));
12
13 for(size_t i = 0; i < _countof(mrg->index) ; i++) {
14 rw_spinlock_init(&mrg->index[i].rw_spinlock);
15
16 char buf[ARAL_MAX_NAME + 1];
17 snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i);
18
19 mrg->index[i].aral = aral_create(buf, sizeof(METRIC), 0, 16384, &mrg_aral_statistics, NULL, NULL,
20 false, false, true);
21 }
22 pulse_aral_register_statistics(&mrg_aral_statistics, "mrg");
23
24 if(load_from_db)
25 mrg_load(mrg);
26
27 return mrg;
28 }
29
30 // ----------------------------------------------------------------------------
31 // public API
32
33 inline MRG *mrg_create(void) {
34 return mrg_create_internal(true);
35 }
36
37 inline MRG *mrg_create_for_unittest(void) {
38 // Skip mrg_load() to avoid pre-loaded metrics with writer counts
39 // This allows deletion tests to work without interference from database metrics
40 return mrg_create_internal(false);
41 }
42
43 struct aral_statistics *mrg_aral_stats(void) {
44 return &mrg_aral_statistics;
45 }
46
47 size_t mrg_destroy(MRG *mrg) {
48 if (!mrg)
49 return 0;
50
51 size_t referenced = 0;
52
53 // Traverse all partitions
54 for (size_t partition = 0; partition < UUIDMAP_PARTITIONS; partition++) {
55 // Lock the partition to prevent new entries while we're cleaning up
56 mrg_index_write_lock(mrg, partition);
57
58 Word_t uuid_index = 0;
59 Pvoid_t *uuid_pvalue;
60
61 // Traverse all UUIDs in this partition
62 for (uuid_pvalue = JudyLFirst(mrg->index[partition].uuid_judy, &uuid_index, PJE0);
63 uuid_pvalue != NULL && uuid_pvalue != PJERR;
64 uuid_pvalue = JudyLNext(mrg->index[partition].uuid_judy, &uuid_index, PJE0)) {
65
66 if (!(*uuid_pvalue))
67 continue;
68
69 // Get the sections judy for this UUID
70 Pvoid_t sections_judy = *uuid_pvalue;
71 Word_t section_index = 0;
72 Pvoid_t *section_pvalue;
73
74 // Traverse all sections for this UUID
75 for (section_pvalue = JudyLFirst(sections_judy, &section_index, PJE0);
76 section_pvalue != NULL && section_pvalue != PJERR;
77 section_pvalue = JudyLNext(sections_judy, &section_index, PJE0)) {
78
79 if (!(*section_pvalue))
80 continue;
81
82 METRIC *metric = *section_pvalue;
83
84 // Try to acquire metric for deletion
85 if (!refcount_acquire_for_deletion(&metric->refcount))
86 referenced++;
87
88 uuidmap_free(metric->uuid);
89 MRG_STATS_DELETED_METRIC(mrg, partition, metric->section);
90 aral_freez(mrg->index[partition].aral, metric);
91 }
92
93 JudyLFreeArray(&sections_judy, PJE0);
94 }
95
96 JudyLFreeArray(&mrg->index[partition].uuid_judy, PJE0);
97
98 // Unlock the partition
99 mrg_index_write_unlock(mrg, partition);
100
101 // Destroy the aral for this partition
102 aral_destroy(mrg->index[partition].aral);
103 }
104
105 // Unregister the aral statistics
106 pulse_aral_unregister_statistics(&mrg_aral_statistics);
107
108 // Free the MRG structure
109 freez(mrg);
110
111 return referenced;
112 }
113
114 ALWAYS_INLINE
115 METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) {
116 // internal_fatal(entry.latest_time_s > max_acceptable_collected_time(),
117 // "DBENGINE METRIC: metric latest time is in the future");
118
119 return metric_add_and_acquire(mrg, &entry, ret);
120 }
121
122 ALWAYS_INLINE
123 METRIC *mrg_metric_get_and_acquire_by_uuid(MRG *mrg, nd_uuid_t *uuid, Word_t section) {
124 UUIDMAP_ID id = uuidmap_create(*uuid);
125 METRIC *metric = metric_get_and_acquire_by_id(mrg, id, section);
126 uuidmap_free(id);
127 return metric;
128 }
129
130 ALWAYS_INLINE
131 METRIC *mrg_metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section) {
132 return metric_get_and_acquire_by_id(mrg, id, section);
133 }
134
135 ALWAYS_INLINE
136 bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
137 return metric_release(mrg, metric);
138 }
139
140 ALWAYS_INLINE
141 METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
142 if(!metric_acquire(mrg, metric))
143 return NULL;
144
145 return metric;
146 }
147
148 ALWAYS_INLINE
149 bool mrg_metric_release(MRG *mrg, METRIC *metric) {
150 return metric_release(mrg, metric);
151 }
152
153 ALWAYS_INLINE
154 Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
155 return (Word_t)metric;
156 }
157
158 ALWAYS_INLINE
159 nd_uuid_t *mrg_metric_uuid(MRG *mrg __maybe_unused, METRIC *metric) {
160 return uuidmap_uuid_ptr(metric->uuid);
161 }
162
163 ALWAYS_INLINE
164 UUIDMAP_ID mrg_metric_uuidmap_id_dup(MRG *mrg __maybe_unused, METRIC *metric) {
165 return uuidmap_dup(metric->uuid);
166 }
167
168 ALWAYS_INLINE
169 Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) {
170 return metric->section;
171 }
172
173 ALWAYS_INLINE
174 bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
175 internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
176
177 if(first_time_s == LONG_MAX)
178 first_time_s = 0;
179
180 if(unlikely(first_time_s < 0))
181 return false;
182
183 __atomic_store_n(&metric->first_time_s, first_time_s, __ATOMIC_RELAXED);
184
185 return true;
186 }
187
188 ALWAYS_INLINE
189 void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, uint32_t update_every_s) {
190 internal_fatal(first_time_s < 0 || last_time_s < 0,
191 "DBENGINE METRIC: timestamp is negative");
192 internal_fatal(first_time_s > max_acceptable_collected_time(),
193 "DBENGINE METRIC: metric first time is in the future");
194 internal_fatal(last_time_s > max_acceptable_collected_time(),
195 "DBENGINE METRIC: metric last time is in the future");
196
197 if(first_time_s > 0 && first_time_s != LONG_MAX)
198 set_metric_field_with_condition(metric->first_time_s, first_time_s, _current <= 0 || (_wanted != 0 && _wanted != LONG_MAX && _wanted < _current));
199
200 if(last_time_s > 0) {
201 if(set_metric_field_with_condition(metric->latest_time_s_clean, last_time_s, _current <= 0 || _wanted > _current) &&
202 update_every_s > 0)
203 // set the latest update every too
204 set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
205 }
206 else if(update_every_s > 0)
207 // set it only if it is invalid
208 set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
209 }
210
211 ALWAYS_INLINE
212 bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
213 internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
214 return set_metric_field_with_condition(metric->first_time_s, first_time_s, _wanted != 0 && _wanted != LONG_MAX && _wanted > _current);
215 }
216
217 ALWAYS_INLINE
218 time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
219 return mrg_metric_get_first_time_s_smart(mrg, metric);
220 }
221
222 void mrg_metric_clear_retention(MRG *mrg __maybe_unused, METRIC *metric) {
223 __atomic_store_n(&metric->first_time_s, 0, __ATOMIC_RELAXED);
224 __atomic_store_n(&metric->latest_time_s_clean, 0, __ATOMIC_RELAXED);
225 __atomic_store_n(&metric->latest_time_s_hot, 0, __ATOMIC_RELAXED);
226 }
227
228 ALWAYS_INLINE_HOT
229 void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, uint32_t *update_every_s) {
230 time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
231 time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
232
233 *last_time_s = MAX(clean, hot);
234 *first_time_s = mrg_metric_get_first_time_s_smart(mrg, metric);
235 if (update_every_s)
236 *update_every_s = __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
237 }
238
239 ALWAYS_INLINE
240 bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
241 internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
242
243 // internal_fatal(latest_time_s > max_acceptable_collected_time(),
244 // "DBENGINE METRIC: metric latest time is in the future");
245
246 // internal_fatal(metric->latest_time_s_clean > latest_time_s,
247 // "DBENGINE METRIC: metric new clean latest time is older than the previous one");
248
249 if(latest_time_s > 0) {
250 if(set_metric_field_with_condition(metric->latest_time_s_clean, latest_time_s, true)) {
251 set_metric_field_with_condition(metric->first_time_s, latest_time_s, _current <= 0 || _wanted < _current);
252
253 return true;
254 }
255 }
256
257 return false;
258 }
259
260 // returns true when metric still has retention
261 ALWAYS_INLINE
262 bool mrg_metric_has_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
263 Word_t section = mrg_metric_section(mrg, metric);
264 bool do_again = false;
265 size_t countdown = 5;
266
267 do {
268 time_t min_first_time_s = LONG_MAX;
269 time_t max_end_time_s = 0;
270 PGC_PAGE *page;
271 PGC_SEARCH method = PGC_SEARCH_FIRST;
272 time_t page_first_time_s = 0;
273 while ((page = pgc_page_get_and_acquire(main_cache, section, (Word_t)metric, page_first_time_s, method))) {
274 method = PGC_SEARCH_NEXT;
275
276 bool is_hot = pgc_is_page_hot(page);
277 bool is_dirty = pgc_is_page_dirty(page);
278 page_first_time_s = pgc_page_start_time_s(page);
279 time_t page_end_time_s = pgc_page_end_time_s(page);
280
281 if ((is_hot || is_dirty) && page_first_time_s > 0 && page_first_time_s < min_first_time_s)
282 min_first_time_s = page_first_time_s;
283
284 if (is_dirty && page_end_time_s > max_end_time_s)
285 max_end_time_s = page_end_time_s;
286
287 pgc_page_release(main_cache, page);
288 }
289
290 if (min_first_time_s == LONG_MAX)
291 min_first_time_s = 0;
292
293 if (--countdown && !min_first_time_s && __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED))
294 do_again = true;
295 else {
296 internal_error(!countdown, "METRIC: giving up on updating the retention of metric without disk retention");
297
298 do_again = false;
299 set_metric_field_with_condition(metric->first_time_s, min_first_time_s, true);
300 set_metric_field_with_condition(metric->latest_time_s_clean, max_end_time_s, true);
301 }
302 } while(do_again);
303
304 time_t first, last;
305 mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
306 return (first && last && first < last);
307 }
308
309 ALWAYS_INLINE_HOT
310 bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
311 internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
312
313 // internal_fatal(latest_time_s > max_acceptable_collected_time(),
314 // "DBENGINE METRIC: metric latest time is in the future");
315
316 if(likely(latest_time_s > 0)) {
317 __atomic_store_n(&metric->latest_time_s_hot, latest_time_s, __ATOMIC_RELAXED);
318 return true;
319 }
320
321 return false;
322 }
323
324 ALWAYS_INLINE
325 time_t mrg_metric_get_latest_clean_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
326 time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
327 return clean;
328 }
329
330 ALWAYS_INLINE_HOT
331 time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
332 time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
333 time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
334
335 return MAX(clean, hot);
336 }
337
338 ALWAYS_INLINE
339 bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
340 if(likely(update_every_s > 0))
341 return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
342
343 return false;
344 }
345
346 ALWAYS_INLINE_HOT
347 bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
348 if(likely(update_every_s > 0))
349 return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
350
351 return false;
352 }
353
354 ALWAYS_INLINE
355 uint32_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
356 return __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
357 }
358
359 #ifdef NETDATA_INTERNAL_CHECKS
360 ALWAYS_INLINE bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
361 pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
362 pid_t wanted = gettid_cached();
363 bool done = true;
364
365 do {
366 if(expected != 0) {
367 done = false;
368 break;
369 }
370 } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
371
372 if(done)
373 __atomic_add_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
374 else
375 __atomic_add_fetch(&mrg->index[metric->partition].stats.writers_conflicts, 1, __ATOMIC_RELAXED);
376
377 return done;
378 }
379
380 ALWAYS_INLINE bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
381 // this function can be called from a different thread than the one than the writer
382
383 pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
384 pid_t wanted = 0;
385 bool done = true;
386
387 do {
388 if(!expected) {
389 done = false;
390 break;
391 }
392 } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
393
394 if(done)
395 __atomic_sub_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
396
397 return done;
398 }
399 #endif
400
401 inline void mrg_update_metric_retention_and_granularity_by_uuid(
402 MRG *mrg,
403 Word_t section,
404 nd_uuid_t(*uuid),
405 time_t first_time_s,
406 time_t last_time_s,
407 uint32_t update_every_s,
408 time_t now_s,
409 uint64_t *journal_samples)
410 {
411 if(unlikely(last_time_s > now_s)) {
412 nd_log_limit_static_global_var(erl, 1, 0);
413 nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
414 "DBENGINE JV2: wrong last time on-disk (%ld - %ld, now %ld), "
415 "fixing last time to now",
416 first_time_s, last_time_s, now_s);
417 last_time_s = now_s;
418 }
419
420 if (unlikely(first_time_s > last_time_s)) {
421 nd_log_limit_static_global_var(erl, 1, 0);
422 nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
423 "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
424 "fixing first time to last time",
425 first_time_s, last_time_s, now_s);
426
427 first_time_s = last_time_s;
428 }
429
430 if (unlikely(first_time_s == 0 || last_time_s == 0)) {
431 nd_log_limit_static_global_var(erl, 1, 0);
432 nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
433 "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
434 "using them as-is",
435 first_time_s, last_time_s, now_s);
436 }
437
438 bool added = false;
439 METRIC *metric = mrg_metric_get_and_acquire_by_uuid(mrg, uuid, section);
440 if (!metric) {
441 MRG_ENTRY entry = {
442 .uuid = uuid,
443 .section = section,
444 .first_time_s = first_time_s,
445 .last_time_s = last_time_s,
446 .latest_update_every_s = update_every_s,
447 };
448 metric = mrg_metric_add_and_acquire(mrg, entry, &added);
449 }
450
451 if (likely(!added)) {
452 uint64_t old_samples = 0;
453
454 if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
455 old_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
456
457 mrg_metric_expand_retention(mrg, metric, first_time_s, last_time_s, update_every_s);
458
459 uint64_t new_samples = 0;
460 if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
461 new_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
462
463 if (journal_samples)
464 *journal_samples += (new_samples - old_samples);
465 }
466 else {
467 // Newly added
468 if (update_every_s) {
469 uint64_t samples = (last_time_s - first_time_s) / update_every_s;
470 if (journal_samples)
471 *journal_samples += samples;
472 }
473 }
474
475 mrg_metric_release(mrg, metric);
476 }
477
478 inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
479 memset(s, 0, sizeof(struct mrg_statistics));
480
481 for(size_t i = 0; i < _countof(mrg->index) ;i++) {
482 s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
483 s->entries_acquired += __atomic_load_n(&mrg->index[i].stats.entries_acquired, __ATOMIC_RELAXED);
484 s->size += __atomic_load_n(&mrg->index[i].stats.size, __ATOMIC_RELAXED);
485 s->current_references += __atomic_load_n(&mrg->index[i].stats.current_references, __ATOMIC_RELAXED);
486 s->additions += __atomic_load_n(&mrg->index[i].stats.additions, __ATOMIC_RELAXED);
487 s->additions_duplicate += __atomic_load_n(&mrg->index[i].stats.additions_duplicate, __ATOMIC_RELAXED);
488 s->deletions += __atomic_load_n(&mrg->index[i].stats.deletions, __ATOMIC_RELAXED);
489 s->delete_having_retention_or_referenced += __atomic_load_n(&mrg->index[i].stats.delete_having_retention_or_referenced, __ATOMIC_RELAXED);
490 s->delete_misses += __atomic_load_n(&mrg->index[i].stats.delete_misses, __ATOMIC_RELAXED);
491 s->search_hits += __atomic_load_n(&mrg->index[i].stats.search_hits, __ATOMIC_RELAXED);
492 s->search_misses += __atomic_load_n(&mrg->index[i].stats.search_misses, __ATOMIC_RELAXED);
493 s->writers += __atomic_load_n(&mrg->index[i].stats.writers, __ATOMIC_RELAXED);
494 s->writers_conflicts += __atomic_load_n(&mrg->index[i].stats.writers_conflicts, __ATOMIC_RELAXED);
495 }
496
497 s->size += sizeof(MRG);
498 }