mimic: cephfs: some tool commands silently operate on only rank 0, even if multiple ranks exist

qa/tasks/cephfs/filesystem.py

@@ -1228,6 +1228,9 @@ def _prefix(self):
         """
         return ""
 
+    def _make_rank(self, rank):
+        return "{}:{}".format(self.name, rank)
+
     def _run_tool(self, tool, args, rank=None, quiet=False):
         # Tests frequently have [client] configuration that jacks up
         # the objecter log level (unlikely to be interesting here)
@@ -1238,7 +1241,7 @@ def _run_tool(self, tool, args, rank=None, quiet=False):
             base_args = [os.path.join(self._prefix, tool), '--debug-mds=4', '--debug-objecter=1']
 
         if rank is not None:
-            base_args.extend(["--rank", "%d" % rank])
+            base_args.extend(["--rank", "%s" % str(rank)])
 
         t1 = datetime.datetime.now()
         r = self.tool_remote.run(
@@ -1260,11 +1263,12 @@ def tool_remote(self):
         mds_id = self.mds_ids[0]
         return self.mds_daemons[mds_id].remote
 
-    def journal_tool(self, args, rank=None, quiet=False):
+    def journal_tool(self, args, rank, quiet=False):
         """
-        Invoke cephfs-journal-tool with the passed arguments, and return its stdout
+        Invoke cephfs-journal-tool with the passed arguments for a rank, and return its stdout
         """
-        return self._run_tool("cephfs-journal-tool", args, rank, quiet)
+        fs_rank = self._make_rank(rank)
+        return self._run_tool("cephfs-journal-tool", args, fs_rank, quiet)
 
     def table_tool(self, args, quiet=False):
         """

qa/tasks/cephfs/test_damage.py

@@ -482,7 +482,7 @@ def test_open_ino_errors(self):
 
         # Drop everything from the MDS cache
         self.mds_cluster.mds_stop()
-        self.fs.journal_tool(['journal', 'reset'])
+        self.fs.journal_tool(['journal', 'reset'], 0)
         self.mds_cluster.mds_fail_restart()
         self.fs.wait_for_daemons()
 

qa/tasks/cephfs/test_data_scan.py

@@ -362,9 +362,9 @@ def get_state(mds_id):
         if False:
             with self.assertRaises(CommandFailedError):
                 # Normal reset should fail when no objects are present, we'll use --force instead
-                self.fs.journal_tool(["journal", "reset"])
+                self.fs.journal_tool(["journal", "reset"], 0)
 
-        self.fs.journal_tool(["journal", "reset", "--force"])
+        self.fs.journal_tool(["journal", "reset", "--force"], 0)
         self.fs.data_scan(["init"])
         self.fs.data_scan(["scan_extents", self.fs.get_data_pool_name()], worker_count=workers)
         self.fs.data_scan(["scan_inodes", self.fs.get_data_pool_name()], worker_count=workers)

qa/tasks/cephfs/test_flush.py

@@ -44,7 +44,7 @@ def test_flush(self):
 
         # ...and the journal is truncated to just a single subtreemap from the
         # newly created segment
-        summary_output = self.fs.journal_tool(["event", "get", "summary"])
+        summary_output = self.fs.journal_tool(["event", "get", "summary"], 0)
         try:
             self.assertEqual(summary_output,
                              dedent(
@@ -72,7 +72,7 @@ def test_flush(self):
                              ).strip())
             flush_data = self.fs.mds_asok(["flush", "journal"])
             self.assertEqual(flush_data['return_code'], 0)
