Merge PR ceph#55421 into main

* refs/pull/55421/head:
	qa/cephfs: add test to verify backtrace update failure on deleted data pool
	mds: batch backtrace updates by pool-id when expiring a log segment
	mds: dump log segment in segment expiry callback
	mds: dump log segment end along with offset

Reviewed-by: Xiubo Li <[email protected]>
Reviewed-by: Patrick Donnelly <[email protected]>

Author: vshankar

qa/tasks/cephfs/test_backtrace.py

@@ -100,3 +100,29 @@ def test_backtrace(self):
         # we don't update the layout in all the old pools whenever it changes
         old_pool_layout = self.fs.read_layout(file_ino, pool=old_data_pool_name)
         self.assertEqual(old_pool_layout['object_size'], 4194304)
+
+    def test_backtrace_flush_on_deleted_data_pool(self):
+        """
+        that the MDS does not go read-only when handling backtrace update errors
+        when backtrace updates are batched and flushed to RADOS (during journal trim)
+        and some of the pool have been removed.
+        """
+        data_pool = self.fs.get_data_pool_name()
+        extra_data_pool_name_1 = data_pool + '_extra1'
+        self.fs.add_data_pool(extra_data_pool_name_1)
+
+        self.mount_a.run_shell(["mkdir", "dir_x"])
+        self.mount_a.setfattr("dir_x", "ceph.dir.layout.pool", extra_data_pool_name_1)
+        self.mount_a.run_shell(["touch", "dir_x/file_x"])
+        self.fs.flush()
+
+        extra_data_pool_name_2 = data_pool + '_extra2'
+        self.fs.add_data_pool(extra_data_pool_name_2)
+        self.mount_a.setfattr("dir_x/file_x", "ceph.file.layout.pool", extra_data_pool_name_2)
+        self.mount_a.run_shell(["setfattr", "-x", "ceph.dir.layout", "dir_x"])
+        self.run_ceph_cmd("fs", "rm_data_pool", self.fs.name, extra_data_pool_name_1)
+        self.fs.flush()
+
+        # quick test to check if the mds has handled backtrace update failure
+        # on the deleted data pool without going read-only.
+        self.mount_a.run_shell(["mkdir", "dir_y"])

src/mds/CInode.cc

