@cryptotaxi247 / netdata-1 / commits / 02f18d13c

Improve datafile rotation and indexing during shutdown (#20464)

quiesce opcode only marks tier for shutdown Shutdown tiers (quiesce) before flushing dirty pages New opcode to flush all pages (hot and dirty) Journal indexing disabled during shutdown Allow quota check and datafile deletion during shutdown

Stelios Fragkakis committed Jun 11, 2025 at 20:45 UTC 02f18d13ce605a7b2caa4b0970fed69427a369d7

7 files changed +57 -29

src/daemon/daemon-shutdown.c

+14 -4

@@ -104,6 +104,11 @@ static void *rrdeng_exit_background(void *ptr) {

104 return NULL;

105 }

106

107 +static void rrdeng_quiesce_all()

108 +{

109 + for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)

110 + rrdeng_quiesce(multidb_ctx[tier]);

111 +}

112

113 static void rrdeng_flush_everything_and_wait(bool wait_flush, bool wait_collectors, bool dirty_only) {

114 static size_t starting_size_to_flush = 0;

@@ -112,8 +117,12 @@ static void rrdeng_flush_everything_and_wait(bool wait_flush, bool wait_collecto

117 return;

118

119 nd_log(NDLS_DAEMON, NDLP_INFO, "Flushing DBENGINE %s dirty pages...", dirty_only ? "only" : "hot &");

115 - for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)

116 - rrdeng_quiesce(multidb_ctx[tier], dirty_only);

120 + for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {

121 + if (dirty_only)

122 + rrdeng_flush_dirty(multidb_ctx[tier]);

123 + else

124 + rrdeng_flush_all(multidb_ctx[tier]);

125 + }

126

127 struct pgc_statistics pgc_main_stats = pgc_get_statistics(main_cache);

128 size_t size_to_flush = pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size;

@@ -187,9 +196,10 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

196 watcher_shutdown_begin();

197

198 #ifdef ENABLE_DBENGINE

190 - if(!abnormal && dbengine_enabled)

191 - // flush all dirty pages asap

199 + if(!abnormal && dbengine_enabled) {

200 + rrdeng_quiesce_all();

201 rrdeng_flush_everything_and_wait(false, false, true);

202 + }

203 #endif

204

205 // notify we are exiting

src/database/engine/dbengine-stresstest.c

+1 -1

@@ -447,7 +447,7 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi

447 }

448 freez(query_threads);

449 rrd_wrlock();

450 - rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si, false);

450 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

451 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

452 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

453 rrd_wrunlock();

src/database/engine/dbengine-unittest.c

+2 -1

@@ -411,7 +411,8 @@ int test_dbengine(void) {

411 sleep(5);

412

413 rrd_wrlock();

414 - rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si, false);

414 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

415 + rrdeng_flush_all((struct rrdengine_instance *)host->db[0].si);

416 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

417 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

418 rrd_wrunlock();

src/database/engine/rrdengine.c

+18 -15

@@ -182,7 +182,7 @@ static inline void check_and_schedule_db_rotation(struct rrdengine_instance *ctx

182 return;

183 }

184