-            self.assertEqual(self.fs.journal_tool(["event", "get", "summary"]),
+            self.assertEqual(self.fs.journal_tool(["event", "get", "summary"], 0),
                              dedent(
                                  """
                                  Events by type:

qa/tasks/cephfs/test_forward_scrub.py

@@ -242,10 +242,10 @@ def test_inotable_sync(self):
         # is all that will be in the InoTable in memory)
 
         self.fs.journal_tool(["event", "splice",
-            "--inode={0}".format(inos["./file2_sixmegs"]), "summary"])
+                              "--inode={0}".format(inos["./file2_sixmegs"]), "summary"], 0)
 
         self.fs.journal_tool(["event", "splice",
-            "--inode={0}".format(inos["./file3_sixmegs"]), "summary"])
+                              "--inode={0}".format(inos["./file3_sixmegs"]), "summary"], 0)
 
         # Revert to old inotable.
         for key, value in inotable_copy.iteritems():

qa/tasks/cephfs/test_journal_migration.py

@@ -82,13 +82,14 @@ def replay_names():
             ))
 
         # Verify that cephfs-journal-tool can now read the rewritten journal
-        inspect_out = self.fs.journal_tool(["journal", "inspect"])
+        inspect_out = self.fs.journal_tool(["journal", "inspect"], 0)
         if not inspect_out.endswith(": OK"):
             raise RuntimeError("Unexpected journal-tool result: '{0}'".format(
                 inspect_out
             ))
 
-        self.fs.journal_tool(["event", "get", "json", "--path", "/tmp/journal.json"])
+        self.fs.journal_tool(["event", "get", "json",
+                              "--path", "/tmp/journal.json"], 0)
         p = self.fs.tool_remote.run(
             args=[
                 "python",

qa/tasks/cephfs/test_journal_repair.py

@@ -77,7 +77,7 @@ def test_inject_to_empty(self):
         self.assertEqual(self.fs.list_dirfrag(ROOT_INO), [])
 
         # Execute the dentry recovery, this should populate the backing store
-        self.fs.journal_tool(['event', 'recover_dentries', 'list'])
+        self.fs.journal_tool(['event', 'recover_dentries', 'list'], 0)
 
         # Dentries in ROOT_INO are present
         self.assertEqual(sorted(self.fs.list_dirfrag(ROOT_INO)), sorted(['rootfile_head', 'subdir_head', 'linkdir_head']))
@@ -87,7 +87,7 @@ def test_inject_to_empty(self):
 
         # Now check the MDS can read what we wrote: truncate the journal
         # and start the mds.
-        self.fs.journal_tool(['journal', 'reset'])
+        self.fs.journal_tool(['journal', 'reset'], 0)
         self.fs.mds_fail_restart()
         self.fs.wait_for_daemons()
 
@@ -265,10 +265,10 @@ def get_state():
         self.fs.mds_stop(active_mds_names[0])
         self.fs.mds_fail(active_mds_names[0])
         # Invoke recover_dentries quietly, because otherwise log spews millions of lines
-        self.fs.journal_tool(["event", "recover_dentries", "summary"], rank=0, quiet=True)
-        self.fs.journal_tool(["event", "recover_dentries", "summary"], rank=1, quiet=True)
+        self.fs.journal_tool(["event", "recover_dentries", "summary"], 0, quiet=True)
+        self.fs.journal_tool(["event", "recover_dentries", "summary"], 1, quiet=True)
         self.fs.table_tool(["0", "reset", "session"])
-        self.fs.journal_tool(["journal", "reset"], rank=0)
+        self.fs.journal_tool(["journal", "reset"], 0)
         self.fs.erase_mds_objects(1)
         self.fs.mon_manager.raw_cluster_cmd('fs', 'reset', self.fs.name,
                 '--yes-i-really-mean-it')

qa/tasks/cephfs/test_recovery_pool.py

@@ -149,7 +149,7 @@ def _rebuild_metadata(self, workload, other_pool=None, workers=1):
         if False:
             with self.assertRaises(CommandFailedError):
                 # Normal reset should fail when no objects are present, we'll use --force instead
-                self.fs.journal_tool(["journal", "reset"])
+                self.fs.journal_tool(["journal", "reset"], 0)
 
         self.fs.mds_stop()
         self.fs.data_scan(['scan_extents', '--alternate-pool',
@@ -159,22 +159,18 @@ def _rebuild_metadata(self, workload, other_pool=None, workers=1):
                            recovery_pool, '--filesystem', self.fs.name,
                            '--force-corrupt', '--force-init',
                            self.fs.get_data_pool_name()])
-        self.fs.journal_tool(['--rank=' + self.fs.name + ":0", 'event',
-                              'recover_dentries', 'list',
-                              '--alternate-pool', recovery_pool])
+        self.fs.journal_tool(['event', 'recover_dentries', 'list',
+                              '--alternate-pool', recovery_pool], 0)
 
         self.fs.data_scan(['init', '--force-init', '--filesystem',
                            self.fs.name])
         self.fs.data_scan(['scan_inodes', '--filesystem', self.fs.name,
                            '--force-corrupt', '--force-init',
                            self.fs.get_data_pool_name()])
-        self.fs.journal_tool(['--rank=' + self.fs.name + ":0", 'event',
-                              'recover_dentries', 'list'])
+        self.fs.journal_tool(['event', 'recover_dentries', 'list'], 0)
 
-        self.fs.journal_tool(['--rank=' + recovery_fs + ":0", 'journal',
-                              'reset', '--force'])
-        self.fs.journal_tool(['--rank=' + self.fs.name + ":0", 'journal',
-                              'reset', '--force'])
+        self.recovery_fs.journal_tool(['journal', 'reset', '--force'], 0)
+        self.fs.journal_tool(['journal', 'reset', '--force'], 0)
         self.fs.mon_manager.raw_cluster_cmd('mds', 'repaired',
                                             recovery_fs + ":0")
 

qa/workunits/suites/cephfs_journal_tool_smoke.sh

@@ -3,7 +3,7 @@
 set -e
 set -x
 
-export BIN="${BIN:-cephfs-journal-tool}"
+export BIN="${BIN:-cephfs-journal-tool --rank=cephfs:0}"
 export JOURNAL_FILE=/tmp/journal.bin
 export JSON_OUTPUT=/tmp/json.tmp
 export BINARY_OUTPUT=/tmp/binary.tmp

src/tools/cephfs/JournalTool.cc

@@ -62,9 +62,7 @@ void JournalTool::usage()
     << "    <output>: [summary|list|binary|json] [--path <path>]\n"
     << "\n"
     << "General options:\n"
-    << "  --rank=filesystem:mds-rank  Journal rank (required if multiple\n"
-    << "                              file systems, default is rank 0 on\n"
-    << "                              the only filesystem otherwise.)\n"
+    << "  --rank=filesystem:mds-rank|all Journal rank (mandatory)\n"
     << "  --journal=<mdlog|purge_queue>  Journal type (purge_queue means\n"
     << "                                 this journal is used to queue for purge operation,\n"
     << "                                 default is mdlog, and only mdlog support event mode)\n" 
@@ -96,9 +94,8 @@ int JournalTool::main(std::vector<const char*> &argv)
 
   std::string rank_str;
   if (!ceph_argparse_witharg(argv, arg, &rank_str, "--rank", (char*)NULL)) {
-    // Default: act on rank 0.  Will give the user an error if they
-    // try invoking this way when they have more than one filesystem.
-    rank_str = "0";
+    derr << "missing mandatory \"--rank\" argument" << dendl;
+    return -EINVAL;
   }
 
   if (!ceph_argparse_witharg(argv, arg, &type, "--journal", (char*)NULL)) {
@@ -112,7 +109,7 @@ int JournalTool::main(std::vector<const char*> &argv)
     return r;
   }
 
-  r = role_selector.parse(*fsmap, rank_str);
+  r = role_selector.parse(*fsmap, rank_str, false);
   if (r != 0) {
     derr << "Couldn't determine MDS rank." << dendl;
     return r;
@@ -161,15 +158,27 @@ int JournalTool::main(std::vector<const char*> &argv)
   // =========
   // journal and header are general journal mode
   // event mode is only specific for mdlog
-  for (auto role : role_selector.get_roles()) {
+  auto roles = role_selector.get_roles();
+  if (roles.size() > 1) {
+    const std::string &command = argv[0];
+    bool allowed = can_execute_for_all_ranks(mode, command);
+    if (!allowed) {
+      derr << "operation not allowed for all ranks" << dendl;
+      return -EINVAL;
+    }
+
+    all_ranks = true;
+  }
+  for (auto role : roles) {
     rank = role.rank;
+    std::vector<const char *> rank_argv(argv);
     dout(4) << "Executing for rank " << rank << dendl;
     if (mode == std::string("journal")) {
-      r = main_journal(argv);
+      r = main_journal(rank_argv);
     } else if (mode == std::string("header")) {
-      r = main_header(argv);
+      r = main_header(rank_argv);
     } else if (mode == std::string("event")) {
-      r = main_event(argv);
+      r = main_event(rank_argv);
     } else {
       cerr << "Bad command '" << mode << "'" << std::endl;
       return -EINVAL;
@@ -191,6 +200,23 @@ int JournalTool::validate_type(const std::string &type)
   return -1;
 }
 
+std::string JournalTool::gen_dump_file_path(const std::string &prefix) {
+  if (!all_ranks) {
+    return prefix;
+  }
+
+  return prefix + "." + std::to_string(rank);
+}
+
+bool JournalTool::can_execute_for_all_ranks(const std::string &mode,
+                                            const std::string &command) {
+  if (mode == "journal" && command == "import") {
+    return false;
+  }
+
+  return true;
+}
+
 /**
  * Handle arguments for 'journal' mode
  *
@@ -397,6 +423,8 @@ int JournalTool::main_event(std::vector<const char*> &argv)
     }
   }
 
+  const std::string dump_path = gen_dump_file_path(output_path);
+
   // Execute command
   // ===============
   JournalScanner js(input, rank, type, filter);
@@ -513,7 +541,7 @@ int JournalTool::main_event(std::vector<const char*> &argv)
 
   // Generate output
   // ===============
-  EventOutput output(js, output_path);
+  EventOutput output(js, dump_path);
   int output_result = 0;
   if (output_style == "binary") {
       output_result = output.binary();
@@ -600,7 +628,8 @@ int JournalTool::journal_export(std::string const &path, bool import, bool force
     if (import) {
       r = dumper.undump(path.c_str(), force);
     } else {
-      r = dumper.dump(path.c_str());
+      const std::string ex_path = gen_dump_file_path(path);
+      r = dumper.dump(ex_path.c_str());
     }
   }
 

src/tools/cephfs/JournalTool.h

@@ -37,6 +37,8 @@ class JournalTool : public MDSUtility
     // Bit hacky, use this `rank` member to control behaviour of the
     // various main_ functions.
     mds_rank_t rank;
+    // when set, generate per rank dump file path
+    bool all_ranks = false;
 
     std::string type;
 
@@ -82,6 +84,14 @@ class JournalTool : public MDSUtility
 
     //validate type
     int validate_type(const std::string &type);
+
+    // generate output file path for dump/export
+    std::string gen_dump_file_path(const std::string &prefix);
+
+    // check if an operation (mode, command) is safe to be
+    // executed on all ranks.
+    bool can_execute_for_all_ranks(const std::string &mode,
+                                   const std::string &command);
   public:
     static void usage();
     JournalTool() :

src/tools/cephfs/RoleSelector.cc

@@ -29,13 +29,14 @@ int MDSRoleSelector::parse_rank(
   }
 }
 
-int MDSRoleSelector::parse(const FSMap &fsmap, std::string const &str)
+int MDSRoleSelector::parse(const FSMap &fsmap, std::string const &str,
+                           bool allow_unqualified_rank)
 {
   auto colon_pos = str.find(":");
   if (colon_pos == std::string::npos) {
     // An unqualified rank.  Only valid if there is only one
     // namespace.
-    if (fsmap.filesystem_count() == 1) {
+    if (fsmap.filesystem_count() == 1 && allow_unqualified_rank) {
       fscid = fsmap.get_filesystem()->fscid;
       return parse_rank(fsmap, str);
     } else {

src/tools/cephfs/RoleSelector.h

@@ -15,7 +15,8 @@ class MDSRoleSelector
 {
   public:
     const std::vector<mds_role_t> &get_roles() const {return roles;}
-    int parse(const FSMap &fsmap, std::string const &str);
+    int parse(const FSMap &fsmap, std::string const &str,
+            bool allow_unqualified_rank=true);
     MDSRoleSelector()
       : fscid(FS_CLUSTER_ID_NONE)
     {}