@@ -1386,7 +1386,7 @@ void CInode::_commit_ops(int r, C_GatherBuilder &gather_bld,
 }
 
 void CInode::_store_backtrace(std::vector<CInodeCommitOperation> &ops_vec,
-                              inode_backtrace_t &bt, int op_prio)
+                              inode_backtrace_t &bt, int op_prio, bool ignore_old_pools)
 {
   dout(10) << __func__ << " on " << *this << dendl;
   ceph_assert(is_dirty_parent());
@@ -1407,8 +1407,8 @@ void CInode::_store_backtrace(std::vector<CInodeCommitOperation> &ops_vec,
   ops_vec.emplace_back(op_prio, pool, get_inode()->layout,
                        mdcache->mds->mdsmap->get_up_features(), slink);
 
-  if (!state_test(STATE_DIRTYPOOL) || get_inode()->old_pools.empty()) {
-    dout(20) << __func__ << ": no dirtypool or no old pools" << dendl;
+  if (!state_test(STATE_DIRTYPOOL) || get_inode()->old_pools.empty() || ignore_old_pools) {
+    dout(20) << __func__ << ": no dirtypool or no old pools or ignore_old_pools" << dendl;
     return;
   }
 
@@ -1431,7 +1431,7 @@ void CInode::store_backtrace(MDSContext *fin, int op_prio)
   inode_backtrace_t bt;
   auto version = get_inode()->backtrace_version;
 
-  _store_backtrace(ops_vec, bt, op_prio);
+  _store_backtrace(ops_vec, bt, op_prio, false);
 
   C_GatherBuilder gather(g_ceph_context,
 			 new C_OnFinisher(
@@ -1442,12 +1442,14 @@ void CInode::store_backtrace(MDSContext *fin, int op_prio)
   gather.activate();
 }
 
-void CInode::store_backtrace(CInodeCommitOperations &op, int op_prio)
+void CInode::store_backtrace(CInodeCommitOperations &op, int op_prio,
+			     bool ignore_old_pools)
 {
   op.version = get_inode()->backtrace_version;
   op.in = this;
 
-  _store_backtrace(op.ops_vec, op.bt, op_prio);
+  // update backtraces in old pools
+  _store_backtrace(op.ops_vec, op.bt, op_prio, ignore_old_pools);
 }
 
 void CInode::_stored_backtrace(int r, version_t v, Context *fin)

src/mds/CInode.h

@@ -754,8 +754,9 @@ class CInode : public MDSCacheObject, public InodeStoreBase, public Counter<CIno
                    inode_backtrace_t &bt);
   void build_backtrace(int64_t pool, inode_backtrace_t& bt);
   void _store_backtrace(std::vector<CInodeCommitOperation> &ops_vec,
-                        inode_backtrace_t &bt, int op_prio);
-  void store_backtrace(CInodeCommitOperations &op, int op_prio);
+                        inode_backtrace_t &bt, int op_prio, bool ignore_old_pools);
+  void store_backtrace(CInodeCommitOperations &op, int op_prio,
+		       bool ignore_old_pools=false);
   void store_backtrace(MDSContext *fin, int op_prio=-1);
   void _stored_backtrace(int r, version_t v, Context *fin);
   void fetch_backtrace(Context *fin, ceph::buffer::list *backtrace);
@@ -1142,6 +1143,14 @@ class CInode : public MDSCacheObject, public InodeStoreBase, public Counter<CIno
   // client caps
   client_t loner_cap = -1, want_loner_cap = -1;
 
+  /**
+   * Return the pool ID where we currently write backtraces for
+   * this inode (in addition to inode.old_pools)
+   *
+   * @returns a pool ID >=0
+   */
+  int64_t get_backtrace_pool() const;
+
 protected:
   ceph_lock_state_t *get_fcntl_lock_state() {
     if (!fcntl_locks)
@@ -1192,14 +1201,6 @@ class CInode : public MDSCacheObject, public InodeStoreBase, public Counter<CIno
       clear_flock_lock_state();
   }
 
-  /**
-   * Return the pool ID where we currently write backtraces for
-   * this inode (in addition to inode.old_pools)
-   *
-   * @returns a pool ID >=0
-   */
-  int64_t get_backtrace_pool() const;
-
   // parent dentries in cache
   CDentry         *parent = nullptr;             // primary link
   mempool::mds_co::compact_set<CDentry*>    remote_parents;     // if hard linked

src/mds/LogSegment.h

@@ -108,7 +108,7 @@ class LogSegment {
 
 static inline std::ostream& operator<<(std::ostream& out, const LogSegment& ls) {
   return out << "LogSegment(" << ls.seq << "/0x" << std::hex << ls.offset
-             << std::dec << " events=" << ls.num_events << ")";
+             << "~" << ls.end << std::dec << " events=" << ls.num_events << ")";
 }
 
 #endif

src/mds/MDLog.cc

@@ -749,6 +749,7 @@ class C_MaybeExpiredSegment : public MDSInternalContext {
   C_MaybeExpiredSegment(MDLog *mdl, LogSegment *s, int p) :
     MDSInternalContext(mdl->mds), mdlog(mdl), ls(s), op_prio(p) {}
   void finish(int res) override {
+    dout(10) << __func__ << ": ls=" << *ls << ", r=" << res << dendl;
     if (res < 0)
       mdlog->mds->handle_write_error(res);
     mdlog->_maybe_expired(ls, op_prio);

src/mds/journal.cc

@@ -237,27 +237,53 @@ void LogSegment::try_to_expire(MDSRank *mds, MDSGatherBuilder &gather_bld, int o
 
   ceph_assert(g_conf()->mds_kill_journal_expire_at != 3);
 
-  size_t count = 0;
-  for (elist<CInode*>::iterator it = dirty_parent_inodes.begin(); !it.end(); ++it)
-    count++;
-
-  std::vector<CInodeCommitOperations> ops_vec;
-  ops_vec.reserve(count);
+  std::map<int64_t, std::vector<CInodeCommitOperations>> ops_vec_map;
   // backtraces to be stored/updated
   for (elist<CInode*>::iterator p = dirty_parent_inodes.begin(); !p.end(); ++p) {
     CInode *in = *p;
     ceph_assert(in->is_auth());
     if (in->can_auth_pin()) {
       dout(15) << "try_to_expire waiting for storing backtrace on " << *in << dendl;
-      ops_vec.resize(ops_vec.size() + 1);
-      in->store_backtrace(ops_vec.back(), op_prio);
+      auto pool_id = in->get_backtrace_pool();
+
+      // this is for the default data pool
+      dout(20) << __func__ << ": updating pool=" << pool_id << dendl;
+      ops_vec_map[pool_id].push_back(CInodeCommitOperations());
+      in->store_backtrace(ops_vec_map[pool_id].back(), op_prio, true);
+
+
+      if (!in->state_test(CInode::STATE_DIRTYPOOL)) {
+	dout(20) << __func__ << ": no dirtypool" << dendl;
+	continue;
+      }
+
+      // dispatch separate ops for backtrace updates for old pools
+      for (auto _pool_id : in->get_inode()->old_pools) {
+	if (_pool_id == pool_id) {
+	  continue;
+	}
+
+	in->auth_pin(in); // CInode::_stored_backtrace() does auth_unpin()
+	dout(20) << __func__ << ": updating old_pool=" << _pool_id << dendl;
+
+	auto cco = CInodeCommitOperations();
+	cco.in = in;
+	// use backtrace from the main pool so as to pickup the main
+	// pool-id for old pool updates.
+	cco.bt = ops_vec_map[pool_id].back().bt;
+	cco.ops_vec.emplace_back(op_prio, _pool_id);
+	cco.version = in->get_inode()->backtrace_version;
+	ops_vec_map[_pool_id].push_back(cco);
+      }
     } else {
       dout(15) << "try_to_expire waiting for unfreeze on " << *in << dendl;
       in->add_waiter(CInode::WAIT_UNFREEZE, gather_bld.new_sub());
     }
   }
-  if (!ops_vec.empty())
+
+  for (auto& [pool_id, ops_vec] : ops_vec_map) {
     mds->finisher->queue(new BatchCommitBacktrace(mds, gather_bld.new_sub(), std::move(ops_vec)));
+  }
 
   ceph_assert(g_conf()->mds_kill_journal_expire_at != 4);