Merge PR ceph#54031 into main

* refs/pull/54031/head:
	qa: add test to mangle lost+found directory object and ensure safety
	qa: run scrub before mounting client and validations

Reviewed-by: Kotresh Hiremath Ravishankar <[email protected]>

qa/tasks/cephfs/test_data_scan.py

@@ -101,6 +101,16 @@ def scrub(self):
         self.assert_equal(out_json["return_code"], 0)
         self.assert_equal(self._filesystem.wait_until_scrub_complete(tag=out_json["scrub_tag"]), True)
 
+    def mangle(self):
+        """
+        Gives an opportunity to fiddle with metadata objects before bringing back
+        the MDSs online. This is used in testing the lost+found case (when recovering
+        a file without a backtrace) to verify if lost+found directory object can be
+        removed via RADOS operation and the file system can be continued to be used
+        as expected.
+        """
+        pass
+
 class SimpleWorkload(Workload):
     """
     Single file, single directory, check that it gets recovered and so does its size
@@ -190,6 +200,23 @@ def validate(self):
 
         return self._errors
 
+class BacktracelessFileRemoveLostAndFoundDirectory(Workload):
+    def write(self):
+        self._mount.run_shell(["mkdir", "subdir"])
+        self._mount.write_n_mb("subdir/sixmegs", 6)
+        self._initial_state = self._mount.stat("subdir/sixmegs")
+
+    def flush(self):
+        # Never flush metadata, so backtrace won't be written
+        pass
+
+    def mangle(self):
+        self._filesystem.rados(["-p", self._filesystem.get_metadata_pool_name(), "rm", "4.00000000"])
+        self._filesystem.rados(["-p", self._filesystem.get_metadata_pool_name(), "rmomapkey", "1.00000000", "lost+found_head"])
+
+    def validate(self):
+        # The dir should be gone since we manually removed it
+        self.assert_not_equal(self._mount.ls(sudo=True), ["lost+found"])
 
 class StripedStashedLayout(Workload):
     def __init__(self, fs, m, pool=None):
@@ -427,6 +454,8 @@ def get_state(mds_id):
         self.fs.data_scan(["scan_inodes"], worker_count=workers)
         self.fs.data_scan(["scan_links"])
 
+        workload.mangle()
+
         # Mark the MDS repaired
         self.run_ceph_cmd('mds', 'repaired', '0')
 
@@ -435,13 +464,13 @@ def get_state(mds_id):
         self.fs.wait_for_daemons()
         log.info(str(self.mds_cluster.status()))
 
-        # Mount a client
-        self.mount_a.mount_wait()
-
         # run scrub as it is recommended post recovery for most
         # (if not all) recovery mechanisms.
         workload.scrub()
 
+        # Mount a client
+        self.mount_a.mount_wait()
+
         # See that the files are present and correct
         errors = workload.validate()
         if errors:
@@ -465,6 +494,9 @@ def test_rebuild_moved_file(self):
     def test_rebuild_backtraceless(self):
         self._rebuild_metadata(BacktracelessFile(self.fs, self.mount_a))
 
+    def test_rebuild_backtraceless_with_lf_dir_removed(self):
+        self._rebuild_metadata(BacktracelessFileRemoveLostAndFoundDirectory(self.fs, self.mount_a))
+
     def test_rebuild_moved_dir(self):
         self._rebuild_metadata(MovedDir(self.fs, self.mount_a))