Improve retention calculation after datafile deletion (#20350)
* Improve retention calculation after datafile deletion * No more entries in this journalfile, remaining source uuids do not match * Quick check if we are on the uuid we want to avoid bsearch * Make sure we have a datafile before releasing * Add checks to prevent indexing and database rotation during agent shutdown
Stelios Fragkakis committed
Jun 3, 2025 at 23:06 UTC
17320a50c26a4478a1f5340bd4f04c5dcee0fd55
1 file changed
+84
-24
src/database/engine/rrdengine.c
+84
-24
@@ -966,7 +966,7 @@ struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struc
966
return next_datafile;
967
}
968
969
-time_t find_uuid_first_time(
969
+static time_t find_uuid_first_time(
970
struct rrdengine_instance *ctx,
971
struct rrdengine_datafile *datafile,
972
struct uuid_first_time_s *uuid_first_entry_list,
@@ -987,6 +987,7 @@ time_t find_uuid_first_time(
987
size_t binary_match = 0;
988
size_t not_matching_bsearches = 0;
989
990
+ bool agent_shutdown = false;
991
while (datafile) {
992
struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0);
993
if (!j2_header) {
@@ -994,45 +995,59 @@ time_t find_uuid_first_time(
995
continue;
996
}
997
997
- time_t journal_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
998
+ bool any_matching = false;
999
999
- if(journal_start_time_s < global_first_time_s)
1000
+ time_t journal_start_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
1001
+
1002
+ if (journal_start_time_s < global_first_time_s)
1003
global_first_time_s = journal_start_time_s;
1004
1002
- struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
1005
+ struct journal_metric_list *uuid_list =
1006
+ (struct journal_metric_list *)((uint8_t *)j2_header + j2_header->metric_offset);
1007
struct uuid_first_time_s *uuid_original_entry;
1008
1009
size_t journal_metric_count = j2_header->metric_count;
1010
char file_path[RRDENG_PATH_MAX];
1011
journalfile_v2_generate_path(datafile, file_path, sizeof(file_path));
1012
PROTECTED_ACCESS_SETUP(datafile->journalfile->mmap.data, datafile->journalfile->mmap.size, file_path, "read");
1009
- if(no_signal_received) {
1013
+ if (no_signal_received) {
1014
+ size_t journal_search_start = 0; // Start of remaining search space
1015
+ any_matching = false;
1016
for (size_t index = 0; index < count; ++index) {
1017
uuid_original_entry = &uuid_first_entry_list[index];
1018
1013
- // Check here if we should skip this
1019
if (uuid_original_entry->df_matched > 3 || uuid_original_entry->pages_found > 5)
1020
continue;
1021
1017
- struct journal_metric_list *live_entry = bsearch(
1018
- uuid_original_entry->uuid,
1019
- uuid_list,
1020
- journal_metric_count,
1021
- sizeof(*uuid_list),
1022
- journal_metric_uuid_compare);
1022
+ any_matching = true;
1023
+ struct journal_metric_list *live_entry = &uuid_list[journal_search_start];
1024
+ // Check if we avoid bsearch
1025
+ if (journal_metric_uuid_compare(uuid_original_entry->uuid, live_entry->uuid) != 0) {
1026
+ live_entry = bsearch(
1027
+ uuid_original_entry->uuid,
1028
+ uuid_list + journal_search_start,
1029
+ journal_metric_count - journal_search_start,
1030
+ sizeof(*uuid_list),
1031
+ journal_metric_uuid_compare);
1032
+
1033
+ if (!live_entry) {
1034
+ not_matching_bsearches++;
1035
+ continue;
1036
+ }
1037
+ }
1038
+
1039
+ size_t found_index = live_entry - uuid_list;
1040
+ journal_search_start = found_index + 1; // Next search starts after this match
1041
1024
- if (!live_entry) {
1025
- // Not found in this journal
1026
- not_matching_bsearches++;
1027
- continue;
1042
+ if (journal_search_start >= journal_metric_count) {
1043
+ not_matching_bsearches += (count - index - 1);
1044
+ break;
1045
}
1046
1047
uuid_original_entry->pages_found += live_entry->entries;
1048
uuid_original_entry->df_matched++;
1049
1050
time_t old_first_time_s = uuid_original_entry->first_time_s;
1034
-
1035
- // Calculate first / last for this match
1051
time_t first_time_s = live_entry->delta_start_s + journal_start_time_s;
1052
uuid_original_entry->first_time_s = MIN(uuid_original_entry->first_time_s, first_time_s);
1053
@@ -1040,17 +1055,34 @@ time_t find_uuid_first_time(
1055
uuid_original_entry->df_index_oldest = uuid_original_entry->df_matched;
1056
1057
binary_match++;
1058
+
1059
+ if (unlikely(!ctx_is_available_for_queries(ctx))) {
1060
+ agent_shutdown = true;
1061
+ break;
1062
+ }
1063
}
1064
} else {
1045
- nd_log_daemon(
1046
- NDLP_ERR, "DBENGINE: journalfile \"%s\" is corrupted, skipping it", file_path);
1065
+ nd_log_daemon(NDLP_ERR, "DBENGINE: journalfile \"%s\" is corrupted, skipping it", file_path);
1066
+ }
1067
+ journalfile_v2_data_release(datafile->journalfile);
1068
+
1069
+ if (agent_shutdown) {
1070
+ datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1071
+ break;
1072
}
1073
1074
journalfile_count++;
1050
- journalfile_v2_data_release(datafile->journalfile);
1075
datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1076
+ if (!any_matching) {
1077
+ if (datafile)
1078
+ datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1079
+ break;
1080
+ }
1081
}
1082
1083
+ if (agent_shutdown)
1084
+ return global_first_time_s;
1085
+
1086
// Let's scan the open cache for almost exact match
1087
size_t open_cache_count = 0;
1088
@@ -1174,15 +1206,28 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1206
1207
global_first_time_s = find_uuid_first_time(ctx, first_datafile_remaining, uuid_first_entry_list, added);
1208
1209
+ if (!ctx_is_available_for_queries(ctx)) {
1210
+ for (size_t index = 0; index < added; ++index) {
1211
+ uuid_first_t_entry = &uuid_first_entry_list[index];
1212
+ mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1213
+ }
1214
+ goto done;
1215
+ }
1216
+
1217
if(worker)
1218
worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1219
1180
- netdata_log_info("DBENGINE: updating tier %d metrics registry retention for %zu metrics",
1181
- ctx->config.tier, added);
1220
+ netdata_log_info("DBENGINE: updating tier %d metrics registry retention for %zu metrics", ctx->config.tier, added);
1221
1222
size_t deleted_metrics = 0, zero_retention_referenced = 0, zero_disk_retention = 0, zero_disk_but_live = 0;
1223
for (size_t index = 0; index < added; ++index) {
1224
uuid_first_t_entry = &uuid_first_entry_list[index];
1225
+
1226
+ if (!ctx_is_available_for_queries(ctx)) {
1227
+ mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1228
+ continue;
1229
+ }
1230
+
1231
if (likely(uuid_first_t_entry->first_time_s != LONG_MAX)) {
1232
1233
time_t old_first_time_s = mrg_metric_get_first_time_s(main_mrg, uuid_first_t_entry->metric);
@@ -1223,7 +1268,9 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1268
}
1269
}
1270
}
1226
- freez(uuid_first_entry_list);
1271
+
1272
+ if (!ctx_is_available_for_queries(ctx))
1273
+ goto done;
1274
1275
internal_error(zero_disk_retention,
1276
"DBENGINE: deleted %zu metrics, zero retention but referenced %zu (out of %zu total, of which %zu have main cache retention) zero on-disk retention tier %d metrics from metrics registry",
@@ -1232,6 +1279,9 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1279
if(global_first_time_s != LONG_MAX)
1280
__atomic_store_n(&ctx->atomic.first_time_s, global_first_time_s, __ATOMIC_RELAXED);
1281
1282
+done:
1283
+ freez(uuid_first_entry_list);
1284
+
1285
if(worker)
1286
worker_is_idle();
1287
}
@@ -1274,6 +1324,13 @@ void datafile_delete(
1324
if (update_retention)
1325
update_metrics_first_time_s(ctx, datafile, datafile->next, worker);
1326
1327
+ if (!ctx_is_available_for_queries(ctx)) {
1328
+ // agent is shutting down, we cannot continue
1329
+ if(worker)
1330
+ worker_is_idle();
1331
+ return;
1332
+ }
1333
+
1334
__atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);
1335
netdata_log_info("DBENGINE: deleting data file \"%s/"
1336
DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
@@ -1735,6 +1792,9 @@ static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(
1792
static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx, void *data, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1793
unsigned count = 0;
1794
1795
+ if (unlikely(!ctx_is_available_for_queries(ctx)))
1796
+ return data;
1797
+
1798
worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
1799
struct rrdengine_datafile *datafile = NULL;
1800
char path[RRDENG_PATH_MAX];