Merge PR ceph#54485 into main

* refs/pull/54485/head:
	mds/quiesce-db: keep the db thread alive until shutdown
	mds/quiesce-db: incorporate review comments
	mds/quiesce: declare QuiesceDbPeerListing and QuiesceDbPeerAck
	mds/quiesce: resolve the quiesce cluster at the mds monitor
	include/types: add an I/O helper for std::unordered_map
	messages: avoid using mutable members in MMDSQuiesce*
	mds/quiesce-db: incorporate review comments
	doc/cephfs/fs-volumes: doc fixes and updates
	pybind/mgr: correct type hints for `get_quiesce_leader_info`
	mds/quiesce: only use ACTIVE daemons for the quiesce cluster
	mds,messages: quiesce db inter-rank messaging
	mds/quiesce: MDSRankQuiesce - integration of the quiesce db manager
	doc/cephfs/fs-volumes: Add info about the quiesce command
	doc: fixes for local dev builds
	mgr/volumes: support for `fs subvolume quiesce`
	mgr/volumes: use `volume_exception_to_retval` as a decorator
	pybind/mgr: add a `one-shot` parameter to send_command
	mds/quiesce: QuiesceAgent implementation and unit tests
	mds/quiesce: QuiesceDb.h and QuiesceDbManager with tests
	common/Timer.cc: improve debug messages from the timer_thread
	mds: MDSRank.cc: return status from `send_message_mds`
	encoding: add emplace variants for map dencoders
	common/Cond: make C_SaferCond private members protected to facilitate inheritance
	qa/tasks/cephfs: give the tests more time to run heavy fs workloads

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Patrick Donnelly <[email protected]>

Author: batrick

doc/.gitignore

@@ -1,2 +1,3 @@
 /overview.png
 /object_store.png
+_build/

doc/_ext/ceph_commands.py

@@ -177,7 +177,7 @@ def parse_cmd(cmd):
 
     @staticmethod
     def parse_args(args):
