diff options
author | Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com> | 2022-11-15 23:00:53 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2022-11-15 23:00:53 +0200 |
commit | 224b051a2b2bab39a4b536e531ab9ca590bf31bb (patch) | |
tree | adb3ca35d6d6a4d4f1b7aad50542619c3efb38c0 /database/engine/journalfile.c | |
parent | b4a0298bd48f217c4a6f2eaf729e0684966ea7a3 (diff) |
New journal disk based indexing for agent memory reduction (#13885)
* Add read only option to netdata_mmap so files are accessed ousing PROT_READ
* Initial functions to write the new journal file and switch to the new indexing
* Cleanup code, add parameters to pg_cache_punch_hole to avoid updating page latets oldest times
pg_cache insert to have parameter if page index locked needs to be done
Page eviction functions will try to deallocate the descriptor as well (pg_cache_punch_hole without page_index time updates)
Cleanup messages during startup
* Cleanup messages during startup
* Disbale extent caching for now, add placeholder for journal indexing and activation while the agent is running
* Add main function to populate descriptors by checking the new journal indexing
* prevent crash
* fix for binary search crash
* Avoid Time-of-check time-of-use filesystem race condition
* always add a page
* populate fixes - it is still incomplete
* pg_cache_insert returns the descriptor that ends up in the page_index
* Add populate next (Fix 1)
* Fix compilation warnings, reactivate extent caching
* Add populate next (Fix 2)
* Add populate next (Fix 3) switch to the next entry or journal file when asking to populate descriptor with next
* Fix resource leak and wrong sizeof
* Rework page population (part 1)
* Additional checksums added / journal validation
* Cleanup (part 1)
* Locking added and Cleanup (part 2)
* Close journal file after new journal index activation
* Skip warning when compiling without NETDATA_INTERNAL_CHECKS
* Ignore empty index file (header and trailer and no metrics)
* Try to remove all evicted descriptors (may prevent slight memory increase)
* Evict pages also when we succesfully do try_reserve
* Precache pages and cleanup
* Add a separate cleanup thread to release unused descriptors
* Check existence of key correctly
* Fix total file size calculation
* Statistics for journal descriptors
* Track and release jourval v2 descriptors
* Do not try to allocate pages for locality if under pressure
* Do not track v2 descriptors when populating the page_index
* Track page descriptors as they are inserted in the page index (per journal file)
Scan journal files for pending items to cleanup
Cleanup v2 descriptors only if they are not populated
Check before adding to page cache to avoid memory allocation /free
* Close journal file that has been processed and migrated to the new index
Check for valid file before trying to truncate / close. This file has been closed during startup
* Better calculation for the number of prefetched data pages based on the query end time
Code cleanup and comments
Add v2 populated descriptor expiration based on journal access time
* Code cleanup
* Faster indexing
Better journal validation (more sanity checks)
Detect new datafile/ journal creation and trigger index generation
Switch to the new index / mark descriptors in memory as needed
Update journal access time when a descriptor is returned
Code cleanup (part 1)
* Re activate descriptor clean
Code cleanup
* Allow locality precaching
* Allow locality precaching for the same page alignment
* Descriptor cleanup internal changed
* Disable locality precaching
* Precache only if not under pressure / internal cleanup at 60 seconds
* Remove unused functions
* Migrate on startup always
Make sure the metric uuid is valid (we have a page_index)
Prevent crash if no datafile is available when logging an error
Remove unused functions
* New warn limit for precaching
Stress test v2 descriptor cleanup
- Every 1s cleanup if it doesnt exist in cache
- 60s cache eviction
* Arrayalloc internal checks on free activated with NETDATA_ARRAYALLOC_INTERNAL_CHECKS
Ability to add DESCRIPTOR_EXPIRATION_TIME and DESCRIPTOR_INTERVAL_CLEANUP during compile
Defaults DESCRIPTOR_INTERVAL_CLEANUP = 60 and DESCRIPTOR_EXPIRATION_TIME = 600
* Lookup page index correctly
* Calculate index time once
* Detect a duplicate page when doing cache insert and during flushing of pages
* Better logging
* Descriptor validation (extent vs page index) when building an index file while the agent is running
* Mark invalid entries in the journal v2 file
* Schedule an index rebuild if a descriptor is found without an extent in the timerange we are processing
Release descriptor lock to prevent random shutdown locks
* Proper unlock
* Skip descriptor cleanup when journal file v2 migration is running
* Fix page cache statistics
Remove multiple entries of the page_index from the page cache
Cleanup
* Adjust preload pages on pg_cache_next. Handle invalid descriptor properly
Unlock properly
* Better handling of invalid pages
Journal indexing during runtime will scan all files to find potential ones to index
* Reactivate migration on startup
Evict descriptors to cause migration
Don't count the entries in page index (calculate when processing the extent list)
Check for valid extent since we may set the extent to NULL on startup if it is invalid
Better structure init
Address valgrind issues
* Add don't fork/dump option
* Add separate lock to protect accessing a datafile's extent list
Comment out some unused code (for now)
Abort descriptor cleanup if we are force flushing pages (page cache under pressure)
* Check for index and schedule when data flush completes
Configure max datafile size during compilation
Keep a separate JudyL array for descriptors
Skip quota test if we are deleting descriptors or explicitly flushing pages under pressure
* Fix
* set function when waiters are waken up
* add the line number to trace the deadlock
* add thread id
* add wait list
* init to zero
* disable thread cancelability inside dbengine rrdeng_load_page_next()
* make sure the owner is the thread
* disable thread cancelability for replication as a whole
* Check and queue indexing after first page flush
* Queue indexing after a small delay to allow some time for page flushing
* tracing of waiters only when compiled with internal checks
* Mark descr with extent_entry
* Return page timeout
* Check if a journalfile is ready to be indexed
Migrate the descriptors or evict if possible
Compilation warning fix
* Use page index if indexing during startup
Mark if journalfile should be checked depending on whether we can migrate or delete a page during indexing
* require 3x max message size as sender buffer
* fix for the msg of the adaptive buffer size
* fix for the msg of the duplicate replication commands
* Disable descriptor deletion during migration
* Detect descriptor with same start page time
* sender sorts replication requests before fullfilling them; receiver does not send duplicate replication requests
* dbengine never allows past timestamps to be collected
* do not accept values same as last data point stored in dbengine
* replicate non-overlapping ranges
* a better replication logic to avoid sending overlapping data to parents
* Do not start journal migration in parallel
* Always update page index times
* Fix page index first / last times on load
* internal log when replication responses do not match the requests or when replication commands are sent while there are others inflight
* do not log out of bounds RBEGIN if it is the last replication command we sent
* better checking of past data collection points
* better checking of past data collection points - optimized
* fix corruption during decompression of streaming
* Add config to disable journal indexing
Add config parameter for detailed journal integrity check (Metric chain validation check during startup)
pg cache insert drop check for existing page
Fix crc calculation for metric headers
* children disable compression globally, only when the compression gives an error
* turn boolean member into RRDHOST OPTION
* Compilation warnings
* Remove unused code
* replication sender statistics
* replication sender statistics set to 100% when no replication requests are pending
* Fix casting warning
Co-authored-by: Costa Tsaousis <costa@netdata.cloud>
Diffstat (limited to 'database/engine/journalfile.c')
-rw-r--r-- | database/engine/journalfile.c | 1126 |
1 files changed, 1092 insertions, 34 deletions
diff --git a/database/engine/journalfile.c b/database/engine/journalfile.c index 500dd78800..037603b9b6 100644 --- a/database/engine/journalfile.c +++ b/database/engine/journalfile.c @@ -1,6 +1,25 @@ // SPDX-License-Identifier: GPL-3.0-or-later #include "rrdengine.h" +void queue_journalfile_v2_migration(struct rrdengine_worker_config *wc) +{ + struct rrdeng_work *work_request; + + if (unlikely(wc->running_journal_migration)) + return; + + work_request = mallocz(sizeof(*work_request)); + work_request->req.data = work_request; + work_request->wc = wc; + work_request->count = 0; + work_request->rerun = false; + wc->running_journal_migration = 1; + if (unlikely(uv_queue_work(wc->loop, &work_request->req, start_journal_indexing, after_journal_indexing))) { + freez(work_request); + wc->running_journal_migration = 0; + } +} + static void flush_transaction_buffer_cb(uv_fs_t* req) { struct generic_io_descriptor *io_descr = req->data; @@ -47,6 +66,7 @@ void wal_flush_transaction_buffer(struct rrdengine_worker_config* wc) io_descr->bytes = size; io_descr->pos = journalfile->pos; io_descr->req.data = io_descr; + io_descr->data = journalfile; io_descr->completion = NULL; io_descr->iov = uv_buf_init((void *)io_descr->buf, size); @@ -93,6 +113,12 @@ void * wal_get_transaction_buffer(struct rrdengine_worker_config* wc, unsigned s return ctx->commit_log.buf + buf_pos; } +void generate_journalfilepath_v2(struct rrdengine_datafile *datafile, char *str, size_t maxlen) +{ + (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION_V2, + datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno); +} + void generate_journalfilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen) { (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION, @@ -104,28 +130,51 @@ void journalfile_init(struct rrdengine_journalfile *journalfile, struct rrdengin journalfile->file = (uv_file)0; journalfile->pos = 0; journalfile->datafile = datafile; + journalfile->journal_data = NULL; + journalfile->journal_data_size = 0; + journalfile->JudyL_array = (Pvoid_t) NULL; + journalfile->data = NULL; + journalfile->file_index = 0; + journalfile->last_access = 0; } -int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile) +static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file) { - struct rrdengine_instance *ctx = datafile->ctx; - uv_fs_t req; int ret; char path[RRDENG_PATH_MAX]; - generate_journalfilepath(datafile, path, sizeof(path)); - - ret = uv_fs_close(NULL, &req, journalfile->file, NULL); + uv_fs_t req; + ret = uv_fs_close(NULL, &req, file, NULL); if (ret < 0) { + generate_journalfilepath(datafile, path, sizeof(path)); error("uv_fs_close(%s): %s", path, uv_strerror(ret)); - ++ctx->stats.fs_errors; + ++datafile->ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); } uv_fs_req_cleanup(&req); - return ret; } +int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile) +{ + struct rrdengine_instance *ctx = datafile->ctx; + char path[RRDENG_PATH_MAX]; + + if (likely(journalfile->journal_data)) { + if (munmap(journalfile->journal_data, journalfile->journal_data_size)) { + generate_journalfilepath_v2(datafile, path, sizeof(path)); + error("Failed to unmap journal index file for %s", path); + ++ctx->stats.fs_errors; + rrd_stat_atomic_add(&global_fs_errors, 1); + } + journalfile->journal_data = NULL; + journalfile->journal_data_size = 0; + return 0; + } + + return close_uv_file(datafile, journalfile->file); +} + int unlink_journal_file(struct rrdengine_journalfile *journalfile) { struct rrdengine_datafile *datafile = journalfile->datafile; @@ -149,14 +198,16 @@ int unlink_journal_file(struct rrdengine_journalfile *journalfile) return ret; } -int destroy_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile) +int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile) { struct rrdengine_instance *ctx = datafile->ctx; uv_fs_t req; int ret; char path[RRDENG_PATH_MAX]; + char path_v2[RRDENG_PATH_MAX]; generate_journalfilepath(datafile, path, sizeof(path)); + generate_journalfilepath_v2(datafile, path_v2, sizeof(path)); ret = uv_fs_ftruncate(NULL, &req, journalfile->file, 0, NULL); if (ret < 0) { @@ -166,9 +217,12 @@ int destroy_journal_file(struct rrdengine_journalfile *journalfile, struct rrden } uv_fs_req_cleanup(&req); - ret = uv_fs_close(NULL, &req, journalfile->file, NULL); + (void) close_uv_file(datafile, journalfile->file); + + // This is the new journal v2 index file + ret = uv_fs_unlink(NULL, &req, path_v2, NULL); if (ret < 0) { - error("uv_fs_close(%s): %s", path, uv_strerror(ret)); + error("uv_fs_fsunlink(%s): %s", path, uv_strerror(ret)); ++ctx->stats.fs_errors; rrd_stat_atomic_add(&global_fs_errors, 1); } @@ -183,6 +237,13 @@ int destroy_journal_file(struct rrdengine_journalfile *journalfile, struct rrden uv_fs_req_cleanup(&req); ++ctx->stats.journalfile_deletions; + ++ctx->stats.journalfile_deletions; + + if (journalfile->journal_data) { + if (munmap(journalfile->journal_data, journalfile->journal_data_size)) { + error("Failed to unmap index file %s", path_v2); + } + } return ret; } @@ -227,7 +288,7 @@ int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdeng uv_fs_req_cleanup(&req); posix_memfree(superblock); if (ret < 0) { - destroy_journal_file(journalfile, datafile); + destroy_journal_file_unsafe(journalfile, datafile); return ret; } @@ -366,36 +427,67 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden uv_rwlock_wrlock(&pg_cache->metrics_index.lock); PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, temp_id, sizeof(uuid_t), PJE0); fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */ - *PValue = page_index = create_page_index(temp_id, ctx); - page_index->prev = pg_cache->metrics_index.last_page_index; - pg_cache->metrics_index.last_page_index = page_index; + *PValue = page_index = create_page_index(temp_id, ctx); + page_index->prev = pg_cache->metrics_index.last_page_index; + pg_cache->metrics_index.last_page_index = page_index; uv_rwlock_wrunlock(&pg_cache->metrics_index.lock); } - descr = pg_cache_create_descr(); + // Lookup this descriptor + Word_t p_time = start_time_ut / USEC_PER_SEC; + uv_rwlock_rdlock(&page_index->lock); + PValue = JudyLFirst(page_index->JudyL_array, &p_time, PJE0); + descr = (NULL == PValue) ? NULL: *PValue; + uv_rwlock_rdunlock(&page_index->lock); + + bool descr_found = false; + if (unlikely(descr && descr->start_time_ut == start_time_ut)) { + // We have this descriptor already + descr_found = true; + +#ifdef NETDATA_INTERNAL_CHECKS + char uuid_str[UUID_STR_LEN]; + uuid_unparse_lower(page_index->id, uuid_str); + internal_error(true, "REMOVING UUID %s with %lu, %lu length=%u (fileno=%u), extent database offset = %lu, size = %u", + uuid_str, start_time_ut, end_time_ut, jf_metric_data->descr[i].page_length, descr->extent->datafile->fileno, + descr->extent->offset, descr->extent->size); + internal_error(true, "APPLYING UUID %s with %lu, %lu length=%u (fileno=%u), extent database offset = %lu, size = %u", + uuid_str, start_time_ut, end_time_ut, jf_metric_data->descr[i].page_length, extent->datafile->fileno, + extent->offset, extent->size); +#endif + // Remove entry from previous extent + unlink_descriptor_extent_unsafe(descr); + internal_error(true, "REMOVING UUID %s with %lu OK", uuid_str, start_time_ut); + } + else { + descr = pg_cache_create_descr(); + descr->id = &page_index->id; + } + descr->page_length = jf_metric_data->descr[i].page_length; descr->start_time_ut = start_time_ut; descr->end_time_ut = end_time_ut; descr->update_every_s = (update_every_s > 0) ? (uint32_t)update_every_s : (page_index->latest_update_every_s); - descr->id = &page_index->id; + descr->extent = extent; descr->type = page_type; extent->pages[valid_pages++] = descr; - pg_cache_insert(ctx, page_index, descr); + if (likely(!descr_found)) + (void)pg_cache_insert(ctx, page_index, descr, true); - if(page_index->latest_time_ut == descr->end_time_ut) + if (page_index->latest_time_ut == descr->end_time_ut) page_index->latest_update_every_s = descr->update_every_s; if(descr->update_every_s == 0) - fatal( - "DBENGINE: page descriptor update every is zero, end_time_ut = %llu, start_time_ut = %llu, entries = %zu", + fatal("DBENGINE: page descriptor update every is zero, end_time_ut = %llu, start_time_ut = %llu, entries = %zu", (unsigned long long)end_time_ut, (unsigned long long)start_time_ut, entries); } extent->number_of_pages = valid_pages; - if (likely(valid_pages)) + if (likely(valid_pages)) { df_extent_insert(extent); + } else { freez(extent); ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter++; @@ -444,13 +536,13 @@ static unsigned replay_transaction(struct rrdengine_instance *ctx, struct rrdeng return size_bytes; } switch (jf_header->type) { - case STORE_DATA: - debug(D_RRDENGINE, "Replaying transaction %"PRIu64"", jf_header->id); - restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length); - break; - default: - error("Unknown transaction type. Skipping record."); - break; + case STORE_DATA: + debug(D_RRDENGINE, "Replaying transaction %"PRIu64"", jf_header->id); + restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length); + break; + default: + error("Unknown transaction type. Skipping record."); + break; } return size_bytes; @@ -518,12 +610,905 @@ static uint64_t iterate_transactions(struct rrdengine_instance *ctx, struct rrde if (likely(journal_is_mmapped)) buf += size_bytes; } -skip_file: + skip_file: if (unlikely(!journal_is_mmapped)) posix_memfree(buf); return max_id; } +bool unlink_descriptor_extent_unsafe(struct rrdeng_page_descr *descr) +{ + if (unlikely(!descr || !descr->extent)) + return true; + + struct extent_info *extent = descr->extent; + for (uint8_t index = 0; index < extent->number_of_pages; index++) { + if (extent->pages[index] == descr) { + extent->pages[index] = NULL; + return true; + } + } + return false; +} + +// Checks that the extent list checksum is valid +static int check_journal_v2_extent_list (void *data_start, size_t file_size) +{ + UNUSED(file_size); + uLong crc; + + struct journal_v2_header *j2_header = (void *) data_start; + struct journal_v2_block_trailer *journal_v2_trailer; + + journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + j2_header->extent_trailer_offset); + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (uint8_t *) data_start + j2_header->extent_offset, j2_header->extent_count * sizeof(struct journal_extent_list)); + if (unlikely(crc32cmp(journal_v2_trailer->checksum, crc))) { + error("Extent list CRC32 check: FAILED"); + return 1; + } + + return 0; +} + +// Checks that the metric list (UUIDs) checksum is valid +static int check_journal_v2_metric_list(void *data_start, size_t file_size) +{ + UNUSED(file_size); + uLong crc; + + struct journal_v2_header *j2_header = (void *) data_start; + struct journal_v2_block_trailer *journal_v2_trailer; + + journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + j2_header->metric_trailer_offset); + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (uint8_t *) data_start + j2_header->metric_offset, j2_header->metric_count * sizeof(struct journal_metric_list)); + if (unlikely(crc32cmp(journal_v2_trailer->checksum, crc))) { + error("Metric list CRC32 check: FAILED"); + return 1; + } + return 0; +} + +static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t original_size) +{ + int rc; + uLong crc; + + struct journal_v2_header *j2_header = (void *) data_start; + struct journal_v2_block_trailer *journal_v2_trailer; + + if (j2_header->magic == JOURVAL_V2_REBUILD_MAGIC) + return 2; + + // Magic failure + if (j2_header->magic != JOURVAL_V2_MAGIC) + return 1; + + if (j2_header->total_file_size != file_size) + return 1; + + if (original_size && j2_header->original_file_size != original_size) + return 1; + + journal_v2_trailer = (struct journal_v2_block_trailer *) ((uint8_t *) data_start + file_size - sizeof(*journal_v2_trailer)); + + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (void *) j2_header, sizeof(*j2_header)); + + rc = crc32cmp(journal_v2_trailer->checksum, crc); + if (unlikely(rc)) { + error("File CRC32 check: FAILED"); + return 1; + } + + rc = check_journal_v2_extent_list(data_start, file_size); + if (rc) return 1; + + rc = check_journal_v2_metric_list(data_start, file_size); + if (rc) return 1; + + if (!db_engine_journal_check) + return 0; + + // Verify complete UUID chain + + struct journal_metric_list *metric = (void *) (data_start + j2_header->metric_offset); + + unsigned verified = 0; + unsigned entries; + unsigned total_pages = 0; + + info("Checking %u metrics that exist in the journal", j2_header->metric_count); + for (entries = 0; entries < j2_header->metric_count; entries++) { + + char uuid_str[UUID_STR_LEN]; + uuid_unparse_lower(metric->uuid, uuid_str); + struct journal_page_header *metric_list_header = (void *) (data_start + metric->page_offset); + struct journal_page_header local_metric_list_header = *metric_list_header; + + local_metric_list_header.crc = JOURVAL_V2_MAGIC; + + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (void *) &local_metric_list_header, sizeof(local_metric_list_header)); + rc = crc32cmp(metric_list_header->checksum, crc); + + internal_error(true, "Index %u : %s entries %u at offset %u (%llu -- %llu) verified, HEADER CRC computed %lu, stored %u", entries, uuid_str, metric->entries, metric->page_offset, + j2_header->start_time_ut + (usec_t) metric->delta_start * USEC_PER_SEC, j2_header->start_time_ut + (usec_t) metric->delta_end * USEC_PER_SEC, + crc, metric_list_header->crc); + if (!rc) { + struct journal_v2_block_trailer *journal_trailer = + (void *) data_start + metric->page_offset + sizeof(struct journal_page_header) + (metric_list_header->entries * sizeof(struct journal_page_list)); + + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (uint8_t *) metric_list_header + sizeof(struct journal_page_header), metric_list_header->entries * sizeof(struct journal_page_list)); + rc = crc32cmp(journal_trailer->checksum, crc); + internal_error(rc, "Index %u : %s entries %u at offset %u verified, DATA CRC computed %lu, stored %u", entries, uuid_str, metric->entries, metric->page_offset, + crc, metric_list_header->crc); + if (!rc) { + total_pages += metric_list_header->entries; + verified++; + } + } + + metric++; + if (((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) file_size) { + info("Verification failed EOF reached -- total entries %u, verified %u", entries, verified); + return 1; + } + } + + if (entries != verified) { + info("Verification failed -- total entries %u, verified %u", entries, verified); + return 1; + } + info("Verification succeeded -- total entries %u, verified %u (%u total pages)", entries, verified, total_pages); + + return 0; +} + +int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile) +{ + struct page_cache *pg_cache = &ctx->pg_cache; + struct pg_cache_page_index *page_index; + int ret, fd; + uint64_t file_size; + char path[RRDENG_PATH_MAX]; + struct stat statbuf; + uint32_t original_file_size = 0; + + generate_journalfilepath(datafile, path, sizeof(path)); + ret = stat(path, &statbuf); + if (!ret) + original_file_size = (uint32_t)statbuf.st_size; + + generate_journalfilepath_v2(datafile, path, sizeof(path)); + + fd = open(path, O_RDONLY); + if (fd < 0) { + if (errno == ENOENT) + return 1; + ++ctx->stats.fs_errors; + rrd_stat_atomic_add(&global_fs_errors, 1); + error("Failed to open %s", path); + return 1; + } + + ret = fstat(fd, &statbuf); + if (ret) { + error("Failed to get file information for %s", path); + close(fd); + return 1; + } + + file_size = (size_t)statbuf.st_size; + + if (file_size < sizeof(struct journal_v2_header)) { + error_report("Invalid file %s. Not the expected size", path); + close(fd); + return 1; + } + + usec_t start_loading = now_realtime_usec(); + uint8_t *data_start = mmap(NULL, file_size, PROT_READ, MAP_SHARED, fd, 0); + if (data_start == MAP_FAILED) { + close(fd); + return 1; + } + close(fd); + + info("Checking integrity of %s", path); + int rc = check_journal_v2_file(data_start, file_size, original_file_size); + if (rc) { + if (rc == 2) + error_report("File %s needs to be rebuilt", path); + else + error_report("File %s is invalid", path); + if (unlikely(munmap(data_start, file_size))) + error("Failed to unmap %s", path); + return 1; + } + + struct journal_v2_header *j2_header = (void *) data_start; + + size_t entries = j2_header->metric_count; + + if (!entries) { + if (unlikely(munmap(data_start, file_size))) + error("Failed to unmap %s", path); + return 1; + } + + rc = madvise(data_start, file_size, MADV_DONTFORK); + if (rc) + error("MADV_DONTFORK: setting failed"); + + rc = madvise(data_start, file_size, MADV_DONTDUMP); + if (rc) + error("MADV_DONTDUMP: setting failed"); + + struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset); + + uv_rwlock_wrlock(&pg_cache->metrics_index.lock); + + // Initialize the journal file to be able to access the data + journalfile->journal_data = data_start; + journalfile->journal_data_size = file_size; + + usec_t header_start_time = j2_header->start_time_ut; + for (size_t i=0; i < entries; i++) { + Pvoid_t *PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, metric->uuid, sizeof(uuid_t)); + if (likely(NULL != PValue)) { + page_index = *PValue; + } + else { + PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, metric->uuid, sizeof(uuid_t), PJE0); + fatal_assert(NULL == *PValue); + *PValue = page_index = create_page_index(&metric->uuid, ctx); + page_index->oldest_time_ut = LLONG_MAX; + page_index->latest_time_ut = 0; + page_index->prev = pg_cache->metrics_index.last_page_index; + pg_cache->metrics_index.last_page_index = page_index; + } + + usec_t metric_start_ut = header_start_time + (usec_t ) metric->delta_start * USEC_PER_SEC; + usec_t metric_end_ut = header_start_time + (usec_t ) metric->delta_end * USEC_PER_SEC; + + if (page_index->oldest_time_ut > metric_start_ut) + page_index->oldest_time_ut = metric_start_ut; + + if (page_index->latest_time_ut < metric_end_ut) + page_index->latest_time_ut = metric_end_ut; + + ++page_index->page_count; + ++pg_cache->page_descriptors; + metric++; + } + + uv_rwlock_wrunlock(&pg_cache->metrics_index.lock); + + info("Journal file \"%s\" loaded (size:%"PRIu64") with %lu metrics in %d ms", + path, file_size, entries, + (int) ((now_realtime_usec() - start_loading) / USEC_PER_MS)); + return 0; +} + + +struct metric_info_s { + uuid_t *id; + struct pg_cache_page_index *page_index; + uint32_t entries; + time_t min_index_time_s; + time_t max_index_time_s; + usec_t min_time_ut; + usec_t max_time_ut; + uint32_t page_list_header; + Pvoid_t JudyL_array; +}; + +struct journal_metric_list_to_sort { + struct metric_info_s *metric_info; +}; + +static int journal_metric_compare (const void *item1, const void *item2) +{ + const struct metric_info_s *metric1 = ((struct journal_metric_list_to_sort *) item1)->metric_info; + const struct metric_info_s *metric2 = ((struct journal_metric_list_to_sort *) item2)->metric_info; + + return uuid_compare(*(metric1->id), *(metric2->id)); +} + + +// Write list of extents for the journalfile +void *journal_v2_write_extent_list(struct rrdengine_journalfile *journalfile, void *data) +{ + struct extent_info *extent = journalfile->datafile->extents.first; + struct journal_extent_list *j2_extent = (void *) data; + while (extent) { + j2_extent->datafile_offset = extent->offset; + j2_extent->datafile_size = extent->size; + j2_extent->pages = extent->number_of_pages; + j2_extent->file_index = journalfile->file_index; + j2_extent++; + extent = extent->next; + }; + return j2_extent; +} + +static int verify_journal_space(struct journal_v2_header *j2_header, void *data, uint32_t bytes) +{ + if ((unsigned long)(((uint8_t *) data - (uint8_t *) j2_header->data) + bytes) > (j2_header->total_file_size - sizeof(struct journal_v2_block_trailer))) + return 1; + + return 0; +} + +void *journal_v2_write_metric_page(struct journal_v2_header *j2_header, void *data, struct metric_info_s *metric_info, uint32_t pages_offset) +{ + struct journal_metric_list *metric = (void *) data; + + if (verify_journal_space(j2_header, data, sizeof(*metric))) + return NULL; + + uuid_copy(metric->uuid, *metric_info->id); + metric->entries = metric_info->entries; + metric->page_offset = pages_offset; + metric->delta_start = (metric_info->min_time_ut - j2_header->start_time_ut) / USEC_PER_SEC; + metric->delta_end = (metric_info->max_time_ut - j2_header->start_time_ut) / USEC_PER_SEC; + + return ++metric; +} + +void *journal_v2_write_data_page_header(struct journal_v2_header *j2_header __maybe_unused, void *data, struct metric_info_s *metric_info, uint32_t uuid_offset) +{ + struct journal_page_header *data_page_header = (void *) data; + uLong crc; + + uuid_copy(data_page_header->uuid, *metric_info->id); + data_page_header->entries = metric_info->entries; + data_page_header->uuid_offset = uuid_offset; // data header OFFSET poings to METRIC in the directory + data_page_header->crc = JOURVAL_V2_MAGIC; + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (void *) data_page_header, sizeof(*data_page_header)); + crc32set(data_page_header->checksum, crc); + return ++data_page_header; +} + +void *journal_v2_write_data_page_trailer(struct journal_v2_header *j2_header __maybe_unused, void *data, void *page_header) +{ + struct journal_page_header *data_page_header = (void *) page_header; + struct journal_v2_block_trailer *journal_trailer = (void *) data; + uLong crc; + + crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, (uint8_t *) page_header + sizeof(struct journal_page_header), data_page_header->entries * sizeof(struct journal_page_list)); + crc32set(journal_trailer->checksum, crc); + return ++journal_trailer; +} + +void *journal_v2_write_data_page(struct journal_v2_header *j2_header, void *data, struct rrdeng_page_descr *descr) +{ + if (unlikely(!descr)) + return data; + + struct journal_page_list *data_page = data; + + // verify that we can write number of bytes + if (verify_journal_space(j2_header, data, sizeof(*data_page))) + return NULL; + + fatal_assert(descr->extent != NULL); + + uint32_t extent_index = unlikely(NULL == descr->extent) ? UINT32_MAX : descr->extent->index; + + data_page->delta_start_s = (descr->start_time_ut - j2_header->start_time_ut) / USEC_PER_SEC; + data_page->delta_end_s = (descr->end_time_ut - j2_header->start_time_ut) / USEC_PER_SEC; + data_page->extent_index = extent_index; + data_page->update_every_s = (uint16_t) descr->update_every_s; + data_page->page_length = descr->page_length; + data_page->type = descr->type; + + // Rebuild on start to resolve unknown entry + if (unlikely(UINT32_MAX == extent_index)) + j2_header->magic = JOURVAL_V2_REBUILD_MAGIC; + + return ++data_page; +} + +// For a page_index write all descr @ time entries +// Must be recorded in metric_info->entries +void *journal_v2_write_descriptors(struct journal_v2_header *j2_header, void *data, struct metric_info_s *metric_info, struct rrdengine_journalfile *journalfile) +{ + struct rrdeng_page_descr *descr; + Pvoid_t *PValue; + + struct journal_page_list *data_page = (void *)data; + struct page_cache *pg_cache = &journalfile->datafile->ctx->pg_cache; + struct pg_cache_page_index *page_index; + + uv_rwlock_rdlock(&pg_cache->metrics_index.lock); + PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, metric_info->id, sizeof(uuid_t)); + page_index = (NULL == PValue) ? NULL : *PValue; + uv_rwlock_rdunlock(&pg_cache->metrics_index.lock); + + if (page_index == NULL) + return data_page; + + // We need to write all descriptors with index metric_info->min_index_time_s, metric_info->max_index_time_s + // that belong to this journal file + + Word_t index_time = metric_info->min_index_time_s; + unsigned entries = 0; + + uv_rwlock_rdlock(&page_index->lock); + + Pvoid_t JudyL_array = metric_info->JudyL_array ? metric_info->JudyL_array : page_index->JudyL_array; + + // Need page_index lock if running live + for (PValue = JudyLFirst(JudyL_array, &index_time, PJE0), + descr = unlikely(NULL == PValue) ? NULL : *PValue; + descr != NULL; + PValue = JudyLNext(JudyL_array, &index_time, PJE0), + descr = unlikely(NULL == PValue) ? NULL : *PValue) { + + if (unlikely((time_t) index_time > metric_info->max_index_time_s) || entries == metric_info->entries) + break; + + // Write one descriptor and return the next data page location + data_page = journal_v2_write_data_page(j2_header, (void *)data_page, descr); + + entries++; + if (unlikely(!data_page)) + break; + } + + uv_rwlock_rdunlock(&page_index->lock); + return data_page; +} + +static void journal_v2_remove_active_descriptors(struct rrdengine_journalfile *journalfile, struct metric_info_s *metric_info, bool startup) +{ + if (true == startup) { + // This is during startup, so we are the only ones accessing the structures + // thats why we can safely remote the entire page_index->JudyL_array + struct rrdeng_page_descr *descr; + Word_t index_time; + Pvoid_t *PValue; + struct pg_cache_page_index *page_index; + + page_index = metric_info->page_index; + + for (index_time = 0, PValue = JudyLFirst(page_index->JudyL_array, &index_time, PJE0), + descr = unlikely(NULL == PValue) ? NULL : *PValue; descr != NULL; + PValue = JudyLNext(page_index->JudyL_array, &index_time, PJE0), + descr = unlikely(NULL == PValue) ? NULL : *PValue) { + + rrdeng_page_descr_freez(descr); + } + (void)JudyLFreeArray(&page_index->JudyL_array, PJE0); + } + else { + // This is during runtime + struct rrdeng_page_descr *descr; + Pvoid_t *PValue; + struct pg_cache_page_index *page_index = metric_info->page_index; + struct page_cache *pg_cache = &page_index->ctx->pg_cache; + struct rrdengine_instance *ctx = page_index->ctx; + + Word_t index_time = metric_info->min_index_time_s; + uint32_t metric_info_offset = metric_info->page_list_header; + + struct journal_pa |