Merge PR ceph#61555 into main

* refs/pull/61555/head:
	mds: do not path traverse a damaged dirfrag
	qa: test file create on damaged directory

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Christopher Hoffman <[email protected]>
Reviewed-by: Viacheslav Dubeyko <[email protected]>

Author: batrick

.githubmap

@@ -191,3 +191,4 @@ ffilz Frank S. Filz <[email protected]>
 Jayaprakash-ibm Jaya Prakash Madaka <[email protected]>
 spuiuk Sachin Prabhu <[email protected]>
 anoopcs9 Anoop C S <[email protected]>
+dubeyko Viacheslav Dubeyko <[email protected]>

qa/tasks/cephfs/test_damage.py

@@ -662,3 +662,33 @@ def test_dentry_first_precommit(self):
         self.mount_a.mount_wait()
         self.mount_a.run_shell_payload("stat a/ && find a/")
         self.fs.flush()
+
+    def test_missing_dirfrag_create(self):
+        """
+        That the MDS won't let you manipulate dentries in a missing dirfrag.
+        """
+
+        self.mount_a.run_shell_payload("mkdir -p dir_x/dir_xx/dir_xxx/")
+        self.mount_a.run_shell_payload("touch dir_x/dir_xx/dir_xxx/file_y")
+        self.mount_a.umount()
+        d = self.fs.read_cache("dir_x/dir_xx", depth=0)
+        self.fs.flush()
+        time.sleep(5)
+        self.fs.flush() # EUpdate scatter_writebehind
+        self.fs.fail()
+        dirfrag_obj = "{0:x}.00000000".format(d[0]['ino'])
+        self.fs.radosm(["rm", dirfrag_obj]),
+        self.fs.set_joinable()
+        self.fs.wait_for_daemons()
+        self.mount_a.mount_wait()
+        try:
+            p = self.mount_a.run_shell_payload("touch dir_x/dir_xx/file_x", wait=False)
+            # CDir.cc: 1438: FAILED ceph_assert(get_version() < pv)
+            wait([p], 20)
+            self.fs.flush()
+        except MaxWhileTries:
+            self.fail("MDS probably crashed")
+        except CommandFailedError:
+            self.assertTrue("Input/output error" in p.stderr.getvalue())
+        else:
+            self.fail("command should fail")

src/mds/MDCache.cc

@@ -8452,6 +8452,13 @@ int MDCache::path_traverse(const MDRequestRef& mdr, MDSContextFactory& cf,
       }
     }
     ceph_assert(curdir);
+
+    if (mds->damage_table.is_dirfrag_damaged(curdir)) {
+      dout(4) << "traverse: stopped lookup at dirfrag "
+              << *curdir << "/" << path[depth] << " snap=" << snapid << dendl;
+      return -CEPHFS_EIO;
+    }
+
     if (pdir) {
       *pdir = curdir;
     }