mds: check for errors decoding backtraces

qa/tasks/cephfs/mount.py

@@ -136,7 +136,8 @@ def run_python(self, pyscript):
 
     def run_shell(self, args, wait=True):
         args = ["cd", self.mountpoint, run.Raw('&&'), "sudo"] + args
-        return self.client_remote.run(args=args, stdout=StringIO(), wait=wait)
+        return self.client_remote.run(args=args, stdout=StringIO(),
+                                      stderr=StringIO(), wait=wait)
 
     def open_no_data(self, basename):
         """

qa/tasks/cephfs/test_damage.py

@@ -447,3 +447,49 @@ def test_damaged_dentry(self):
         # Now I should be able to create a file with the same name as the
         # damaged guy if I want.
         self.mount_a.touch("subdir/file_to_be_damaged")
+
+    def test_corrupt_backtrace(self):
+        """
+        That an un-decodeable backtrace leads to an appropriate
+        error trying to follow the backtrace to the file.
+        """
+
+        self.mount_a.run_shell(["mkdir", "alpha"])
+        self.mount_a.run_shell(["mkdir", "bravo"])
+        self.mount_a.run_shell(["touch", "alpha/target"])
+        self.mount_a.run_shell(["ln", "alpha/target", "bravo/hardlink"])
+
+        alpha_ino = self.mount_a.path_to_ino("alpha/target")
+
+        # Ensure everything is written to backing store
+        self.mount_a.umount_wait()
+        self.fs.mds_asok(["flush", "journal"])
+
+        # Validate that the backtrace is present and decodable
+        self.fs.read_backtrace(alpha_ino)
+        # Go corrupt the backtrace of alpha/target (used for resolving
+        # bravo/hardlink).
+        self.fs._write_data_xattr(alpha_ino, "parent", "rhubarb")
+
+        # Drop everything from the MDS cache
+        self.mds_cluster.mds_stop()
+        self.fs.journal_tool(['journal', 'reset'])
+        self.mds_cluster.mds_fail_restart()
+        self.fs.wait_for_daemons()
+
+        # Check that touching the hardlink gives EIO
+        self.mount_a.mount()
+        ran = self.mount_a.run_shell(["ls", "-l", "bravo/hardlink"], wait=False)
+        try:
+            ran.wait()
+        except CommandFailedError:
+            self.assertTrue("Input/output error" in ran.stderr.getvalue())
+
+        # Check that an entry is created in the damage table
+        damage = json.loads(
+            self.fs.mon_manager.raw_cluster_cmd(
+                'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
+                "damage", "ls", '--format=json-pretty'))
+        self.assertEqual(len(damage), 1)
+        self.assertEqual(damage[0]['damage_type'], "backtrace")
+        self.assertEqual(damage[0]['ino'], alpha_ino)

src/mds/MDCache.cc

@@ -8319,7 +8319,14 @@ void MDCache::_open_ino_backtrace_fetched(inodeno_t ino, bufferlist& bl, int err
 
   inode_backtrace_t backtrace;
   if (err == 0) {
-    ::decode(backtrace, bl);
+    try {
+      ::decode(backtrace, bl);
+    } catch (const buffer::error &decode_exc) {
+      derr << "corrupt backtrace on ino x0" << std::hex << ino
+           << std::dec << ": " << decode_exc << dendl;
+      open_ino_finish(ino, info, -EIO);
+      return;
+    }
     if (backtrace.pool != info.pool && backtrace.pool != -1) {
       dout(10) << " old object in pool " << info.pool
 	       << ", retrying pool " << backtrace.pool << dendl;