185 - if(ctx_is_available_for_queries(ctx) && rrdeng_ctx_tier_cap_exceeded(ctx)) {

185 + if(rrdeng_ctx_tier_cap_exceeded(ctx)) {

186 ctx->datafiles.pending_rotate = true;

187 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

188 }

@@ -689,7 +689,7 @@ static bool datafile_is_full(struct rrdengine_instance *ctx, struct rrdengine_da

689 bool ret = false;

690 spinlock_lock(&datafile->writers.spinlock);

691

692 - if(ctx_is_available_for_queries(ctx) && datafile->pos > rrdeng_target_data_file_size(ctx))

692 + if(datafile->pos > rrdeng_target_data_file_size(ctx))

693 ret = true;

694

695 spinlock_unlock(&datafile->writers.spinlock);

@@ -1327,12 +1327,12 @@ void datafile_delete(

1327 if (update_retention)

1328 update_metrics_first_time_s(ctx, datafile, datafile->next, worker);

1329

1330 - if (!ctx_is_available_for_queries(ctx)) {

1331 - // agent is shutting down, we cannot continue

1332 - if(worker)

1333 - worker_is_idle();

1334 - return;

1335 - }

1330 +// if (!ctx_is_available_for_queries(ctx)) {

1331 +// // agent is shutting down, we cannot continue

1332 +// if(worker)

1333 +// worker_is_idle();

1334 +// return;

1335 +// }

1336

1337 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);

1338 netdata_log_info("DBENGINE: deleting data file \"%s/"

@@ -2086,12 +2086,8 @@ bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx __maybe_unused) {

2086 dbengine_initialize_structures();

2087

2088 int retries = 0;

2089 -// int create_uv_thread_rc = create_uv_thread(&rrdeng_main.thread, dbengine_event_loop, &rrdeng_main, &retries);

2089 rrdeng_main.thread = nd_thread_create("DBEV", NETDATA_THREAD_OPTION_DEFAULT, dbengine_event_loop, &rrdeng_main);

2090

2092 -// if (!rrdeng_main.thread)

2093 -// nd_log_daemon(NDLP_ERR, "Failed to create DBENGINE thread, error %s, after %d retries", uv_err_name(create_uv_thread_rc), retries);

2094 -

2091 fatal_assert(0 != rrdeng_main.thread);

2092

2093 if (retries)

@@ -2187,6 +2183,7 @@ void *dbengine_event_loop(void* arg) {

2183 worker_register_job_name(RRDENG_OPCODE_EVICT_MAIN, "evict init");

2184 worker_register_job_name(RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown");

2185 worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty");

2186 + worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY, "ctx flush all");

2187 worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce");

2188 worker_register_job_name(RRDENG_OPCODE_SHUTDOWN_EVLOOP, "dbengine shutdown");

2189

@@ -2347,13 +2344,19 @@ void *dbengine_event_loop(void* arg) {

2344 break;

2345 }

2346

2347 + case RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY: {

2348 + struct rrdengine_instance *ctx = cmd.ctx;

2349 + work_dispatch(ctx, NULL, NULL, opcode,

2350 + flush_all_hot_and_dirty_pages_of_section_tp_worker,

2351 + after_flush_all_hot_and_dirty_pages_of_section);

2352 + break;

2353 + }

2354 +

2355 case RRDENG_OPCODE_CTX_QUIESCE: {

2356 // a ctx will shutdown shortly

2357 struct rrdengine_instance *ctx = cmd.ctx;

2358 + nd_log_daemon(NDLP_INFO, "DBENGINE: Tier %d is shutting down — query processing disabled", ctx->config.tier);

2359 __atomic_store_n(&ctx->quiesce.enabled, true, __ATOMIC_RELEASE);

2354 - work_dispatch(ctx, NULL, NULL, opcode,

2355 - flush_all_hot_and_dirty_pages_of_section_tp_worker,

2356 - after_flush_all_hot_and_dirty_pages_of_section);

2360 break;

2361 }

2362

src/database/engine/rrdengine.h

@@ -269,6 +269,7 @@ enum rrdeng_opcode {

269 RRDENG_OPCODE_EVICT_EXTENT,

270 RRDENG_OPCODE_CTX_SHUTDOWN,

271 RRDENG_OPCODE_CTX_FLUSH_DIRTY,

272 + RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY,

273 RRDENG_OPCODE_CTX_QUIESCE,

274 RRDENG_OPCODE_CTX_POPULATE_MRG,

275 RRDENG_OPCODE_SHUTDOWN_EVLOOP,

src/database/engine/rrdengineapi.c

+17 -7

@@ -1280,18 +1280,28 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {

1280 return 0;

1281 }

1282

1283 -void rrdeng_quiesce(struct rrdengine_instance *ctx, bool dirty_only)

1283 +void rrdeng_flush_dirty(struct rrdengine_instance *ctx)

1284 {

1285 if (NULL == ctx)

1286 return;

1287

1288 - // FIXME - ktsaou - properly cleanup ctx

1289 - // 1. make sure all collectors are stopped

1288 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_FLUSH_DIRTY, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1289 +}

1290

1291 - if (dirty_only)

1292 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_FLUSH_DIRTY, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1293 - else

1294 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1291 +void rrdeng_flush_all(struct rrdengine_instance *ctx)

1292 +{

1293 + if (NULL == ctx)

1294 + return;

1295 +

1296 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1297 +}

1298 +

1299 +void rrdeng_quiesce(struct rrdengine_instance *ctx)

1300 +{

1301 + if (NULL == ctx)

1302 + return;

1303 +

1304 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1305 }

1306

1307 static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats)

src/database/engine/rrdengineapi.h

+4 -1

@@ -74,7 +74,10 @@ int rrdeng_init(

74 void rrdeng_readiness_wait(struct rrdengine_instance *ctx);

76 int rrdeng_exit(struct rrdengine_instance *ctx);

77 -void rrdeng_quiesce(struct rrdengine_instance *ctx, bool dirty_only);

77 +void rrdeng_quiesce(struct rrdengine_instance *ctx);

78 +void rrdeng_flush_dirty(struct rrdengine_instance *ctx);

79 +void rrdeng_flush_all(struct rrdengine_instance *ctx);

80 +

82 bool rrdeng_metric_retention_by_id(STORAGE_INSTANCE *si, UUIDMAP_ID id, time_t *first_entry_s, time_t *last_entry_s);

83 bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, nd_uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s);