From: Dhairya Parmar Date: Fri, 8 May 2026 07:37:20 +0000 (+0530) Subject: tools/cephfs: flush every journal event as soon as it is scanned X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=aa2db56d089d10d4bbd566e3c9f8a5f2dcdee284;p=ceph.git tools/cephfs: flush every journal event as soon as it is scanned NOTE: exclusive to only `event recover_dentries` JournalScanner::scan() runs a loop and loads the entire decoded journal in an `EventMap` called `events`. This can severely shoot up the RSS if the journal is huge (unflushed journal) making the tool less efficient. The way to curb the memory bomb is to scan and flush journal in batches. The flush boundary is based on the batch size and at event EVENT_SUBTREEMAP or EVENT_SUBTREEMAP_TEST(subtree event(s) mark a clear boundary). The knob to control this setting is via providing `--max-rss ` with the journal tool command. This patch divides the provided limit by three to create the batch since it is observed that the decoded journal events weight on an avg ~3x the encoded stream. Also: Currently, the events map retains all decoded events until all dentries are recovered, causing unnecessary memory retention. Fix this by erasing each event from the map right after it is processed. While the primary RSS bloat is caused by dnbl holding the encoded stream in memory, this incremental cleanup provides a minor optimization until a permanent fix is implemented (https://tracker.ceph.com/issues/77909). Partially-Fixes: https://tracker.ceph.com/issues/76422 Co-authored-by: Patrick Donnelly Signed-off-by: Dhairya Parmar --- diff --git a/src/tools/cephfs/JournalScanner.cc b/src/tools/cephfs/JournalScanner.cc index 789dd54f844..03d37b77918 100644 --- a/src/tools/cephfs/JournalScanner.cc +++ b/src/tools/cephfs/JournalScanner.cc @@ -31,7 +31,7 @@ * that we were able to apply our checks, it does *not* mean that the journal is * healthy. */ -int JournalScanner::scan(bool const full) +int JournalScanner::scan(bool const full, EventCallback cb) { int r = 0; @@ -48,7 +48,7 @@ int JournalScanner::scan(bool const full) } if (full && header_present) { - r = scan_events(); + r = scan_events(cb); if (r < 0) { return r; } @@ -151,7 +151,7 @@ int JournalScanner::scan_header() } -int JournalScanner::scan_events() +int JournalScanner::scan_events(EventCallback cb) { uint64_t object_size = g_conf()->mds_log_segment_size; if (object_size == 0) { @@ -208,7 +208,7 @@ int JournalScanner::scan_events() } else { dout(4) << "Read 0x" << std::hex << this_object.length() << std::dec << " bytes from " << oid << " gap=" << gap << dendl; - objects_valid.push_back(oid); + ++num_objects_valid; this_object.begin().copy(this_object.length(), read_buf); } @@ -302,7 +302,12 @@ int JournalScanner::scan_events() } if (filter.apply(read_offset, *le)) { - events.insert_or_assign(read_offset, EventRecord(std::move(le), consumed)); + EventRecord er(std::move(le), consumed); + if (cb) { + cb(read_offset, er); + } else { + events.insert_or_assign(read_offset, std::move(er)); + } } } else { valid_entry = false; @@ -313,7 +318,12 @@ int JournalScanner::scan_events() auto q = le_bl.cbegin(); pi->decode(q); if (filter.apply(read_offset, *pi)) { - events.insert_or_assign(read_offset, EventRecord(std::move(pi), consumed)); + EventRecord er(std::move(pi), consumed); + if (cb) { + cb(read_offset, er); + } else { + events.insert_or_assign(read_offset, std::move(er)); + } } } catch (const buffer::error &err) { valid_entry = false; @@ -328,7 +338,7 @@ int JournalScanner::scan_events() read_offset += consumed; break; } else { - events_valid.push_back(read_offset); + ++num_events_valid; read_offset += consumed; } } @@ -340,10 +350,12 @@ int JournalScanner::scan_events() ranges_invalid.push_back(Range(gap_start, -1)); } - dout(4) << "Scanned objects, " << objects_missing.size() << " missing, " << objects_valid.size() << " valid" << dendl; + dout(4) << "Scanned objects, " << objects_missing.size() << " missing, " << num_objects_valid << " valid" << dendl; dout(4) << "Events scanned, " << ranges_invalid.size() << " gaps" << dendl; - dout(4) << "Found " << events_valid.size() << " valid events" << dendl; - dout(4) << "Selected " << events.size() << " events events for processing" << dendl; + dout(4) << "Found " << num_events_valid << " valid events" << dendl; + if (!cb) { + dout(4) << "Selected " << events.size() << " events for processing" << dendl; + } return 0; } diff --git a/src/tools/cephfs/JournalScanner.h b/src/tools/cephfs/JournalScanner.h index be2d27d039f..abe05c1d82d 100644 --- a/src/tools/cephfs/JournalScanner.h +++ b/src/tools/cephfs/JournalScanner.h @@ -15,6 +15,8 @@ #ifndef JOURNAL_SCANNER_H #define JOURNAL_SCANNER_H +#include + #include "include/rados/librados_fwd.hpp" // For Journaler::Header, can't forward-declare nested classes @@ -74,11 +76,21 @@ class JournalScanner ~JournalScanner(); + struct EventRecord { + EventRecord(std::unique_ptr le, uint32_t rs) : log_event(std::move(le)), raw_size(rs) {} + EventRecord(std::unique_ptr p, uint32_t rs) : pi(std::move(p)), raw_size(rs) {} + std::unique_ptr log_event; + std::unique_ptr pi; + uint32_t raw_size = 0; //< Size from start offset including all encoding overhead + }; + + using EventCallback = std::function; + int set_journal_ino(); - int scan(bool const full=true); + int scan(bool const full=true, EventCallback cb = nullptr); int scan_pointer(); int scan_header(); - int scan_events(); + int scan_events(EventCallback cb = nullptr); void report(std::ostream &out) const; std::string obj_name(uint64_t offset) const; @@ -86,13 +98,6 @@ class JournalScanner // The results of the scan inodeno_t ino; // Corresponds to journal ino according their type - struct EventRecord { - EventRecord(std::unique_ptr le, uint32_t rs) : log_event(std::move(le)), raw_size(rs) {} - EventRecord(std::unique_ptr p, uint32_t rs) : pi(std::move(p)), raw_size(rs) {} - std::unique_ptr log_event; - std::unique_ptr pi; - uint32_t raw_size = 0; //< Size from start offset including all encoding overhead - }; class EventError { public: @@ -114,10 +119,10 @@ class JournalScanner bool is_healthy() const; bool is_readable() const; - std::vector objects_valid; std::vector objects_missing; std::vector ranges_invalid; - std::vector events_valid; + uint64_t num_objects_valid{0}; + uint64_t num_events_valid{0}; EventMap events; // For events present in ::events (i.e. scanned successfully), diff --git a/src/tools/cephfs/JournalTool.cc b/src/tools/cephfs/JournalTool.cc index 9d99b5f5fec..20e14f2b159 100644 --- a/src/tools/cephfs/JournalTool.cc +++ b/src/tools/cephfs/JournalTool.cc @@ -72,7 +72,9 @@ void JournalTool::usage() << "\n" << "Special options\n" << " --alternate-pool Alternative metadata pool to target\n" - << " when using recover_dentries.\n"; + << " when using recover_dentries.\n" + << " --max-rss Maximum RSS allowed per batch during\n" + << " event recover_dentries.\n"; generic_client_usage(); } @@ -454,6 +456,9 @@ int JournalTool::main_event(std::vector &argv) } std::string output_path = "dump"; + uint64_t batch_size = 0; + uint64_t batch_bytes = 0; + JournalScanner::EventCallback flush = nullptr; while(arg != argv.end()) { std::string arg_str; if (ceph_argparse_witharg(argv, arg, &arg_str, "--path", (char*)NULL)) { @@ -464,6 +469,16 @@ int JournalTool::main_event(std::vector &argv) int r = rados.ioctx_create(arg_str.c_str(), output); ceph_assert(r == 0); other_pool = true; + } else if (ceph_argparse_witharg(argv, arg, &arg_str, "--max-rss", + nullptr)) { + std::string parse_err; + /* an estimate based on calculating journal event sizes, a decoded event + holds ~3x its encoded stream */ + batch_size = strict_strtoll(arg_str.c_str(), 0, &parse_err) / 3; + if (!parse_err.empty()) { + derr << "Invalid --max-rss value '" << arg_str << "': " << parse_err << dendl; + return -EINVAL; + } } else { cerr << "Unknown argument: '" << *arg << "'" << std::endl; return -EINVAL; @@ -482,45 +497,69 @@ int JournalTool::main_event(std::vector &argv) return r; } } else if (command == "recover_dentries") { - r = js.scan(); - if (r) { + std::set consumed_inos; + progress_tracker->set_operation_name("Processing events"); + + // decode the journal first to check if the header is valid + r = js.scan(false); + if (r < 0) { derr << "Failed to scan journal (" << cpp_strerror(r) << ")" << dendl; return r; } + if (!js.header_present || !js.header_valid) { + derr << "Journal header is not present or is damaged, cannot recover dentries" << dendl; + return -EIO; + } - /** - * Iterate over log entries, attempting to scavenge from each one - */ - std::set consumed_inos; - uint64_t event_count = js.events.size(); - progress_tracker->set_operation_name("Processing events"); - progress_tracker->start(event_count); - - - for (JournalScanner::EventMap::iterator i = js.events.begin(); - i != js.events.end(); ++i) { - auto& le = i->second.log_event; - EMetaBlob const *mb = le->get_metablob(); - if (mb) { - int scav_r = recover_dentries(*mb, dry_run, &consumed_inos); - if (scav_r) { - dout(1) << "Error processing event 0x" << std::hex << i->first << std::dec - << ": " << cpp_strerror(scav_r) << ", continuing..." << dendl; - if (r == 0) { - r = scav_r; + // start the progress tracker with the diff b/w write and expire position + progress_tracker->start(js.header->write_pos - js.header->expire_pos); + + auto flush_events = [&]() { + for (auto it = js.events.begin(); it != js.events.end(); ) { + const auto& le = it->second.log_event; + EMetaBlob const *mb = le->get_metablob(); + if (mb) { + int scav_r = recover_dentries(*mb, dry_run, &consumed_inos); + if (scav_r) { + dout(1) << "Error processing event 0x" << std::hex << it->first << std::dec + << ": " << cpp_strerror(scav_r) << ", continuing..." << dendl; + js.errors.insert(std::make_pair(it->first, + JournalScanner::EventError(scav_r, cpp_strerror(scav_r)))); } - // Our goal is to read all we can, so don't stop on errors, but - // do record them for possible later output - js.errors.insert(std::make_pair(i->first, - JournalScanner::EventError(scav_r, cpp_strerror(r)))); } + it = js.events.erase(it); + progress_tracker->increment(); + progress_tracker->display_progress(); } + batch_bytes = 0; + }; + + if (batch_size) { + std::cout << "Batch size: " << batch_size << " bytes" << std::endl; + + flush = [&](uint64_t offset, JournalScanner::EventRecord& er) { + batch_bytes += er.raw_size; + bool flushable = er.log_event + // factor in mds_debug_subtrees conf + && (er.log_event->get_type() == EVENT_SUBTREEMAP || er.log_event->get_type() == EVENT_SUBTREEMAP_TEST) + && batch_bytes >= batch_size; + js.events.insert_or_assign(offset, std::move(er)); + if (flushable) { + flush_events(); + } + }; + } - progress_tracker->increment(); - progress_tracker->display_progress(); + r = js.scan_events(flush); + + flush_events(); + + if (r) { + derr << "Failed to scan events (" << cpp_strerror(r) << ")" << dendl; + return r; } - progress_tracker->display_final_summary(); + progress_tracker->display_final_summary(); /**