-        return [Sig._parse_arg_desc(arg) for arg in args.split()]
+        return [Sig._parse_arg_desc(arg) for arg in args]
 
 
 TEMPLATE = '''
@@ -358,8 +358,9 @@ def run(self):
         cmds = sorted(cmds, key=lambda cmd: cmd.prefix)
         self._render_cmds(cmds)
 
-        orig_rgw_mod = sys.modules['pybind_rgw_mod']
-        sys.modules['rgw'] = orig_rgw_mod
+        if 'pybind_rgw_mod' in sys.modules:
+            orig_rgw_mod = sys.modules['pybind_rgw_mod']
+            sys.modules['rgw'] = orig_rgw_mod
 
         return []
 

doc/cephfs/fs-volumes.rst

@@ -13,9 +13,13 @@
         os.path.dirname(
             os.path.abspath(__file__)))
 
-pybind_rgw_mod = __import__('rgw', globals(), locals(), [], 0)
-sys.modules['pybind_rgw_mod'] = pybind_rgw_mod
-
+# it could be that ceph was built without RGW support
+# e.g. in a local development environment
+try:
+    pybind_rgw_mod = __import__('rgw', globals(), locals(), [], 0)
+    sys.modules['pybind_rgw_mod'] = pybind_rgw_mod
+except Exception:
+    pass
 
 def parse_ceph_release():
     with open(os.path.join(top_level, 'src/ceph_release')) as f:

doc/cephfs/quiesce-set-states.svg

@@ -761,17 +761,18 @@ def create_destroy(self):
             'rm', '-f', os.path.join(self.hostfs_mntpt, filename)
         ])
 
-    def _run_python(self, pyscript, py_version='python3', sudo=False):
+    def _run_python(self, pyscript, py_version='python3', sudo=False, timeout=None):
         args, omit_sudo = [], True
         if sudo:
             args.append('sudo')
             omit_sudo = False
-        args += ['stdin-killer', '--', py_version, '-c', pyscript]
+        timeout_args = ['--timeout', "%d" % timeout] if timeout is not None else []
+        args += ['stdin-killer', *timeout_args, '--', py_version, '-c', pyscript]
         return self.client_remote.run(args=args, wait=False, stdin=run.PIPE,
                                       stdout=StringIO(), omit_sudo=omit_sudo)
 
-    def run_python(self, pyscript, py_version='python3', sudo=False):
-        p = self._run_python(pyscript, py_version, sudo=sudo)
+    def run_python(self, pyscript, py_version='python3', sudo=False, timeout=None):
+        p = self._run_python(pyscript, py_version, sudo=sudo, timeout=timeout)
         p.wait()
         return p.stdout.getvalue().strip()
 
@@ -1243,7 +1244,7 @@ def write_test_pattern(self, filename, size):
             size=size
         )))
 
-    def validate_test_pattern(self, filename, size):
+    def validate_test_pattern(self, filename, size, timeout=None):
         log.info("Validating {0} bytes from {1}".format(size, filename))
         # Use sudo because cephfs-data-scan may recreate the file with owner==root
         return self.run_python(dedent("""
@@ -1262,7 +1263,7 @@ def validate_test_pattern(self, filename, size):
         """.format(
             path=os.path.join(self.hostfs_mntpt, filename),
             size=size
-        )), sudo=True)
+        )), sudo=True, timeout=timeout)
 
     def open_n_background(self, fs_path, count):
         """

doc/conf.py

@@ -376,7 +376,7 @@ def test_multimds_mksnap(self):
         self.mount_a.write_test_pattern("d0/d1/file_a", 8 * 1024 * 1024)
         self.mount_a.run_shell(["mkdir", "d0/.snap/s1"])
         self.mount_a.run_shell(["rm", "-f", "d0/d1/file_a"])
-        self.mount_a.validate_test_pattern("d0/.snap/s1/d1/file_a", 8 * 1024 * 1024)
+        self.mount_a.validate_test_pattern("d0/.snap/s1/d1/file_a", 8 * 1024 * 1024, timeout=20)
 
         self.mount_a.run_shell(["rmdir", "d0/.snap/s1"])
         self.mount_a.run_shell(["rm", "-rf", "d0"])

qa/tasks/cephfs/mount.py

@@ -63,7 +63,9 @@ class HealthTest(DashboardTestCase):
             'balance_automate': bool,
         }),
         'ever_allowed_features': int,
-        'root': int
+        'root': int,
+        'qdb_leader': int,
+        'qdb_cluster': JList(int)
     })
 
     def test_minimal_health(self):

qa/tasks/cephfs/test_snapshots.py

@@ -3028,15 +3028,10 @@ void Client::handle_fs_map_user(const MConstRef<MFSMapUser>& m)
 // Cancel all the commands for missing or laggy GIDs
 void Client::cancel_commands(const MDSMap& newmap)
 {
-  std::vector<ceph_tid_t> cancel_ops;
-
-  std::scoped_lock cmd_lock(command_lock);
-  auto &commands = command_table.get_commands();
-  for (const auto &[tid, op] : commands) {
+  cancel_commands_if([=, this](MDSCommandOp const& op) {
     const mds_gid_t op_mds_gid = op.mds_gid;
     if (newmap.is_dne_gid(op_mds_gid) || newmap.is_laggy_gid(op_mds_gid)) {
-      ldout(cct, 1) << __func__ << ": cancelling command op " << tid << dendl;
-      cancel_ops.push_back(tid);
+      ldout(cct, 1) << "cancel_commands: cancelling command op " << op.tid << dendl;
       if (op.outs) {
         std::ostringstream ss;
         ss << "MDS " << op_mds_gid << " went away";
@@ -3048,13 +3043,10 @@ void Client::cancel_commands(const MDSMap& newmap)
        * has its own lock.
        */
       op.con->mark_down();
-      if (op.on_finish)
-        op.on_finish->complete(-CEPHFS_ETIMEDOUT);
+      return -CEPHFS_ETIMEDOUT;
     }
-  }
-
-  for (const auto &tid : cancel_ops)
-    command_table.erase(tid);
+    return 0;
+  });
 }
 
 void Client::handle_mds_map(const MConstRef<MMDSMap>& m)
@@ -6416,7 +6408,8 @@ int Client::mds_command(
     const bufferlist& inbl,
     bufferlist *outbl,
     string *outs,
-    Context *onfinish)
+    Context *onfinish,
+    bool one_shot)
 {
   RWRef_t iref_reader(initialize_state, CLIENT_INITIALIZED);
   if (!iref_reader.is_state_satisfied())
@@ -6475,6 +6468,9 @@ int Client::mds_command(
 
     // Open a connection to the target MDS
     ConnectionRef conn = messenger->connect_to_mds(info.get_addrs());
+    if (one_shot) {
+      conn->send_keepalive();
+    }
 
     cl.unlock();
     {
@@ -6489,6 +6485,7 @@ int Client::mds_command(
       op.inbl = inbl;
       op.mds_gid = target_gid;
       op.con = conn;
+      op.one_shot = one_shot;
 
       ldout(cct, 4) << __func__ << ": new command op to " << target_gid
         << " tid=" << op.tid << " multi_id=" << op.multi_target_id << " "<< cmd << dendl;
@@ -16682,13 +16679,41 @@ void Client::ms_handle_connect(Connection *con)
 bool Client::ms_handle_reset(Connection *con)
 {
   ldout(cct, 0) << __func__ << " on " << con->get_peer_addr() << dendl;
+
+  cancel_commands_if([=, this](MDSCommandOp const& op) {
+    if (op.one_shot && op.con.get() == con) {
+      ldout(cct, 1) << "ms_handle_reset: aborting one-shot command op " << op.tid << dendl;
+      if (op.outs) {
+        std::ostringstream ss;
+        ss << "MDS connection reset";
+        *(op.outs) = ss.str();
+      }
+      return -EPIPE;
+    }
+    return 0;
+  });
+
   return false;
 }
 
 void Client::ms_handle_remote_reset(Connection *con)
 {
-  std::scoped_lock lock(client_lock);
   ldout(cct, 0) << __func__ << " on " << con->get_peer_addr() << dendl;
+
+  cancel_commands_if([=, this](MDSCommandOp const& op) {
+    if (op.one_shot && op.con.get() == con) {
+      ldout(cct, 1) << "ms_handle_remote_reset: aborting one-shot command op " << op.tid << dendl;
+      if (op.outs) {
+        std::ostringstream ss;
+        ss << "MDS remote session reset";
+        *(op.outs) = ss.str();
+      }
+      return -EPIPE;
+    }
+    return 0;
+  });
+
+  std::scoped_lock lock(client_lock);
   switch (con->get_peer_type()) {
   case CEPH_ENTITY_TYPE_MDS:
     {
@@ -16747,6 +16772,20 @@ void Client::ms_handle_remote_reset(Connection *con)
 bool Client::ms_handle_refused(Connection *con)
 {
   ldout(cct, 1) << __func__ << " on " << con->get_peer_addr() << dendl;
+
+  cancel_commands_if([=, this](MDSCommandOp const& op) {
+    if (op.one_shot && op.con.get() == con) {
+      ldout(cct, 1) << "ms_handle_refused: aborting one-shot command op " << op.tid << dendl;
+      if (op.outs) {
+        std::ostringstream ss;
+        ss << "MDS connection refused";
+        *(op.outs) = ss.str();
+      }
+      return -EPIPE;
+    }
+    return 0;
+  });
+
   return false;
 }
 

qa/tasks/mgr/dashboard/test_health.py

@@ -96,6 +96,7 @@ class MDSCommandOp : public CommandOp
 {
   public:
   mds_gid_t     mds_gid;
+  bool          one_shot = false;
 
   explicit MDSCommandOp(ceph_tid_t t) : CommandOp(t) {}
   explicit MDSCommandOp(ceph_tid_t t, ceph_tid_t multi_id) : CommandOp(t, multi_id) {}
@@ -335,7 +336,7 @@ class Client : public Dispatcher, public md_config_obs_t {
     const std::string &mds_spec,
     const std::vector<std::string>& cmd,
     const bufferlist& inbl,
-    bufferlist *poutbl, std::string *prs, Context *onfinish);
+    bufferlist *poutbl, std::string *prs, Context *onfinish, bool one_shot = false);
 
   // these should (more or less) mirror the actual system calls.
   int statfs(const char *path, struct statvfs *stbuf, const UserPerm& perms);
@@ -714,6 +715,27 @@ class Client : public Dispatcher, public md_config_obs_t {
   virtual void shutdown();
 
   // messaging
+  int cancel_commands_if(std::regular_invocable<MDSCommandOp const&> auto && error_for_op)
+  {
+    std::vector<ceph_tid_t> cancel_ops;
+
+    std::scoped_lock cmd_lock(command_lock);
+    auto& commands = command_table.get_commands();
+    for (const auto &[tid, op]: commands) {
+      int rc = static_cast<int>(error_for_op(op));
+      if (rc) {
+        cancel_ops.push_back(tid);
+        if (op.on_finish)
+          op.on_finish->complete(rc);
+      }
+    }
+
+    for (const auto& tid : cancel_ops)
+      command_table.erase(tid);
+
+    return cancel_ops.size();
+  }
+
   void cancel_commands(const MDSMap& newmap);
   void handle_mds_map(const MConstRef<MMDSMap>& m);
   void handle_fs_map(const MConstRef<MFSMap>& m);