@cryptotaxi247 / netdata-1 / commits / 7a21b9663

DBENGINE v2 - improvements part 9 (#14326)

* on shutdown stop data collection for all hosts instead of freeing their memory * print number of sql statements per metadata host scan * print timings with metadata checking * use dbengine API to figure out of a database is legacy * Recalculate retention after a datafile deletion * validate child timestamps during replication * main cache uses a lockless aral per partition, protected by the partition index lock * prevent ML crash * Revert "main cache uses a lockless aral per partition, protected by the partition index lock" This reverts commit 6afc01527dc5c66548b4bc8a1d63c026c3149358. * Log direct index and binary searches * distribute metrics more evenly across time * statistics about retention recalculation * fix crash * Reverse the binary search to calculate retention * more optimization on retention calculation * removed commented old code Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 26, 2023 at 00:55 UTC 7a21b966381022b9dbb15d4377fb09b82d1f6067
14 files changed +388 -167
collectors/plugins.d/pluginsd_parser.c
+3 -8
@@ -326,7 +326,7 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
326 {
327 const char *first_entry_txt = get_word(words, num_words, 1);
328 const char *last_entry_txt = get_word(words, num_words, 2);
329 - const char *world_time_txt = get_word(words, num_words, 3);
329 + const char *wall_clock_time_txt = get_word(words, num_words, 3);
330
331 RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_CHART_DEFINITION_END);
332 if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
@@ -336,12 +336,7 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
336
337 time_t first_entry_child = (first_entry_txt && *first_entry_txt) ? (time_t)str2ul(first_entry_txt) : 0;
338 time_t last_entry_child = (last_entry_txt && *last_entry_txt) ? (time_t)str2ul(last_entry_txt) : 0;
339 - time_t child_world_time = (world_time_txt && *world_time_txt) ? (time_t)str2ul(world_time_txt) : now_realtime_sec();
340 -
341 - if((first_entry_child != 0 || last_entry_child != 0) && (first_entry_child == 0 || last_entry_child == 0))
342 - error("PLUGINSD REPLAY ERROR: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_CHART_DEFINITION_END " with malformed timings (first time %ld, last time %ld, world time %ld).",
343 - rrdhost_hostname(host), rrdset_id(st),
344 - first_entry_child, last_entry_child, child_world_time);
339 + time_t child_wall_clock_time = (wall_clock_time_txt && *wall_clock_time_txt) ? (time_t)str2ul(wall_clock_time_txt) : now_realtime_sec();
340
341 bool ok = true;
342 if(!rrdset_flag_check(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS)) {
@@ -358,7 +353,7 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
353
354 PARSER *parser = ((PARSER_USER_OBJECT *)user)->parser;
355 ok = replicate_chart_request(send_to_plugin, parser, host, st,
361 - first_entry_child, last_entry_child, child_world_time,
356 + first_entry_child, last_entry_child, child_wall_clock_time,
357 0, 0);
358 }
359 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
daemon/main.c
+3 -2
@@ -418,9 +418,10 @@ void netdata_cleanup_and_exit(int ret) {
418 #endif
419
420 // free the database
421 - delta_shutdown_time("free rrdhost structures");
421 + delta_shutdown_time("stop collection for all hosts");
422
423 - rrdhost_free_all();
423 + // rrdhost_free_all();
424 + rrd_finalize_collection_for_all_hosts();
425
426 delta_shutdown_time("stop metasync threads");
427
database/engine/pagecache.h
+1
@@ -52,6 +52,7 @@ struct rrdeng_page_info {
52 struct pg_alignment {
53 uint32_t page_position;
54 uint32_t refcount;
55 + uint16_t initial_slots;
56 };
57
58 struct rrdeng_query_handle;
database/engine/rrdengine.c
+117 -59
@@ -1254,11 +1254,13 @@ static void after_database_rotate(struct rrdengine_instance *ctx __maybe_unused,
1254 struct uuid_first_time_s {
1255 uuid_t *uuid;
1256 time_t first_time_s;
1257 - time_t last_time_s;
1257 METRIC *metric;
1258 + size_t pages_found;
1259 + size_t df_matched;
1260 + size_t df_index_oldest;
1261 };
1262
1261 -static int journal_metric_uuid_compare(const void *key, const void *metric)
1263 +static int journal_metric_compare(const void *key, const void *metric)
1264 {
1265 return uuid_compare(*(uuid_t *) key, ((struct journal_metric_list *) metric)->uuid);
1266 }
@@ -1279,7 +1281,12 @@ struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struc
1281 return next_datafile;
1282 }
1283
1282 -void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, Pvoid_t metric_first_time_JudyL) {
1284 +void find_uuid_first_time(
1285 + struct rrdengine_instance *ctx,
1286 + struct rrdengine_datafile *datafile,
1287 + struct uuid_first_time_s *uuid_first_entry_list,
1288 + size_t count)
1289 +{
1290 // acquire the datafile to work with it
1291 uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1292 while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))
@@ -1289,8 +1296,10 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1296 if (unlikely(!datafile))
1297 return;
1298
1292 - unsigned v2_count = 0;
1299 unsigned journalfile_count = 0;
1300 + size_t binary_match = 0;
1301 + size_t not_matching_bsearches = 0;
1302 +
1303 while (datafile) {
1304 struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0);
1305 if (!j2_header) {
@@ -1299,55 +1308,114 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1308 }
1309
1310 time_t journal_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
1302 - size_t journal_metric_count = (size_t)j2_header->metric_count;
1311 struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
1312 + struct uuid_first_time_s *uuid_original_entry;
1313
1305 - Word_t index = 0;
1306 - bool first_then_next = true;
1307 - Pvoid_t *PValue;
1308 - while ((PValue = JudyLFirstThenNext(metric_first_time_JudyL, &index, &first_then_next))) {
1309 - struct uuid_first_time_s *uuid_first_t_entry = *PValue;
1314 + size_t journal_metric_count = j2_header->metric_count;
1315
1311 - struct journal_metric_list *uuid_entry = bsearch(uuid_first_t_entry->uuid,uuid_list,journal_metric_count,sizeof(*uuid_list), journal_metric_uuid_compare);
1316 + for (size_t index = 0; index < count; ++index) {
1317 + uuid_original_entry = &uuid_first_entry_list[index];
1318
1313 - if (unlikely(!uuid_entry))
1319 + // Check here if we should skip this
1320 + if (uuid_original_entry->df_matched > 3 || uuid_original_entry->pages_found > 5)
1321 continue;
1322
1316 - time_t first_time_s = uuid_entry->delta_start_s + journal_start_time_s;
1317 - time_t last_time_s = uuid_entry->delta_end_s + journal_start_time_s;
1318 - uuid_first_t_entry->first_time_s = MIN(uuid_first_t_entry->first_time_s , first_time_s);
1319 - uuid_first_t_entry->last_time_s = MAX(uuid_first_t_entry->last_time_s , last_time_s);
1320 - v2_count++;
1323 + struct journal_metric_list *live_entry = bsearch(uuid_original_entry->uuid,uuid_list,journal_metric_count,sizeof(*uuid_list), journal_metric_compare);
1324 + if (!live_entry) {
1325 + // Not found in this journal
1326 + not_matching_bsearches++;
1327 + continue;
1328 + }
1329 +
1330 + uuid_original_entry->pages_found += live_entry->entries;
1331 + uuid_original_entry->df_matched++;
1332 +
1333 + time_t old_first_time_s = uuid_original_entry->first_time_s;
1334 +
1335 + // Calculate first / last for this match
1336 + time_t first_time_s = live_entry->delta_start_s + journal_start_time_s;
1337 + uuid_original_entry->first_time_s = MIN(uuid_original_entry->first_time_s, first_time_s);
1338 +
1339 + if (uuid_original_entry->first_time_s != old_first_time_s)
1340 + uuid_original_entry->df_index_oldest = uuid_original_entry->df_matched;
1341 +
1342 + binary_match++;
1343 }
1344 +
1345 journalfile_count++;
1346 journalfile_v2_data_release(datafile->journalfile);
1347 datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1348 }
1349
1350 // Let's scan the open cache for almost exact match
1328 - bool first_then_next = true;
1329 - Pvoid_t *PValue;
1330 - Word_t index = 0;
1331 - unsigned open_cache_count = 0;
1332 - while ((PValue = JudyLFirstThenNext(metric_first_time_JudyL, &index, &first_then_next))) {
1333 - struct uuid_first_time_s *uuid_first_t_entry = *PValue;
1351 + size_t open_cache_count = 0;
1352 +
1353 + size_t df_index[10] = { 0 };
1354 + size_t without_metric = 0;
1355 + size_t open_cache_gave_first_time_s = 0;
1356 + size_t metric_count = 0;
1357 + size_t without_retention = 0;
1358 + size_t not_needed_bsearches = 0;
1359 +
1360 + for (size_t index = 0; index < count; ++index) {
1361 + struct uuid_first_time_s *uuid_first_t_entry = &uuid_first_entry_list[index];
1362 +
1363 + metric_count++;
1364 +
1365 + size_t idx = uuid_first_t_entry->df_index_oldest;
1366 + if(idx >= 10)
1367 + idx = 9;
1368 +
1369 + df_index[idx]++;
1370 +
1371 + not_needed_bsearches += uuid_first_t_entry->df_matched - uuid_first_t_entry->df_index_oldest;
1372 +
1373 + if (unlikely(!uuid_first_t_entry->metric)) {
1374 + without_metric++;
1375 + continue;
1376 + }
1377
1378 PGC_PAGE *page = pgc_page_get_and_acquire(
1379 open_cache, (Word_t)ctx,
1337 - (Word_t)uuid_first_t_entry->metric, uuid_first_t_entry->last_time_s,
1338 - PGC_SEARCH_CLOSEST);
1380 + (Word_t)uuid_first_t_entry->metric, 0,
1381 + PGC_SEARCH_FIRST);
1382
1383 if (page) {
1384 + time_t old_first_time_s = uuid_first_t_entry->first_time_s;
1385 +
1386 time_t first_time_s = pgc_page_start_time_s(page);
1342 - time_t last_time_s = pgc_page_end_time_s(page);
1387 uuid_first_t_entry->first_time_s = MIN(uuid_first_t_entry->first_time_s, first_time_s);
1344 - uuid_first_t_entry->last_time_s = MAX(uuid_first_t_entry->last_time_s, last_time_s);
1388 pgc_page_release(open_cache, page);
1389 open_cache_count++;
1390 +
1391 + if(uuid_first_t_entry->first_time_s != old_first_time_s) {
1392 + open_cache_gave_first_time_s++;
1393 + }
1394 + }
1395 + else {
1396 + if(!uuid_first_t_entry->df_index_oldest)
1397 + without_retention++;
1398 }
1399 }
1349 - info("DBENGINE: processed %u journalfiles and matched %u metric pages in v2 files and %u in open cache", journalfile_count,
1350 - v2_count, open_cache_count);
1400 + internal_error(true,
1401 + "DBENGINE: analyzed the retention of %zu rotated metrics, "
1402 + "did %zu jv2 matching binary searches (%zu not matching, %zu overflown) in %u journal files, "
1403 + "%zu metrics with entries in open cache, "
1404 + "metrics first time found per datafile index ([not in jv2]:%zu, [1]:%zu, [2]:%zu, [3]:%zu, [4]:%zu, [5]:%zu, [6]:%zu, [7]:%zu, [8]:%zu, [bigger]: %zu), "
1405 + "open cache found first time %zu, "
1406 + "metrics without any remaining retention %zu, "
1407 + "metrics not in MRG %zu",
1408 + metric_count,
1409 + binary_match,
1410 + not_matching_bsearches,
1411 + not_needed_bsearches,
1412 + journalfile_count,
1413 + open_cache_count,
1414 + df_index[0], df_index[1], df_index[2], df_index[3], df_index[4], df_index[5], df_index[6], df_index[7], df_index[8], df_index[9],
1415 + open_cache_gave_first_time_s,
1416 + without_retention,
1417 + without_metric
1418 + );
1419 }
1420
1421 static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile_to_delete, struct rrdengine_datafile *first_datafile_remaining, bool worker) {
@@ -1360,59 +1428,49 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1428 struct journal_v2_header *j2_header = journalfile_v2_data_acquire(journalfile, NULL, 0, 0);
1429 struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
1430
1363 - Pvoid_t metric_first_time_JudyL = (Pvoid_t) NULL;
1364 - Pvoid_t *PValue;
1365 -
1366 - unsigned count = 0;
1431 + size_t count = j2_header->metric_count;
1432 struct uuid_first_time_s *uuid_first_t_entry;
1368 - for (uint32_t index = 0; index < j2_header->metric_count; ++index) {
1433 + struct uuid_first_time_s *uuid_first_entry_list = callocz(count, sizeof(struct uuid_first_time_s));
1434 +
1435 + size_t added = 0;
1436 + for (size_t index = 0; index < count; ++index) {
1437 METRIC *metric = mrg_metric_get_and_acquire(main_mrg, &uuid_list[index].uuid, (Word_t) ctx);
1438 if (!metric)
1439 continue;
1440
1373 - PValue = JudyLIns(&metric_first_time_JudyL, (Word_t) index, PJE0);
1374 - fatal_assert(NULL != PValue);
1375 - if (!*PValue) {
1376 - uuid_first_t_entry = mallocz(sizeof(*uuid_first_t_entry));
1377 - uuid_first_t_entry->metric = metric;
1378 - uuid_first_t_entry->first_time_s = LONG_MAX;
1379 - uuid_first_t_entry->last_time_s = 0;
1380 - uuid_first_t_entry->uuid = mrg_metric_uuid(main_mrg, metric);
1381 - *PValue = uuid_first_t_entry;
1382 - count++;
1383 - }
1441 + uuid_first_entry_list[added].metric = metric;
1442 + uuid_first_entry_list[added].first_time_s = LONG_MAX;
1443 + uuid_first_entry_list[added].df_matched = 0;
1444 + uuid_first_entry_list[added].df_index_oldest = 0;
1445 + uuid_first_entry_list[added].uuid = mrg_metric_uuid(main_mrg, metric);
1446 + added++;
1447 }
1385 - journalfile_v2_data_release(journalfile);
1448
1387 - info("DBENGINE: recalculating retention for %u metrics starting with datafile %u", count, first_datafile_remaining->fileno);
1449 + info("DBENGINE: recalculating retention for %zu metrics starting with datafile %u", count, first_datafile_remaining->fileno);
1450 +
1451 + journalfile_v2_data_release(journalfile);
1452
1453 // Update the first time / last time for all metrics we plan to delete
1454
1455 if(worker)
1456 worker_is_busy(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION);
1457
1394 - find_uuid_first_time(ctx, first_datafile_remaining, metric_first_time_JudyL);
1458 + find_uuid_first_time(ctx, first_datafile_remaining, uuid_first_entry_list, added);
1459
1460 if(worker)
1461 worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1462
1399 - info("DBENGINE: updating metric registry retention for %u metrics", count);
1400 -
1401 - Word_t index = 0;
1402 - bool first_then_next = true;
1403 - while ((PValue = JudyLFirstThenNext(metric_first_time_JudyL, &index, &first_then_next))) {
1404 - uuid_first_t_entry = *PValue;
1463 + info("DBENGINE: updating metric registry retention for %zu metrics", added);
1464
1406 - if (likely(uuid_first_t_entry->first_time_s != LONG_MAX && uuid_first_t_entry->last_time_s))
1465 + for (size_t index = 0; index < added; ++index) {
1466 + uuid_first_t_entry = &uuid_first_entry_list[index];
1467 + if (likely(uuid_first_t_entry->first_time_s != LONG_MAX))
1468 mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
1469 else
1470 mrg_metric_set_first_time_s(main_mrg, uuid_first_t_entry->metric, 0);
1410 -
1471 mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1412 - freez(uuid_first_t_entry);
1472 }
1414 -
1415 - JudyLFreeArray(&metric_first_time_JudyL, PJE0);
1473 + freez(uuid_first_entry_list);
1474
1475 if(worker)
1476 worker_is_idle();
database/engine/rrdengineapi.c
+49 -19
@@ -412,7 +412,7 @@ static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *ha
412 check_and_fix_mrg_update_every(handle);
413 }
414
415 -static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle, size_t *data_size) {
415 +static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle, size_t *data_size, usec_t point_in_time_ut) {
416 struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric);
417 size_t size;
418
@@ -421,28 +421,45 @@ static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle,
421 size = tier_page_size[ctx->config.tier];
422 }
423 else {
424 + size_t final_slots = 0;
425 +
426 // the first page
427 handle->options |= RRDENG_FIRST_PAGE_ALLOCATED;
428 size_t max_size = tier_page_size[ctx->config.tier];
429 size_t max_slots = max_size / CTX_POINT_SIZE_BYTES(ctx);
428 - size_t min_slots = max_slots / 5;
429 - size_t distribution = max_slots - min_slots;
430 - size_t this_page_end_slot = indexing_partition((Word_t)handle->alignment, distribution);
430
432 - size_t current_end_slot = (size_t)now_monotonic_sec() % distribution;
431 + if(handle->alignment->initial_slots) {
432 + final_slots = handle->alignment->initial_slots;
433 + }
434 + else {
435 + max_slots -= 3;
436 +
437 + size_t smaller_slot = indexing_partition((Word_t)handle->alignment, max_slots);
438 + final_slots = smaller_slot;
439
434 - if(current_end_slot < this_page_end_slot)
435 - this_page_end_slot -= current_end_slot;
436 - else if(current_end_slot > this_page_end_slot)
437 - this_page_end_slot = (max_slots - current_end_slot) + this_page_end_slot;
440 + time_t now_s = (time_t)(point_in_time_ut / USEC_PER_SEC);
441 + size_t current_pos = (now_s % max_slots);
442
439 - size_t final_slots = min_slots + this_page_end_slot;
443 + if(current_pos > final_slots)
444 + final_slots += max_slots - current_pos;
445
441 - if(final_slots > max_slots)
442 - final_slots = max_slots;
446 + else if(current_pos < final_slots)
447 + final_slots -= current_pos;
448
444 - if(final_slots < min_slots)
445 - final_slots = min_slots;
449 + if(final_slots < 3) {
450 + final_slots += 3;
451 + smaller_slot += 3;
452 +
453 + if(smaller_slot >= max_slots)
454 + smaller_slot -= max_slots;
455 + }
456 +
457 + max_slots += 3;
458 + handle->alignment->initial_slots = smaller_slot + 3;
459 +
460 + internal_fatal(handle->alignment->initial_slots < 3 || handle->alignment->initial_slots >= max_slots, "ooops! wrong distribution of metrics across time");
461 + internal_fatal(final_slots < 3 || final_slots >= max_slots, "ooops! wrong distribution of metrics across time");
462 + }
463
464 size = final_slots * CTX_POINT_SIZE_BYTES(ctx);
465 }
@@ -485,7 +502,7 @@ static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_
502 handle->page_flags |= RRDENG_PAGE_UNALIGNED;
503 rrdeng_store_metric_flush_current_page(collection_handle);
504
488 - data = rrdeng_alloc_new_metric_data(handle, &data_size);
505 + data = rrdeng_alloc_new_metric_data(handle, &data_size, point_in_time_ut);
506 }
507 else {
508 data = pgc_page_data(handle->page);
@@ -493,7 +510,7 @@ static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_
510 }
511 }
512 else
496 - data = rrdeng_alloc_new_metric_data(handle, &data_size);
513 + data = rrdeng_alloc_new_metric_data(handle, &data_size, point_in_time_ut);
514
515 switch (ctx->config.page_type) {
516 case PAGE_METRICS: {
@@ -682,9 +699,15 @@ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) {
699 if((handle->options & RRDENG_1ST_METRIC_WRITER) && !mrg_metric_writer_release(main_mrg, handle->metric))
700 internal_fatal(true, "DBENGINE: metric is already released");
701
702 + time_t first_time_s = mrg_metric_get_first_time_s(main_mrg, handle->metric);
703 + time_t last_time_s = mrg_metric_get_latest_time_s(main_mrg, handle->metric);
704 +
705 mrg_metric_release(main_mrg, handle->metric);
706 freez(handle);
707
708 + if(!first_time_s && !last_time_s)
709 + return 1;
710 +
711 return 0;
712 }
713
@@ -1088,6 +1111,11 @@ void rrdeng_readiness_wait(struct rrdengine_instance *ctx) {
1111 info("DBENGINE: tier %d is ready for data collection and queries", ctx->config.tier);
1112 }
1113
1114 +bool rrdeng_is_legacy(STORAGE_INSTANCE *db_instance) {
1115 + struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
1116 + return ctx->config.legacy;
1117 +}
1118 +
1119 void rrdeng_exit_mode(struct rrdengine_instance *ctx) {
1120 __atomic_store_n(&ctx->quiesce.exit_mode, true, __ATOMIC_RELAXED);
1121 }
@@ -1149,7 +1177,7 @@ int rrdeng_init(struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned p
1177 finalize_rrd_files(ctx);
1178 }
1179
1152 - if (!is_storage_engine_shared((STORAGE_INSTANCE *)ctx)) {
1180 + if (ctx->config.legacy) {
1181 freez(ctx);
1182 if (ctxp)
1183 *ctxp = NULL;
@@ -1179,14 +1207,16 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {
1207 bool logged = false;
1208 while(__atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED) && !unittest_running) {
1209 if(!logged) {
1182 - info("Waiting for collectors to finish on tier %d...", ctx->config.tier);
1210 + info("DBENGINE: waiting for collectors to finish on tier %d...", (ctx->config.legacy) ? -1 : ctx->config.tier);
1211 logged = true;
1212 }
1213 sleep_usec(100 * USEC_PER_MS);
1214 }
1215
1216 + info("DBENGINE: flushing main cache for tier %d", (ctx->config.legacy) ? -1 : ctx->config.tier);
1217 pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
1218
1219 + info("DBENGINE: shutting down tier %d", (ctx->config.legacy) ? -1 : ctx->config.tier);
1220 struct completion completion = {};
1221 completion_init(&completion);
1222 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);
@@ -1195,7 +1225,7 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {
1225
1226 finalize_rrd_files(ctx);
1227
1198 - if(!is_storage_engine_shared((STORAGE_INSTANCE *)ctx))
1228 + if(ctx->config.legacy)
1229 freez(ctx);
1230
1231 rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
database/engine/rrdengineapi.h
+1
@@ -225,5 +225,6 @@ struct rrdeng_cache_efficiency_stats rrdeng_get_cache_efficiency_stats(void);
225
226 RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx);
227 size_t rrdeng_collectors_running(struct rrdengine_instance *ctx);
228 +bool rrdeng_is_legacy(STORAGE_INSTANCE *db_instance);
229
230 #endif /* NETDATA_RRDENGINEAPI_H */
database/rrd.h
+7
@@ -571,6 +571,8 @@ typedef enum __attribute__ ((__packed__)) rrdset_flags {
571 RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED = (1 << 25), // the receiving side has completed replication
572
573 RRDSET_FLAG_UPSTREAM_SEND_VARIABLES = (1 << 26), // a custom variable has been updated and needs to be exposed to parent
574 +
575 + RRDSET_FLAG_COLLECTION_FINISHED = (1 << 27), // when set, data collection is not available for this chart
576 } RRDSET_FLAGS;
577
578 #define rrdset_flag_check(st, flag) (__atomic_load_n(&((st)->flags), __ATOMIC_SEQ_CST) & (flag))
@@ -1310,6 +1312,11 @@ collected_number rrddim_timed_set_by_pointer(RRDSET *st, RRDDIM *rd, struct time
1312 collected_number rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_number value);
1313 collected_number rrddim_set(RRDSET *st, const char *id, collected_number value);
1314
1315 +bool rrddim_finalize_collection_and_check_retention(RRDDIM *rd);
1316 +void rrdset_finalize_collection(RRDSET *st, bool dimensions_too);
1317 +void rrdhost_finalize_collection(RRDHOST *host);
1318 +void rrd_finalize_collection_for_all_hosts(void);
1319 +
1320 long align_entries_to_pagesize(RRD_MEMORY_MODE mode, long entries);
1321
1322 #ifdef NETDATA_LOG_COLLECTION_ERRORS
database/rrddim.c
+20 -13
@@ -166,6 +166,25 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
166
167 }
168
169 +bool rrddim_finalize_collection_and_check_retention(RRDDIM *rd) {
170 + size_t tiers_available = 0, tiers_said_no_retention = 0;
171 +
172 + for(size_t tier = 0; tier < storage_tiers ;tier++) {
173 + if(!rd->tiers[tier].db_collection_handle)
174 + continue;
175 +
176 + tiers_available++;
177 +
178 + if(rd->tiers[tier].collect_ops->finalize(rd->tiers[tier].db_collection_handle))
179 + tiers_said_no_retention++;
180 +
181 + rd->tiers[tier].db_collection_handle = NULL;
182 + }
183 +
184 + // return true if the dimension has retention in the db
185 + return (!tiers_said_no_retention || tiers_available > tiers_said_no_retention);
186 +}
187 +
188 static void rrddim_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrddim, void *rrdset) {
189 RRDDIM *rd = rrddim;
190 RRDSET *st = rrdset;
@@ -180,19 +199,7 @@ static void rrddim_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, v
199
200 debug(D_RRD_CALLS, "rrddim_free() %s.%s", rrdset_name(st), rrddim_name(rd));
201
183 - size_t tiers_available = 0, tiers_said_no_retention = 0;
184 - for(size_t tier = 0; tier < storage_tiers ;tier++) {
185 - if(rd->tiers[tier].db_collection_handle) {
186 - tiers_available++;
187 -
188 - if(rd->tiers[tier].collect_ops->finalize(rd->tiers[tier].db_collection_handle))
189 - tiers_said_no_retention++;
190 -
191 - rd->tiers[tier].db_collection_handle = NULL;
192 - }
193 - }
194 -
195 - if (tiers_available == tiers_said_no_retention && tiers_said_no_retention && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
202 + if (!rrddim_finalize_collection_and_check_retention(rd) && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
203 /* This metric has no data and no references */
204 metaqueue_delete_dimension_uuid(&rd->metric_uuid);
205 }
database/rrdhost.c
+20 -4
@@ -33,10 +33,8 @@ time_t rrdhost_free_orphan_time_s = 3600;
33
34 bool is_storage_engine_shared(STORAGE_INSTANCE *engine __maybe_unused) {
35 #ifdef ENABLE_DBENGINE
36 - for(size_t tier = 0; tier < storage_tiers ;tier++) {
37 - if (engine == (STORAGE_INSTANCE *)multidb_ctx[tier])
38 - return true;
39 - }
36 + if(!rrdeng_is_legacy(engine))
37 + return true;
38 #endif
39
40 return false;
@@ -1223,6 +1221,15 @@ void rrdhost_free_all(void) {
1221 rrd_unlock();
1222 }
1223
1224 +void rrd_finalize_collection_for_all_hosts(void) {
1225 + RRDHOST *host;
1226 + rrd_wrlock();
1227 + rrdhost_foreach_read(host) {
1228 + rrdhost_finalize_collection(host);
1229 + }
1230 + rrd_unlock();
1231 +}
1232 +
1233 // ----------------------------------------------------------------------------
1234 // RRDHOST - save host files
1235
@@ -1391,6 +1398,15 @@ void reload_host_labels(void) {
1398 rrdpush_send_host_labels(localhost);
1399 }
1400
1401 +void rrdhost_finalize_collection(RRDHOST *host) {
1402 + info("Stopping data collection for host '%s'...", rrdhost_hostname(host));
1403 +
1404 + RRDSET *st;
1405 + rrdset_foreach_write(st, host)
1406 + rrdset_finalize_collection(st, true);
1407 + rrdset_foreach_done(st);
1408 +}
1409 +
1410 // ----------------------------------------------------------------------------
1411 // RRDHOST - delete host files
1412
database/rrdset.c
+35 -15
@@ -185,6 +185,29 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
185 ml_chart_new(st);
186 }
187
188 +void rrdset_finalize_collection(RRDSET *st, bool dimensions_too) {
189 + RRDHOST *host = st->rrdhost;
190 +
191 + rrdset_flag_set(st, RRDSET_FLAG_COLLECTION_FINISHED);
192 +
193 + if(dimensions_too) {
194 + RRDDIM *rd;
195 + rrddim_foreach_read(rd, st)
196 + rrddim_finalize_collection_and_check_retention(rd);
197 + rrddim_foreach_done(rd);
198 + }
199 +
200 + for(size_t tier = 0; tier < storage_tiers ; tier++) {
201 + STORAGE_ENGINE *eng = st->rrdhost->db[tier].eng;
202 + if(!eng) continue;
203 +
204 + if(st->storage_metrics_groups[tier]) {
205 + eng->api.collect_ops.metrics_group_release(host->db[tier].instance, st->storage_metrics_groups[tier]);
206 + st->storage_metrics_groups[tier] = NULL;
207 + }
208 + }
209 +}
210 +
211 // the destructor - the dictionary is write locked while this runs
212 static void rrdset_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *rrdset, void *rrdhost) {
213 RRDHOST *host = rrdhost;
@@ -192,15 +215,7 @@ static void rrdset_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, v
215
216 rrdset_flag_clear(st, RRDSET_FLAG_INDEXED_ID);
217
195 - // cleanup storage engines
196 - {
197 - for(size_t tier = 0; tier < storage_tiers ; tier++) {
198 - STORAGE_ENGINE *eng = st->rrdhost->db[tier].eng;
199 - if(!eng) continue;
200 -
201 - eng->api.collect_ops.metrics_group_release(host->db[tier].instance, st->storage_metrics_groups[tier]);
202 - }
203 - }
218 + rrdset_finalize_collection(st, false);
219
220 // remove it from the name index
221 rrdset_index_del_name(host, st);
@@ -600,16 +615,17 @@ void rrdset_get_retention_of_tier_for_collected_chart(RRDSET *st, time_t *first_
615 }
616
617 if(unlikely(db_first_entry_s && db_last_entry_s && db_first_entry_s >= db_last_entry_s)) {
603 - internal_error(true,
604 - "RRDSET: 'host:%s/chart:%s' oldest db time %ld is equal or bigger than latest db time %ld, adjusting it last updated time - update every",
618 + internal_error(db_first_entry_s > db_last_entry_s,
619 + "RRDSET: 'host:%s/chart:%s' oldest db time %ld is bigger than latest db time %ld, adjusting it to (latest time %ld - update every %ld)",
620 rrdhost_hostname(st->rrdhost), rrdset_id(st),
606 - db_first_entry_s, db_last_entry_s);
621 + db_first_entry_s, db_last_entry_s,
622 + db_last_entry_s, (time_t)st->update_every);
623 db_first_entry_s = db_last_entry_s - st->update_every;
624 }
625
626 if(unlikely(!db_first_entry_s && db_last_entry_s))
627 // this can be the case on the first data collection of a chart
612 - db_first_entry_s = db_last_entry_s;
628 + db_first_entry_s = db_last_entry_s - st->update_every;
629
630 *first_time_s = db_first_entry_s;
631 *last_time_s = db_last_entry_s;
@@ -1467,9 +1483,13 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
1483 next_store_ut = 0, // the timestamp in microseconds, of the next entry to store in the db
1484 update_every_ut = st->update_every * USEC_PER_SEC; // st->update_every in microseconds
1485
1486 + RRDSET_FLAGS rrdset_flags = rrdset_flag_check(st, ~0);
1487 + if(unlikely(rrdset_flags & RRDSET_FLAG_COLLECTION_FINISHED))
1488 + return;
1489 +
1490 netdata_thread_disable_cancelability();
1491
1472 - if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE))) {
1492 + if (unlikely(rrdset_flags & RRDSET_FLAG_OBSOLETE)) {
1493 error("Chart '%s' has the OBSOLETE flag set, but it is collected.", rrdset_id(st));
1494 rrdset_isnot_obsolete(st);
1495 }
@@ -1554,7 +1574,7 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
1574 last_stored_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
1575 next_store_ut = (st->last_updated.tv_sec + st->update_every) * USEC_PER_SEC;
1576
1557 - if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST))) {
1577 + if(unlikely(rrdset_flags & RRDSET_FLAG_STORE_FIRST)) {
1578 store_this_entry = 1;
1579 last_collect_ut = next_store_ut - update_every_ut;
1580
database/sqlite/sqlite_metadata.c
+35 -7
@@ -881,7 +881,7 @@ static void after_metadata_hosts(uv_work_t *req, int status __maybe_unused)
881 freez(data);
882 }
883
884 -static bool metadata_scan_host(RRDHOST *host, uint32_t max_count) {
884 +static bool metadata_scan_host(RRDHOST *host, uint32_t max_count, size_t *query_counter) {
885 RRDSET *st;
886 int rc;
887
@@ -895,6 +895,8 @@ static bool metadata_scan_host(RRDHOST *host, uint32_t max_count) {
895 break;
896 }
897 if(rrdset_flag_check(st, RRDSET_FLAG_METADATA_UPDATE)) {
898 + (*query_counter)++;
899 +
900 rrdset_flag_clear(st, RRDSET_FLAG_METADATA_UPDATE);
901 scan_count++;
902
@@ -924,6 +926,8 @@ static bool metadata_scan_host(RRDHOST *host, uint32_t max_count) {
926 RRDDIM *rd;
927 rrddim_foreach_read(rd, st) {
928 if(rrddim_flag_check(rd, RRDDIM_FLAG_METADATA_UPDATE)) {
929 + (*query_counter)++;
930 +
931 rrddim_flag_clear(rd, RRDDIM_FLAG_METADATA_UPDATE);
932
933 if (rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN))
@@ -963,12 +967,18 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)
967 struct scan_metadata_payload *data = req->data;
968 struct metadata_wc *wc = data->wc;
969
970 + usec_t all_started_ut = now_monotonic_usec(); (void)all_started_ut;
971 + internal_error(true, "METADATA: checking all hosts...");
972 +
973 bool run_again = false;
974 worker_is_busy(UV_EVENT_METADATA_STORE);
975 dfe_start_reentrant(rrdhost_root_index, host) {
976 if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) || !rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_UPDATE))
977 continue;
971 - internal_error(true, "METADATA: Scanning host %s", rrdhost_hostname(host));
978 +
979 + size_t query_counter = 0; (void)query_counter;
980 + usec_t started_ut = now_monotonic_usec(); (void)started_ut;
981 +
982 rrdhost_flag_clear(host,RRDHOST_FLAG_METADATA_UPDATE);
983
984 if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_LABELS))) {
@@ -981,38 +991,56 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)
991 rrdlabels_walkthrough_read(host->rrdlabels, host_label_store_to_sql_callback, &tmp);
992 db_execute(buffer_tostring(work_buffer));
993 buffer_free(work_buffer);
994 + query_counter++;
995 }
996 }
997
998 if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_CLAIMID))) {
999 rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_CLAIMID);
1000 uuid_t uuid;
1001 +
1002 if (likely(host->aclk_state.claimed_id && !uuid_parse(host->aclk_state.claimed_id, uuid)))
1003 store_claim_id(&host->host_uuid, &uuid);
1004 else
1005 store_claim_id(&host->host_uuid, NULL);
1006 +
1007 + query_counter++;
1008 }
1009
1010 if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_INFO))) {
1011 rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_INFO);
1012
1013 BUFFER *work_buffer = sql_store_host_system_info(host);
1000 - db_execute(buffer_tostring(work_buffer));
1001 - buffer_free(work_buffer);
1014 + if(work_buffer) {
1015 + db_execute(buffer_tostring(work_buffer));
1016 + buffer_free(work_buffer);
1017 + query_counter++;
1018 + }
1019
1020 int rc = sql_store_host_info(host);
1021 if (unlikely(rc))
1005 - error_report("Failed to store host info in the database for %s", string2str(host->hostname));
1022 + error_report("METADATA: 'host:%s': failed to store host info", string2str(host->hostname));
1023 + else
1024 + query_counter++;
1025 }
1026
1008 - if (unlikely(metadata_scan_host(host, data->max_count))) {
1027 + if (unlikely(metadata_scan_host(host, data->max_count, &query_counter))) {
1028 run_again = true;
1029 rrdhost_flag_set(host,RRDHOST_FLAG_METADATA_UPDATE);
1011 - internal_error(true,"METADATA: Rescheduling host %s to run; more charts to store", rrdhost_hostname(host));
1030 + internal_error(true,"METADATA: 'host:%s': scheduling another run, more charts to store", rrdhost_hostname(host));
1031 }
1032 +
1033 + usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
1034 + internal_error(true, "METADATA: 'host:%s': saved metadata with %zu SQL statements, in %0.2f ms",
1035 + rrdhost_hostname(host), query_counter,
1036 + (double)(ended_ut - started_ut) / USEC_PER_MS);
1037 }
1038 dfe_done(host);
1039
1040 + usec_t all_ended_ut = now_monotonic_usec(); (void)all_ended_ut;
1041 + internal_error(true, "METADATA: checking all hosts completed in %0.2f ms",
1042 + (double)(all_ended_ut - all_started_ut) / USEC_PER_MS);
1043 +
1044 if (unlikely(run_again))
1045 wc->check_hosts_after = now_realtime_sec() + METADATA_HOST_CHECK_IMMEDIATE;
1046 else
ml/Host.cc
+1 -1
@@ -303,7 +303,7 @@ void Host::detect() {
303
304 while (service_running((SERVICE_TYPE)(SERVICE_ML_PREDICTION | SERVICE_COLLECTORS))) {
305 worker_is_idle();
306 - heartbeat_next(&HB, RH->rrd_update_every * USEC_PER_SEC);
306 + heartbeat_next(&HB, (RH ? RH->rrd_update_every : default_rrd_update_every) * USEC_PER_SEC);
307 detectOnce();
308 }
309 }
streaming/replication.c
+95 -38
@@ -434,13 +434,31 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
434 }
435
436 static struct replication_query *replication_response_prepare(RRDSET *st, bool requested_enable_streaming, time_t requested_after, time_t requested_before) {
437 + time_t wall_clock_time = now_realtime_sec();
438 +
439 + if(requested_after > requested_before) {
440 + // flip them
441 + time_t t = requested_before;
442 + requested_before = requested_after;
443 + requested_after = t;
444 + }
445 +
446 + if(requested_after > wall_clock_time) {
447 + requested_after = 0;
448 + requested_before = 0;
449 + requested_enable_streaming = true;
450 + }
451 +
452 + if(requested_before > wall_clock_time) {
453 + requested_before = wall_clock_time;
454 + requested_enable_streaming = true;
455 + }
456 +
457 time_t query_after = requested_after;
458 time_t query_before = requested_before;
459 bool query_enable_streaming = requested_enable_streaming;
460
441 - time_t wall_clock_time = now_realtime_sec();
442 -
443 - time_t db_first_entry, db_last_entry;
461 + time_t db_first_entry = 0, db_last_entry = 0;
462 rrdset_get_retention_of_tier_for_collected_chart(st, &db_first_entry, &db_last_entry, wall_clock_time, 0);
463
464 if(requested_after == 0 && requested_before == 0 && requested_enable_streaming == true) {
@@ -580,14 +598,14 @@ struct replication_request_details {
598 struct {
599 time_t first_entry_t; // the first entry time the child has
600 time_t last_entry_t; // the last entry time the child has
583 - time_t world_time_t; // the current time of the child
601 + time_t wall_clock_time; // the current time of the child
602 + bool fixed_last_entry; // when set we set the last entry to wall clock time
603 } child_db;
604
605 struct {
606 time_t first_entry_t; // the first entry time we have
607 time_t last_entry_t; // the last entry time we have
589 - bool last_entry_t_adjusted_to_now; // true, if the last entry time was in the future, and we fixed
590 - time_t now; // the current local world clock time
608 + time_t wall_clock_time; // the current local world clock time
609 } local_db;
610
611 struct {
@@ -607,9 +625,36 @@ struct replication_request_details {
625 } wanted;
626 };
627
610 -static bool send_replay_chart_cmd(struct replication_request_details *r, const char *msg __maybe_unused) {
628 +static void replicate_log_request(struct replication_request_details *r, const char *msg) {
629 +#ifdef NETDATA_INTERNAL_CHECKS
630 + internal_error(true,
631 +#else
632 + error_limit_static_global_var(erl, 1, 0);
633 + error_limit(&erl,
634 +#endif
635 + "REPLAY ERROR: 'host:%s/chart:%s' child sent: "
636 + "db from %ld to %ld%s, wall clock time %ld, "
637 + "last request from %ld to %ld, "
638 + "issue: %s - "
639 + "sending replication request from %ld to %ld, start streaming %s",
640 + rrdhost_hostname(r->st->rrdhost), rrdset_id(r->st),
641 + r->child_db.first_entry_t,
642 + r->child_db.last_entry_t, r->child_db.fixed_last_entry ? " (fixed)" : "",
643 + r->child_db.wall_clock_time,
644 + r->last_request.after,
645 + r->last_request.before,
646 + msg,
647 + r->wanted.after,
648 + r->wanted.before,
649 + r->wanted.start_streaming ? "true" : "false");
650 +}
651 +
652 +static bool send_replay_chart_cmd(struct replication_request_details *r, const char *msg, bool log) {
653 RRDSET *st = r->st;
654
655 + if(log)
656 + replicate_log_request(r, msg);
657 +
658 if(st->rrdhost->receiver && (!st->rrdhost->receiver->replication_first_time_t || r->wanted.after < st->rrdhost->receiver->replication_first_time_t))
659 st->rrdhost->receiver->replication_first_time_t = r->wanted.after;
660
@@ -626,7 +671,7 @@ static bool send_replay_chart_cmd(struct replication_request_details *r, const c
671
672 internal_error(true,
673 "REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: "
629 - "last[%ld - %ld] child[%ld - %ld, now %ld %s] local[%ld - %ld %s, now %ld] gap[%ld - %ld %s] %s"
674 + "last[%ld - %ld] child[%ld - %ld, now %ld %s] local[%ld - %ld, now %ld] gap[%ld - %ld %s] %s"
675 , rrdhost_hostname(r->host), rrdset_id(r->st)
676 , r->wanted.after, wanted_after_buf
677 , r->wanted.before, wanted_before_buf
@@ -636,7 +681,7 @@ static bool send_replay_chart_cmd(struct replication_request_details *r, const c
681 , r->child_db.first_entry_t, r->child_db.last_entry_t
682 , r->child_db.world_time_t, (r->child_db.world_time_t == r->local_db.now) ? "SAME" : (r->child_db.world_time_t < r->local_db.now) ? "BEHIND" : "AHEAD"
683 , r->local_db.first_entry_t, r->local_db.last_entry_t
639 - , r->local_db.last_entry_t_adjusted_to_now?"FIXED":"RAW", r->local_db.now
684 + , r->local_db.now
685 , r->gap.from, r->gap.to
686 , (r->gap.from == r->wanted.after) ? "FULL" : "PARTIAL"
687 , (st->replay.after != 0 || st->replay.before != 0) ? "OVERLAPPING" : ""
@@ -663,7 +708,7 @@ static bool send_replay_chart_cmd(struct replication_request_details *r, const c
708 }
709
710 bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST *host, RRDSET *st,
666 - time_t first_entry_child, time_t last_entry_child, time_t child_world_time,
711 + time_t child_first_entry, time_t child_last_entry, time_t child_wall_clock_time,
712 time_t prev_first_entry_wanted, time_t prev_last_entry_wanted)
713 {
714 struct replication_request_details r = {
@@ -676,16 +721,16 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
721 .st = st,
722
723 .child_db = {
679 - .first_entry_t = first_entry_child,
680 - .last_entry_t = last_entry_child,
681 - .world_time_t = child_world_time,
724 + .first_entry_t = child_first_entry,
725 + .last_entry_t = child_last_entry,
726 + .wall_clock_time = child_wall_clock_time,
727 + .fixed_last_entry = false,
728 },
729
730 .local_db = {
685 - .first_entry_t = rrdset_first_entry_s(st),
686 - .last_entry_t = rrdset_last_entry_s(st),
687 - .last_entry_t_adjusted_to_now = false,
688 - .now = now_realtime_sec(),
731 + .first_entry_t = 0,
732 + .last_entry_t = 0,
733 + .wall_clock_time = now_realtime_sec(),
734 },
735
736 .last_request = {
@@ -700,12 +745,14 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
745 },
746 };
747
703 - // check our local database retention
704 - if(r.local_db.last_entry_t > r.local_db.now) {
705 - r.local_db.last_entry_t = r.local_db.now;
706 - r.local_db.last_entry_t_adjusted_to_now = true;
748 + if(r.child_db.last_entry_t > r.child_db.wall_clock_time) {
749 + replicate_log_request(&r, "child's db last entry > child's wall clock time");
750 + r.child_db.last_entry_t = r.child_db.wall_clock_time;
751 + r.child_db.fixed_last_entry = true;
752 }
753
754 + rrdset_get_retention_of_tier_for_collected_chart(r.st, &r.local_db.first_entry_t, &r.local_db.last_entry_t, r.local_db.wall_clock_time, 0);
755 +
756 // let's find the GAP we have
757 if(!r.last_request.after || !r.last_request.before) {
758 // there is no previous request
@@ -715,7 +762,7 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
762 r.gap.from = r.local_db.last_entry_t;
763 else
764 // we don't have any data, the gap is the max timeframe we are allowed to replicate
718 - r.gap.from = r.local_db.now - r.host->rrdpush_seconds_to_replicate;
765 + r.gap.from = r.local_db.wall_clock_time - r.host->rrdpush_seconds_to_replicate;
766
767 }
768 else {
@@ -726,27 +773,30 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
773 }
774
775 // we want all the data up to now
729 - r.gap.to = r.local_db.now;
776 + r.gap.to = r.local_db.wall_clock_time;
777
778 // The gap is now r.gap.from -> r.gap.to
779
780 if (unlikely(!rrdhost_option_check(host, RRDHOST_OPTION_REPLICATION)))
734 - return send_replay_chart_cmd(&r, "empty replication request, replication is disabled");
735 -
736 - if (unlikely(!r.child_db.last_entry_t))
737 - return send_replay_chart_cmd(&r, "empty replication request, child has no stored data");
781 + return send_replay_chart_cmd(&r, "empty replication request, replication is disabled", false);
782
783 if (unlikely(!rrdset_number_of_dimensions(st)))
740 - return send_replay_chart_cmd(&r, "empty replication request, chart has no dimensions");
784 + return send_replay_chart_cmd(&r, "empty replication request, chart has no dimensions", false);
785 +
786 + if (unlikely(!r.child_db.first_entry_t || !r.child_db.last_entry_t))
787 + return send_replay_chart_cmd(&r, "empty replication request, child has no stored data", false);
788
742 - if (r.child_db.first_entry_t <= 0)
743 - return send_replay_chart_cmd(&r, "empty replication request, first entry of the child db first entry is invalid");
789 + if (unlikely(r.child_db.first_entry_t < 0 || r.child_db.last_entry_t < 0))
790 + return send_replay_chart_cmd(&r, "empty replication request, child db timestamps are invalid", true);
791
745 - if (r.child_db.first_entry_t > r.child_db.last_entry_t)
746 - return send_replay_chart_cmd(&r, "empty replication request, child timings are invalid (first entry > last entry)");
792 + if (unlikely(r.child_db.first_entry_t > r.child_db.wall_clock_time))
793 + return send_replay_chart_cmd(&r, "empty replication request, child db first entry is after its wall clock time", true);
794
748 - if (r.local_db.last_entry_t > r.child_db.last_entry_t)
749 - return send_replay_chart_cmd(&r, "empty replication request, local last entry is later than the child one");
795 + if (unlikely(r.child_db.first_entry_t > r.child_db.last_entry_t))
796 + return send_replay_chart_cmd(&r, "empty replication request, child timings are invalid (first entry > last entry)", true);
797 +
798 + if (unlikely(r.local_db.last_entry_t > r.child_db.last_entry_t))
799 + return send_replay_chart_cmd(&r, "empty replication request, local last entry is later than the child one", false);
800
801 // let's find what the child can provide to fill that gap
802
@@ -768,15 +818,22 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
818 if(r.wanted.before > r.child_db.last_entry_t)
819 r.wanted.before = r.child_db.last_entry_t;
820
771 - if(r.wanted.after > r.wanted.before)
772 - r.wanted.after = r.wanted.before;
821 + if(r.wanted.after > r.wanted.before) {
822 + r.wanted.after = 0;
823 + r.wanted.before = 0;
824 + r.wanted.start_streaming = true;
825 + return send_replay_chart_cmd(&r, "empty replication request, wanted after computed bigger than wanted before", true);
826 + }
827
828 // the child should start streaming immediately if the wanted duration is small, or we reached the last entry of the child
775 - r.wanted.start_streaming = (r.local_db.now - r.wanted.after <= host->rrdpush_replication_step || r.wanted.before == r.child_db.last_entry_t);
829 + r.wanted.start_streaming = (r.local_db.wall_clock_time - r.wanted.after <= host->rrdpush_replication_step ||
830 + r.wanted.before >= r.child_db.last_entry_t ||
831 + r.wanted.before >= r.child_db.wall_clock_time ||
832 + r.wanted.before >= r.local_db.wall_clock_time);
833
834 // the wanted timeframe is now r.wanted.after -> r.wanted.before
835 // send it
779 - return send_replay_chart_cmd(&r, "OK");
836 + return send_replay_chart_cmd(&r, "OK", false);
837 }
838
839 // ----------------------------------------------------------------------------
streaming/replication.h
+1 -1
@@ -21,7 +21,7 @@ typedef int (*send_command)(const char *txt, void *data);
21
22 bool replicate_chart_request(send_command callback, void *callback_data,
23 RRDHOST *rh, RRDSET *rs,
24 - time_t first_entry_child, time_t last_entry_child, time_t child_world_time,
24 + time_t child_first_entry, time_t child_last_entry, time_t child_wall_clock_time,
25 time_t response_first_start_time, time_t response_last_end_time);
26
27 void replication_init_sender(struct sender_state *sender);