Improve datafile rotation and indexing (#20354)
* Check tier quota after indexing each file Do not schedule a db rotate command if quota is ok * Cleanup journalfile deletion code Log a warning if journalfile cannot be unmapped * Schedule database rotation / journal indexing in callbacks Indicate if rotation is due to disk quota or time retention Reduce number of opcodes * Change log level for journal indexing messages from DEBUG to INFO * Report acquired file for deletion * Allow indexing during rotation * Allow rotation during indexing * Allow the indexing to run at least once * Fix compilation warning Report full name of the journal file to be indexed Improve disk space reclaim log with human-readable format * Adjust formatting for file logging Additional use of UNLINK_FILE macro Update logging to include journal file paths * More formatting adjustments for file logging
Stelios Fragkakis committed
May 29, 2025 at 18:04 UTC
4812fe155952f6c791cd2266c9ff2da496dd30e2
5 files changed
+184
-119
src/database/engine/datafile.c
+7
-15
@@ -189,20 +189,14 @@ int close_data_file(struct rrdengine_datafile *datafile)
189
int unlink_data_file(struct rrdengine_datafile *datafile)
190
{
191
struct rrdengine_instance *ctx = datafile_ctx(datafile);
192
- uv_fs_t req;
192
int ret;
193
char path[RRDENG_PATH_MAX];
194
195
generate_datafilepath(datafile, path, sizeof(path));
196
198
- ret = uv_fs_unlink(NULL, &req, path, NULL);
199
- if (ret < 0) {
200
- netdata_log_error("DBENGINE: uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));
201
- ctx_fs_error(ctx);
202
- }
203
- uv_fs_req_cleanup(&req);
204
-
205
- __atomic_add_fetch(&ctx->stats.datafile_deletions, 1, __ATOMIC_RELAXED);
197
+ UNLINK_FILE(ctx, path, ret);
198
+ if (ret == 0)
199
+ __atomic_add_fetch(&ctx->stats.datafile_deletions, 1, __ATOMIC_RELAXED);
200
201
return ret;
202
}
@@ -488,7 +482,7 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx, bool having_lock)
482
char path[RRDENG_PATH_MAX];
483
484
nd_log(NDLS_DAEMON, NDLP_DEBUG,
491
- "DBENGINE: creating new data and journal files in path %s",
485
+ "DBENGINE: creating new data and journal files in path \"%s\"",
486
ctx->config.dbfiles_path);
487
488
datafile = datafile_alloc_and_init(ctx, 1, fileno);
@@ -497,8 +491,7 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx, bool having_lock)
491
goto error_after_datafile;
492
493
generate_datafilepath(datafile, path, sizeof(path));
500
- nd_log(NDLS_DAEMON, NDLP_INFO,
501
- "DBENGINE: created data file \"%s\".", path);
494
+ nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE: created data file \"%s\".", path);
495
496
journalfile = journalfile_alloc_and_init(datafile);
497
ret = journalfile_create(journalfile, datafile);
@@ -506,8 +499,7 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx, bool having_lock)
499
goto error_after_journalfile;
500
501
journalfile_v1_generate_path(datafile, path, sizeof(path));
509
- nd_log(NDLS_DAEMON, NDLP_INFO,
510
- "DBENGINE: created journal file \"%s\".", path);
502
+ nd_log(NDLS_DAEMON, NDLP_INFO, "DBENGINE: created journal file \"%s\".", path);
503
504
ctx_current_disk_space_increase(ctx, datafile->pos + journalfile->unsafe.pos);
505
datafile_list_insert(ctx, datafile, having_lock);
@@ -549,7 +541,7 @@ int init_data_files(struct rrdengine_instance *ctx)
541
create_new_datafile_pair(ctx, false);
542
543
while(rrdeng_ctx_tier_cap_exceeded(ctx))
552
- datafile_delete(ctx, ctx->datafiles.first, false, false);
544
+ datafile_delete(ctx, ctx->datafiles.first, false, true, false);
545
}
546
547
pgc_reset_hot_max(open_cache);
src/database/engine/datafile.h
+1
@@ -33,6 +33,7 @@ typedef enum __attribute__ ((__packed__)) {
33
DATAFILE_ACQUIRE_OPEN_CACHE = 0,
34
DATAFILE_ACQUIRE_PAGE_DETAILS,
35
DATAFILE_ACQUIRE_RETENTION,
36
+ DATAFILE_ACQUIRE_INDEXING,
37
38
// terminator
39
DATAFILE_ACQUIRE_MAX,
src/database/engine/journalfile.c
+43
-57
@@ -1,7 +1,6 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
#include "rrdengine.h"
3
4
-
4
// the default value is set in ND_PROFILE, not here
5
time_t dbengine_journal_v2_unmount_time = 120;
6
@@ -265,8 +264,8 @@ static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *jo
264
if (nd_munmap(journalfile->mmap.data, journalfile->mmap.size)) {
265
char path[RRDENG_PATH_MAX];
266
journalfile_v2_generate_path(journalfile->datafile, path, sizeof(path));
268
- netdata_log_error("DBENGINE: failed to unmap index file '%s'", path);
269
- internal_fatal(true, "DBENGINE: failed to unmap file '%s'", path);
267
+ netdata_log_error("DBENGINE: failed to unmap index file \"%s\"", path);
268
+ internal_fatal(true, "DBENGINE: failed to unmap file \"%s\"", path);
269
ctx_fs_error(datafile_ctx(journalfile->datafile));
270
}
271
else {
@@ -430,6 +429,9 @@ static void journalfile_v2_data_unmap_permanently(struct rrdengine_journalfile *
429
njfv2idx_remove(journalfile->datafile);
430
431
bool has_references = false;
432
+ char path_v2[RRDENG_PATH_MAX];
433
+
434
+ journalfile_v2_generate_path(journalfile->datafile, path_v2, sizeof(path_v2));
435
436
do {
437
if (has_references)
@@ -450,7 +452,8 @@ static void journalfile_v2_data_unmap_permanently(struct rrdengine_journalfile *
452
}
453
else {
454
has_references = true;
453
- internal_error(true, "DBENGINE JOURNALFILE: waiting for journalfile to be available to unmap...");
455
+ nd_log_limit_static_global_var(journalfile_erl, 10, 0);
456
+ nd_log_limit(&journalfile_erl, NDLS_DAEMON, NDLP_WARNING, "DBENGINE: journalfile \"%s\" is not available for unmap", path_v2);
457
}
458
459
spinlock_unlock(&journalfile->data_spinlock);
@@ -478,7 +481,7 @@ static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
481
ret = uv_fs_close(NULL, &req, file, NULL);
482
if (ret < 0) {
483
journalfile_v1_generate_path(datafile, path, sizeof(path));
481
- netdata_log_error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));
484
+ netdata_log_error("DBENGINE: uv_fs_close(\"%s\"): %s", path, uv_strerror(ret));
485
ctx_fs_error(datafile_ctx(datafile));
486
}
487
uv_fs_req_cleanup(&req);
@@ -499,20 +502,14 @@ int journalfile_unlink(struct rrdengine_journalfile *journalfile)
502
{
503
struct rrdengine_datafile *datafile = journalfile->datafile;
504
struct rrdengine_instance *ctx = datafile_ctx(datafile);
502
- uv_fs_t req;
505
int ret;
504
- char path[RRDENG_PATH_MAX];
506
507
+ char path[RRDENG_PATH_MAX];
508
journalfile_v1_generate_path(datafile, path, sizeof(path));
509
508
- ret = uv_fs_unlink(NULL, &req, path, NULL);
509
- if (ret < 0) {
510
- netdata_log_error("DBENGINE: uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));
511
- ctx_fs_error(ctx);
512
- }
513
- uv_fs_req_cleanup(&req);
514
-
515
- __atomic_add_fetch(&ctx->stats.journalfile_deletions, 1, __ATOMIC_RELAXED);
510
+ UNLINK_FILE(ctx, path, ret);
511
+ if (ret == 0)
512
+ __atomic_add_fetch(&ctx->stats.journalfile_deletions, 1, __ATOMIC_RELAXED);
513
514
return ret;
515
}
@@ -520,7 +517,6 @@ int journalfile_unlink(struct rrdengine_journalfile *journalfile)
517
int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
518
{
519
struct rrdengine_instance *ctx = datafile_ctx(datafile);
523
- uv_fs_t req;
520
int ret;
521
char path[RRDENG_PATH_MAX];
522
char path_v2[RRDENG_PATH_MAX];
@@ -528,32 +524,20 @@ int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct
524
journalfile_v1_generate_path(datafile, path, sizeof(path));
525
journalfile_v2_generate_path(datafile, path_v2, sizeof(path));
526
531
- if (journalfile->file) {
532
- ret = uv_fs_ftruncate(NULL, &req, journalfile->file, 0, NULL);
533
- if (ret < 0) {
534
- netdata_log_error("DBENGINE: uv_fs_ftruncate(%s): %s", path, uv_strerror(ret));
535
- ctx_fs_error(ctx);
536
- }
537
- uv_fs_req_cleanup(&req);
527
+ if (journalfile->file)
528
(void)close_uv_file(datafile, journalfile->file);
539
- }
529
530
// This is the new journal v2 index file
542
- ret = uv_fs_unlink(NULL, &req, path_v2, NULL);
543
- if (ret < 0) {
544
- netdata_log_error("DBENGINE: uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));
545
- ctx_fs_error(ctx);
546
- }
547
- uv_fs_req_cleanup(&req);
531
+ int deleted = 0;
532
+ UNLINK_FILE(ctx, path_v2, ret);
533
+ if (ret == 0)
534
+ deleted++;
535
549
- ret = uv_fs_unlink(NULL, &req, path, NULL);
550
- if (ret < 0) {
551
- netdata_log_error("DBENGINE: uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));
552
- ctx_fs_error(ctx);
553
- }
554
- uv_fs_req_cleanup(&req);
536
+ UNLINK_FILE(ctx, path, ret);
537
+ if (ret == 0)
538
+ deleted++;
539
556
- __atomic_add_fetch(&ctx->stats.journalfile_deletions, 2, __ATOMIC_RELAXED);
540
+ __atomic_add_fetch(&ctx->stats.journalfile_deletions, deleted, __ATOMIC_RELAXED);
541
542
if(journalfile_v2_data_available(journalfile))
543
journalfile_v2_data_unmap_permanently(journalfile);
@@ -604,7 +588,7 @@ int journalfile_create(struct rrdengine_journalfile *journalfile, struct rrdengi
588
journalfile_destroy_unsafe(journalfile, datafile);
589
ctx_io_error(ctx);
590
nd_log_limit_static_global_var(dbengine_erl, 10, 0);
607
- nd_log_limit(&dbengine_erl, NDLS_DAEMON, NDLP_ERR, "DBENGINE: Failed to create journlfile %s", path);
591
+ nd_log_limit(&dbengine_erl, NDLS_DAEMON, NDLP_ERR, "DBENGINE: Failed to create journlfile \"%s\"", path);
592
return ret;
593
}
594
@@ -1074,13 +1058,13 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1058
if (errno == ENOENT)
1059
return 1;
1060
ctx_fs_error(ctx);
1077
- netdata_log_error("DBENGINE: failed to open '%s'", path_v2);
1061
+ netdata_log_error("DBENGINE: failed to open \"%s\"", path_v2);
1062
return 1;
1063
}
1064
1065
ret = fstat(fd, &statbuf);
1066
if (ret) {
1083
- netdata_log_error("DBENGINE: failed to get file information for '%s'", path_v2);
1067
+ netdata_log_error("DBENGINE: failed to get file information for \"%s\"", path_v2);
1068
close(fd);
1069
return 1;
1070
}
@@ -1088,7 +1072,7 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1072
journal_v2_file_size = (size_t)statbuf.st_size;
1073
1074
if (journal_v2_file_size < sizeof(struct journal_v2_header)) {
1091
- error_report("Invalid file %s. Not the expected size", path_v2);
1075
+ error_report("Invalid file \"%s\". Not the expected size", path_v2);
1076
close(fd);
1077
return 1;
1078
}
@@ -1100,7 +1084,7 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1084
return 1;
1085
}
1086
1103
- nd_log_daemon(NDLP_DEBUG, "DBENGINE: checking integrity of '%s'", path_v2);
1087
+ nd_log_daemon(NDLP_DEBUG, "DBENGINE: checking integrity of \"%s\"", path_v2);
1088
1089
usec_t validation_start_ut = now_monotonic_usec();
1090
@@ -1115,14 +1099,14 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1099
1100
if (unlikely(rc)) {
1101
if (rc == 2)
1118
- error_report("File %s needs to be rebuilt", path_v2);
1102
+ error_report("File \"%s\" needs to be rebuilt", path_v2);
1103
else if (rc == 3)
1120
- error_report("File %s will be skipped", path_v2);
1104
+ error_report("File \"%s\" will be skipped", path_v2);
1105
else
1122
- error_report("File %s is invalid and it will be rebuilt", path_v2);
1106
+ error_report("File \"%s\" is invalid and it will be rebuilt", path_v2);
1107
1108
if (unlikely(nd_munmap(data_start, journal_v2_file_size)))
1125
- netdata_log_error("DBENGINE: failed to unmap '%s'", path_v2);
1109
+ netdata_log_error("DBENGINE: failed to unmap \"%s\"", path_v2);
1110
1111
close(fd);
1112
return rc;
@@ -1133,7 +1117,7 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1117
1118
if (unlikely(!entries)) {
1119
if (unlikely(nd_munmap(data_start, journal_v2_file_size)))
1136
- netdata_log_error("DBENGINE: failed to unmap '%s'", path_v2);
1120
+ netdata_log_error("DBENGINE: failed to unmap \"%s\"", path_v2);
1121
1122
close(fd);
1123
return 1;
@@ -1141,7 +1125,7 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1125
1126
usec_t finished_ut = now_monotonic_usec();
1127
1144
- nd_log_daemon(NDLP_DEBUG, "DBENGINE: journal v2 '%s' loaded, size: %0.2f MiB, metrics: %0.2f k, "
1128
+ nd_log_daemon(NDLP_DEBUG, "DBENGINE: journal v2 \"%s\" loaded, size: %0.2f MiB, metrics: %0.2f k, "
1129
"mmap: %0.2f ms, validate: %0.2f ms"
1130
, path_v2
1131
, (double)journal_v2_file_size / 1024 / 1024
@@ -1318,7 +1302,7 @@ bool journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1302
1303
journalfile_v2_generate_path(datafile, path, sizeof(path));
1304
1321
- netdata_log_info("DBENGINE: indexing file '%s': extents %zu, metrics %zu, pages %zu",
1305
+ netdata_log_info("DBENGINE: indexing file \"%s\": extents %zu, metrics %zu, pages %zu",
1306
path,
1307
number_of_extents,
1308
number_of_metrics,
@@ -1357,7 +1341,7 @@ bool journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1341
int fd_v2;
1342
uint8_t *data_start = nd_mmap_advanced(path, total_file_size, MAP_SHARED, 0, false, true, &fd_v2);
1343
if(!data_start) {
1360
- nd_log_daemon(NDLP_WARNING, "DBENGINE: Failed to allocate %"PRIu64" bytes of memory for journal file '%s'. Will retry later", total_file_size, path);
1344
+ nd_log_daemon(NDLP_WARNING, "DBENGINE: Failed to allocate %"PRIu64" bytes of memory for journal file \"%s\". Will retry later", total_file_size, path);
1345
return false;
1346
}
1347
@@ -1510,7 +1494,9 @@ bool journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1494
internal_error(
1495
true, "DBENGINE: FILE COMPLETED --------> %llu", (now_monotonic_usec() - start_loading) / USEC_PER_MS);
1496
1513
- netdata_log_info("DBENGINE: migrated journal file '%s', file size %zu", path, total_file_size);
1497
+ char size_for_humans[128];
1498
+ size_snprintf(size_for_humans, sizeof(size_for_humans), total_file_size, "B", false);
1499
+ netdata_log_info("DBENGINE: migrated journal file \"%s\", file size %zu bytes (%s)", path, total_file_size, size_for_humans);
1500
1501
// msync(data_start, total_file_size, MS_SYNC);
1502
journalfile_v2_data_set(journalfile, fd_v2, data_start, total_file_size);
@@ -1523,12 +1509,12 @@ bool journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1509
}
1510
}
1511
else {
1526
- nd_log(NDLS_DAEMON, NDLP_ERR, "DBENGINE: failed to write journal file '%s' (SIGBUS)", path);
1512
+ nd_log(NDLS_DAEMON, NDLP_ERR, "DBENGINE: failed to write journal file \"%s\" (SIGBUS)", path);
1513
}
1514
1515
freez(uuid_list);
1516
1531
- netdata_log_info("DBENGINE: failed to build index '%s', file will be skipped", path);
1517
+ netdata_log_info("DBENGINE: failed to build index \"%s\", file will be skipped", path);
1518
1519
nd_munmap(data_start, total_file_size);
1520
unlink(path);
@@ -1579,19 +1565,19 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil
1565
1566
ret = journalfile_check_superblock(file);
1567
if (ret) {
1582
- netdata_log_info("DBENGINE: invalid journal file '%s' ; superblock check failed.", path);
1568
+ netdata_log_info("DBENGINE: invalid journal file \"%s\" ; superblock check failed.", path);
1569
error = ret;
1570
goto cleanup;
1571
}
1572
ctx_io_read_op_bytes(ctx, sizeof(struct rrdeng_jf_sb));
1573
1588
- nd_log_daemon(NDLP_DEBUG, "DBENGINE: loading journal file '%s'", path);
1574
+ nd_log_daemon(NDLP_DEBUG, "DBENGINE: loading journal file \"%s\"", path);
1575
1576
max_id = journalfile_iterate_transactions(ctx, journalfile);
1577
1578
__atomic_store_n(&ctx->atomic.transaction_id, MAX(__atomic_load_n(&ctx->atomic.transaction_id, __ATOMIC_RELAXED), max_id + 1), __ATOMIC_RELAXED);
1579
1594
- nd_log_daemon(NDLP_DEBUG, "DBENGINE: journal file '%s' loaded (size:%" PRIu64 ").", path, file_size);
1580
+ nd_log_daemon(NDLP_DEBUG, "DBENGINE: journal file \"%s\" loaded (size:%" PRIu64 ").", path, file_size);
1581
1582
bool is_last_file = (ctx_last_fileno_get(ctx) == journalfile->datafile->fileno);
1583
if (is_last_file && journalfile->datafile->pos <= rrdeng_target_data_file_size(ctx) / 3) {
@@ -1610,7 +1596,7 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil
1596
cleanup:
1597
ret = uv_fs_close(NULL, &req, file, NULL);
1598
if (ret < 0) {
1613
- netdata_log_error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));
1599
+ netdata_log_error("DBENGINE: uv_fs_close(\"%s\"): %s", path, uv_strerror(ret));
1600
ctx_fs_error(ctx);
1601
}
1602
uv_fs_req_cleanup(&req);
src/database/engine/rrdengine.c
+113
-46
@@ -165,6 +165,21 @@ static inline enum LIBUV_WORKERS_STATUS work_request_full(void) {
165
return LIBUV_WORKERS_RELAXED;
166
}
167
168
+static inline void check_and_schedule_db_rotation(struct rrdengine_instance *ctx)
169
+{
170
+ internal_fatal(rrdeng_main.tid != gettid_cached(), "check_and_schedule_db_rotation() can only be run from the event loop thread");
171
+
172
+ if (ctx->datafiles.pending_rotate) {
173
+ nd_log_daemon(NDLP_DEBUG, "DBENGINE: tier %d is already pending rotation", ctx->config.tier);
174
+ return;
175
+ }
176
+
177
+ if(ctx_is_available_for_queries(ctx) && rrdeng_ctx_tier_cap_exceeded(ctx)) {
178
+ ctx->datafiles.pending_rotate = true;
179
+ rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
180
+ }
181
+}
182
+
183
static inline void work_done(struct rrdeng_work *work_request) {
184
aral_freez(rrdeng_main.work_cmd.ar, work_request);
185
}
@@ -858,6 +873,8 @@ static void after_extent_write(struct rrdengine_instance *ctx __maybe_unused, vo
873
{
874
if(completion)
875
completion_mark_complete(completion);
876
+
877
+ check_and_schedule_db_rotation(ctx);
878
}
879
880
static void *extent_write_tp_worker(
@@ -909,8 +926,6 @@ static void *extent_write_tp_worker(
926
927
extent_flush_to_open(ctx, xt_io_descr, ret < 0);
928
912
- if(ctx_is_available_for_queries(ctx) && rrdeng_ctx_tier_cap_exceeded(ctx))
913
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
929
done:
930
__atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
931
worker_is_idle();
@@ -919,8 +934,11 @@ done:
934
935
static void after_database_rotate(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
936
__atomic_store_n(&ctx->atomic.now_deleting_files, false, __ATOMIC_RELAXED);
937
+
938
if (__atomic_load_n(&ctx->atomic.needs_indexing, __ATOMIC_RELAXED))
923
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
939
+ rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
940
+
941
+ check_and_schedule_db_rotation(ctx);
942
}
943
944
struct uuid_first_time_s {
@@ -1218,7 +1236,13 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1236
worker_is_idle();
1237
}
1238
1221
-void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool update_retention, bool worker) {
1239
+void datafile_delete(
1240
+ struct rrdengine_instance *ctx,
1241
+ struct rrdengine_datafile *datafile,
1242
+ bool update_retention,
1243
+ bool disk_time,
1244
+ bool worker)
1245
+{
1246
if(worker)
1247
worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
1248
@@ -1242,13 +1266,18 @@ void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *
1266
}
1267
}
1268
1269
+ netdata_log_info("DBENGINE: acquired data file \"%s/"
1270
+ DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
1271
+ "\" for deletion.",
1272
+ ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
1273
+
1274
if (update_retention)
1275
update_metrics_first_time_s(ctx, datafile, datafile->next, worker);
1276
1277
__atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);
1249
- netdata_log_info("DBENGINE: deleting data file '%s/"
1278
+ netdata_log_info("DBENGINE: deleting data file \"%s/"
1279
DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
1251
- "'.",
1280
+ "\".",
1281
ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
1282
1283
if(worker)
@@ -1268,7 +1297,7 @@ void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *
1297
journal_file_bytes = journalfile_current_size(journal_file);
1298
deleted_bytes = journalfile_v2_data_size_get(journal_file);
1299
1271
- netdata_log_info("DBENGINE: deleting data and journal files to maintain disk quota");
1300
+ netdata_log_info("DBENGINE: deleting data and journal files to maintain %s", disk_time ? "disk quota" : "time retention");
1301
// This will delete journalfile_v2 and journalfile_v1
1302
ret = journalfile_destroy_unsafe(journal_file, datafile);
1303
if (!ret) {
@@ -1295,14 +1324,13 @@ void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *
1324
freez(datafile);
1325
1326
ctx_current_disk_space_decrease(ctx, deleted_bytes);
1298
- netdata_log_info("DBENGINE: reclaimed %zu bytes of disk space.", deleted_bytes);
1327
+ char size_for_humans[128];
1328
+ size_snprintf(size_for_humans, sizeof(size_for_humans), deleted_bytes, "B", false);
1329
+ netdata_log_info("DBENGINE: reclaimed %zu bytes (%s) of disk space.", deleted_bytes, size_for_humans);
1330
}
1331
1332
static void *database_rotate_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1302
- datafile_delete(ctx, ctx->datafiles.first, ctx_is_available_for_queries(ctx), true);
1303
-
1304
- if (rrdeng_ctx_tier_cap_exceeded(ctx))
1305
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1333
+ datafile_delete(ctx, ctx->datafiles.first, ctx_is_available_for_queries(ctx), true, true);
1334
1335
rrdcontext_db_rotation();
1336
@@ -1675,46 +1703,80 @@ NOT_INLINE_HOT void pdc_route_synchronously_first(struct rrdengine_instance *ctx
1703
pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_asynchronously);
1704
}
1705
1678
-static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1679
- unsigned count = 0;
1706
+static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(struct rrdengine_instance *ctx, struct rrdengine_datafile *release_datafile)
1707
+{
1708
+ struct rrdengine_datafile *datafile = NULL;
1709
+
1710
+ uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1711
+ if (release_datafile) {
1712
+ datafile = release_datafile->next;
1713
+ datafile_release(release_datafile, DATAFILE_ACQUIRE_INDEXING);
1714
+ }
1715
+ else
1716
+ datafile = ctx->datafiles.first;
1717
1681
- struct rrdengine_datafile *datafile = ctx->datafiles.first;
1682
- worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
1683
- count = 0;
1718
while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {
1719
if(journalfile_v2_data_available(datafile->journalfile)) {
1686
- // journal file v2 is already there for this datafile
1720
datafile = datafile->next;
1721
continue;
1722
}
1723
+ bool locked = datafile_acquire(datafile, DATAFILE_ACQUIRE_INDEXING);
1724
+ if (locked) {
1725
+ uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1726
+ return datafile;
1727
+ }
1728
+ nd_log_daemon(NDLP_INFO, "DBENGINE: Datafile %u CANNOT be locked for indexing; skipping", datafile->fileno);
1729
+ }
1730
+ uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1731
+ return NULL;
1732
+}
1733
+
1734
+
1735
+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) {
1736
+ unsigned count = 0;
1737
+
1738
+ worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
1739
+ struct rrdengine_datafile *datafile = NULL;
1740
+ char path[RRDENG_PATH_MAX];
1741
+
1742
+ bool index_once = false;
1743
+ while ((datafile = release_and_aquire_next_datafile_for_indexing(ctx, datafile))) {
1744
1745
spinlock_lock(&datafile->writers.spinlock);
1746
bool available = (datafile->writers.running || datafile->writers.flushed_to_open_running) ? false : true;
1747
spinlock_unlock(&datafile->writers.spinlock);
1748
1749
+ journalfile_v1_generate_path(datafile, path, sizeof(path));
1750
+
1751
if(!available) {
1696
- nd_log(NDLS_DAEMON, NDLP_NOTICE,
1697
- "DBENGINE: journal file %u needs to be indexed, but it has writers working on it - "
1752
+ nd_log_daemon(NDLP_NOTICE,
1753
+ "DBENGINE: journal file \"%s\" needs to be indexed, but it has writers working on it - "
1754
"skipping it for now",
1699
- datafile->fileno);
1700
-
1701
- datafile = datafile->next;
1755
+ path);
1756
continue;
1757
}
1758
1705
- nd_log(NDLS_DAEMON, NDLP_DEBUG,
1706
- "DBENGINE: journal file %u is ready to be indexed",
1707
- datafile->fileno);
1759
+ if (index_once && unlikely(rrdeng_ctx_tier_cap_exceeded(ctx))) {
1760
+ nd_log_daemon(
1761
+ NDLP_INFO, "DBENGINE: tier %d reached quota limit, stopping journal indexing", ctx->config.tier);
1762
+ __atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
1763
+ datafile_release(datafile, DATAFILE_ACQUIRE_INDEXING);
1764
+ break;
1765
+ }
1766
+ nd_log_daemon(NDLP_INFO, "DBENGINE: journal file \"%s\" is ready to be indexed", path);
1767
1768
pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->config.page_type,
1769
journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
1770
1712
- count++;
1771
+ index_once = true;
1772
1714
- datafile = datafile->next;
1773
+ count++;
1774
1716
- if (unlikely(!ctx_is_available_for_queries(ctx)))
1775
+ // check if we are shutting down
1776
+ if (unlikely(!ctx_is_available_for_queries(ctx))) {
1777
+ datafile_release(datafile, DATAFILE_ACQUIRE_INDEXING);
1778
break;
1779
+ }
1780
}
1781
1782
errno_clear();
@@ -1746,7 +1808,8 @@ static void after_do_extent_cache_evict(struct rrdengine_instance *ctx __maybe_u
1808
1809
static void after_journal_v2_indexing(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
1810
__atomic_store_n(&ctx->atomic.migration_to_v2_running, false, __ATOMIC_RELAXED);
1749
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1811
+
1812
+ check_and_schedule_db_rotation(ctx);
1813
}
1814
1815
struct rrdeng_buffer_sizes rrdeng_pulse_memory_sizes(void) {
@@ -1835,7 +1898,6 @@ static time_t get_tier_retention(struct rrdengine_instance *ctx)
1898
// Check if disk or retention time cap reached
1899
bool rrdeng_ctx_tier_cap_exceeded(struct rrdengine_instance *ctx)
1900
{
1838
-
1901
uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1902
if (!ctx->datafiles.first || !ctx->datafiles.first->next) {
1903
uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
@@ -1846,13 +1908,18 @@ bool rrdeng_ctx_tier_cap_exceeded(struct rrdengine_instance *ctx)
1908
1909
uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1910
1849
- time_t retention = get_tier_retention(ctx);
1850
-
1851
- if (ctx->config.max_retention_s && retention > ctx->config.max_retention_s)
1852
- return true;
1911
+ if (ctx->config.max_retention_s) {
1912
+ time_t retention = get_tier_retention(ctx);
1913
+ if (retention > ctx->config.max_retention_s) {
1914
+ __atomic_store_n(&ctx->datafiles.disk_time, false, __ATOMIC_RELAXED);
1915
+ return true;
1916
+ }
1917
+ }
1918
1854
- if (ctx->config.max_disk_space && estimated_disk_space > ctx->config.max_disk_space)
1919
+ if (ctx->config.max_disk_space && estimated_disk_space > ctx->config.max_disk_space) {
1920
+ __atomic_store_n(&ctx->datafiles.disk_time, true, __ATOMIC_RELAXED);
1921
return true;
1922
+ }
1923
1924
return false;
1925
}
@@ -1869,9 +1936,7 @@ static void retention_timer_cb(uv_timer_t *handle) {
1936
STORAGE_ENGINE *eng = localhost->db[tier].eng;
1937
if (!eng || eng->seb != STORAGE_ENGINE_BACKEND_DBENGINE)
1938
continue;
1872
- bool cleanup = rrdeng_ctx_tier_cap_exceeded(multidb_ctx[tier]);
1873
- if (cleanup)
1874
- rrdeng_enq_cmd(multidb_ctx[tier], RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1939
+ check_and_schedule_db_rotation(multidb_ctx[tier]);
1940
}
1941
1942
worker_is_idle();
@@ -2034,9 +2099,11 @@ void rrdeng_calculate_tier_disk_space_percentage(void)
2099
}
2100
}
2101
2037
-#define NOT_INDEXING_OR_DELETING_FILES(ctx) \
2038
- (!__atomic_load_n(&(ctx)->atomic.migration_to_v2_running, __ATOMIC_RELAXED) && \
2039
- !__atomic_load_n(&(ctx)->atomic.now_deleting_files, __ATOMIC_RELAXED))
2102
+#define NOT_DELETING_FILES(ctx) \
2103
+ (!__atomic_load_n(&(ctx)->atomic.now_deleting_files, __ATOMIC_RELAXED))
2104
+
2105
+#define NOT_INDEXING_FILES(ctx) \
2106
+ (!__atomic_load_n(&(ctx)->atomic.migration_to_v2_running, __ATOMIC_RELAXED))
2107
2108
void *dbengine_event_loop(void* arg) {
2109
sanity_check();
@@ -2181,19 +2248,19 @@ void *dbengine_event_loop(void* arg) {
2248
case RRDENG_OPCODE_JOURNAL_INDEX: {
2249
struct rrdengine_instance *ctx = cmd.ctx;
2250
struct rrdengine_datafile *datafile = cmd.data;
2184
- if (NOT_INDEXING_OR_DELETING_FILES(ctx) && ctx_is_available_for_queries(ctx)) {
2251
+ ctx->datafiles.pending_index = false;
2252
+ if (NOT_INDEXING_FILES(ctx) && ctx_is_available_for_queries(ctx)) {
2253
__atomic_store_n(&ctx->atomic.migration_to_v2_running, true, __ATOMIC_RELAXED);
2254
__atomic_store_n(&ctx->atomic.needs_indexing, false, __ATOMIC_RELAXED);
2255
work_dispatch(ctx, datafile, NULL, opcode, journal_v2_indexing_tp_worker, after_journal_v2_indexing);
2256
}
2189
- else
2190
- __atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
2257
break;
2258
}
2259
2260
case RRDENG_OPCODE_DATABASE_ROTATE: {
2261
struct rrdengine_instance *ctx = cmd.ctx;
2196
- if (NOT_INDEXING_OR_DELETING_FILES(ctx) && ctx->datafiles.first->next != NULL &&
2262
+ ctx->datafiles.pending_rotate = false;
2263
+ if (NOT_DELETING_FILES(ctx) && ctx->datafiles.first->next != NULL &&
2264
ctx->datafiles.first->next->next != NULL && rrdeng_ctx_tier_cap_exceeded(ctx)) {
2265
__atomic_store_n(&ctx->atomic.now_deleting_files, true, __ATOMIC_RELAXED);
2266
work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate);
src/database/engine/rrdengine.h
+20
-1
@@ -24,6 +24,17 @@
24
25
extern unsigned rrdeng_pages_per_extent;
26
27
+#define UNLINK_FILE(ctx, path, ret_var) \
28
+ do { \
29
+ uv_fs_t _req; \
30
+ (ret_var) = uv_fs_unlink(NULL, &(_req), (path), NULL); \
31
+ if ((ret_var) < 0) { \
32
+ netdata_log_error("DBENGINE: uv_fs_unlink(\"%s\"): %s", (path), uv_strerror(ret_var)); \
33
+ ctx_fs_error(ctx); \
34
+ } \
35
+ uv_fs_req_cleanup(&(_req)); \
36
+ } while (0)
37
+
38
/* Forward declarations */
39
struct rrdengine_instance;
40
struct rrdeng_cmd;
@@ -375,6 +386,9 @@ struct rrdengine_instance {
386
387
struct {
388
uv_rwlock_t rwlock; // the linked list of datafiles is protected by this lock
389
+ bool disk_time; // true: delete for disk quota, false: delete for retention
390
+ bool pending_rotate;
391
+ bool pending_index;
392
struct rrdengine_datafile *first; // oldest - the newest with ->first->prev
393
} datafiles;
394
@@ -535,7 +549,12 @@ static inline time_t max_acceptable_collected_time(void) {
549
return now_realtime_sec() + 1;
550
}
551
538
-void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool update_retention, bool worker);
552
+void datafile_delete(
553
+ struct rrdengine_instance *ctx,
554
+ struct rrdengine_datafile *datafile,
555
+ bool update_retention,
556
+ bool disk_time,
557
+ bool worker);
558
559
// --------------------------------------------------------------------------------------------------------------------
560
// the following functions are used to sort UUIDs in the journal files