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 <bytes>` 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 <pdonnell@ibm.com>
Signed-off-by: Dhairya Parmar <dparmar@redhat.com>

diff --git a/src/tools/cephfs/JournalScanner.cc b/src/tools/cephfs/JournalScanner.cc
index 789dd54f8444296f1815360aab8fb372b5c17956..03d37b779186f7197f69955fa700e5347321cea0 100644 (file)
--- 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 be2d27d039f8a9788b95c6b04a138522e314a51e..abe05c1d82db6b23857664713a336d08c3c03871 100644 (file)
--- 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 <functional>
+
 #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<LogEvent> le, uint32_t rs) : log_event(std::move(le)), raw_size(rs) {}
+    EventRecord(std::unique_ptr<PurgeItem> p, uint32_t rs) : pi(std::move(p)), raw_size(rs) {}
+    std::unique_ptr<LogEvent> log_event;
+    std::unique_ptr<PurgeItem> pi;
+    uint32_t raw_size = 0;  //< Size from start offset including all encoding overhead
+  };
+
+  using EventCallback = std::function<void(uint64_t, EventRecord&)>;
+
   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<LogEvent> le, uint32_t rs) : log_event(std::move(le)), raw_size(rs) {}
-    EventRecord(std::unique_ptr<PurgeItem> p, uint32_t rs) : pi(std::move(p)), raw_size(rs) {}
-    std::unique_ptr<LogEvent> log_event;
-    std::unique_ptr<PurgeItem> 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<std::string> objects_valid;
   std::vector<uint64_t> objects_missing;
   std::vector<Range> ranges_invalid;
-  std::vector<uint64_t> 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 9d99b5f5feca79d4b2657b3c86854ae7dd813b67..20e14f2b159af1dd3260a3eab0fefa61c6e21256 100644 (file)
--- 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 <name>     Alternative metadata pool to target\n"
-    << "                              when using recover_dentries.\n";
+    << "                              when using recover_dentries.\n"
+    << "  --max-rss <bytes>           Maximum RSS allowed per batch during\n"
+    << "                              event recover_dentries.\n";
 
   generic_client_usage();
 }
@@ -454,6 +456,9 @@ int JournalTool::main_event(std::vector<const char*> &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<const char*> &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<const char*> &argv)
       return r;
     }
   } else if (command == "recover_dentries") {
-    r = js.scan();
-    if (r) {
+    std::set<inodeno_t> 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<inodeno_t> 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();
 
 
     /**