ceph/qa/tasks/cephfs/test_damage.py

   1 from io import BytesIO, StringIO
   2 import json
   3 import logging
   4 import errno
   5 import re
   6 import time
   7 from teuthology.contextutil import MaxWhileTries
   8 from teuthology.exceptions import CommandFailedError
   9 from teuthology.orchestra.run import wait
  10 from tasks.cephfs.fuse_mount import FuseMount
  11 from tasks.cephfs.cephfs_test_case import CephFSTestCase, for_teuthology
  12
  13 DAMAGED_ON_START = "damaged_on_start"
  14 DAMAGED_ON_LS = "damaged_on_ls"
  15 CRASHED = "server crashed"
  16 NO_DAMAGE = "no damage"
  17 READONLY = "readonly"
  18 FAILED_CLIENT = "client failed"
  19 FAILED_SERVER = "server failed"
  20
  21 # An EIO in response to a stat from the client
  22 EIO_ON_LS = "eio"
  23
  24 # An EIO, but nothing in damage table (not ever what we expect)
  25 EIO_NO_DAMAGE = "eio without damage entry"
  26
  27
  28 log = logging.getLogger(__name__)
  29
  30
  31 class TestDamage(CephFSTestCase):
  32     def _simple_workload_write(self):
  33         self.mount_a.run_shell(["mkdir", "subdir"])
  34         self.mount_a.write_n_mb("subdir/sixmegs", 6)
  35         return self.mount_a.stat("subdir/sixmegs")
  36
  37     def is_marked_damaged(self, rank):
  38         mds_map = self.fs.get_mds_map()
  39         return rank in mds_map['damaged']
  40
  41     @for_teuthology #459s
  42     def test_object_deletion(self):
  43         """
  44         That the MDS has a clean 'damaged' response to loss of any single metadata object
  45         """
  46
  47         self._simple_workload_write()
  48
  49         # Hmm, actually it would be nice to permute whether the metadata pool
  50         # state contains sessions or not, but for the moment close this session
  51         # to avoid waiting through reconnect on every MDS start.
  52         self.mount_a.umount_wait()
  53         for mds_name in self.fs.get_active_names():
  54             self.fs.mds_asok(["flush", "journal"], mds_name)
  55
  56         self.fs.fail()
  57
  58         serialized = self.fs.radosmo(['export', '-'])
  59
  60         def is_ignored(obj_id, dentry=None):
  61             """
  62             A filter to avoid redundantly mutating many similar objects (e.g.
  63             stray dirfrags) or similar dentries (e.g. stray dir dentries)
  64             """
  65             if re.match("60.\.00000000", obj_id) and obj_id != "600.00000000":
  66                 return True
  67
  68             if dentry and obj_id == "100.00000000":
  69                 if re.match("stray.+_head", dentry) and dentry != "stray0_head":
  70                     return True
  71
  72             return False
  73
  74         def get_path(obj_id, dentry=None):
  75             """
  76             What filesystem path does this object or dentry correspond to?   i.e.
  77             what should I poke to see EIO after damaging it?
  78             """
  79
  80             if obj_id == "1.00000000" and dentry == "subdir_head":
  81                 return "./subdir"
  82             elif obj_id == "10000000000.00000000" and dentry == "sixmegs_head":
  83                 return "./subdir/sixmegs"
  84
  85             # None means ls will do an "ls -R" in hope of seeing some errors
  86             return None
  87
  88         objects = self.fs.radosmo(["ls"], stdout=StringIO()).strip().split("\n")
  89         objects = [o for o in objects if not is_ignored(o)]
  90
  91         # Find all objects with an OMAP header
  92         omap_header_objs = []
  93         for o in objects:
  94             header = self.fs.radosmo(["getomapheader", o], stdout=StringIO())
  95             # The rados CLI wraps the header output in a hex-printed style
  96             header_bytes = int(re.match("header \((.+) bytes\)", header).group(1))
  97             if header_bytes > 0:
  98                 omap_header_objs.append(o)
  99
 100         # Find all OMAP key/vals
 101         omap_keys = []
 102         for o in objects:
 103             keys_str = self.fs.radosmo(["listomapkeys", o], stdout=StringIO())
 104             if keys_str:
 105                 for key in keys_str.strip().split("\n"):
 106                     if not is_ignored(o, key):
 107                         omap_keys.append((o, key))
 108
 109         # Find objects that have data in their bodies
 110         data_objects = []
 111         for obj_id in objects:
 112             stat_out = self.fs.radosmo(["stat", obj_id], stdout=StringIO())
 113             size = int(re.match(".+, size (.+)$", stat_out).group(1))
 114             if size > 0:
 115                 data_objects.append(obj_id)
 116
 117         # Define the various forms of damage we will inflict
 118         class MetadataMutation(object):
 119             def __init__(self, obj_id_, desc_, mutate_fn_, expectation_, ls_path=None):
 120                 self.obj_id = obj_id_
 121                 self.desc = desc_
 122                 self.mutate_fn = mutate_fn_
 123                 self.expectation = expectation_
 124                 if ls_path is None:
 125                     self.ls_path = "."
 126                 else:
 127                     self.ls_path = ls_path
 128
 129             def __eq__(self, other):
 130                 return self.desc == other.desc
 131
 132             def __hash__(self):
 133                 return hash(self.desc)
 134
 135         junk = "deadbeef" * 10
 136         mutations = []
 137
 138         # Removals
 139         for o in objects:
 140             if o in [
 141                 # JournalPointers are auto-replaced if missing (same path as upgrade)
 142                 "400.00000000",
 143                 # Missing dirfrags for non-system dirs result in empty directory
 144                 "10000000000.00000000",
 145                 # PurgeQueue is auto-created if not found on startup
 146                 "500.00000000",
 147                 # open file table is auto-created if not found on startup
 148                 "mds0_openfiles.0"
 149             ]:
 150                 expectation = NO_DAMAGE
 151             else:
 152                 expectation = DAMAGED_ON_START
 153
 154             log.info("Expectation on rm '{0}' will be '{1}'".format(
 155                 o, expectation
 156             ))
 157
 158             mutations.append(MetadataMutation(
 159                 o,
 160                 "Delete {0}".format(o),
 161                 lambda o=o: self.fs.radosm(["rm", o]),
 162                 expectation
 163             ))
 164
 165         # Blatant corruptions
 166         for obj_id in data_objects:
 167             if obj_id == "500.00000000":
 168                 # purge queue corruption results in read-only FS
 169                 mutations.append(MetadataMutation(
 170                     obj_id,
 171                     "Corrupt {0}".format(obj_id),
 172                     lambda o=obj_id: self.fs.radosm(["put", o, "-"], stdin=StringIO(junk)),
 173                     READONLY
 174                 ))
 175             else:
 176                 mutations.append(MetadataMutation(
 177                     obj_id,
 178                     "Corrupt {0}".format(obj_id),
 179                     lambda o=obj_id: self.fs.radosm(["put", o, "-"], stdin=StringIO(junk)),
 180                     DAMAGED_ON_START
 181                 ))
 182
 183         # Truncations
 184         for o in data_objects:
 185             if o == "500.00000000":
 186                 # The PurgeQueue is allowed to be empty: Journaler interprets
 187                 # an empty header object as an empty journal.
 188                 expectation = NO_DAMAGE
 189             else:
 190                 expectation = DAMAGED_ON_START
 191
 192             mutations.append(
 193                 MetadataMutation(
 194                     o,
 195                     "Truncate {0}".format(o),
 196                     lambda o=o: self.fs.radosm(["truncate", o, "0"]),
 197                     expectation
 198             ))
 199
 200         # OMAP value corruptions
 201         for o, k in omap_keys:
 202             if o.startswith("100."):
 203                 # Anything in rank 0's 'mydir'
 204                 expectation = DAMAGED_ON_START
 205             else:
 206                 expectation = EIO_ON_LS
 207
 208             mutations.append(
 209                 MetadataMutation(
 210                     o,
 211                     "Corrupt omap key {0}:{1}".format(o, k),
 212                     lambda o=o,k=k: self.fs.radosm(["setomapval", o, k, junk]),
 213                     expectation,
 214                     get_path(o, k)
 215                 )
 216             )
 217
 218         # OMAP header corruptions
 219         for o in omap_header_objs:
 220             if re.match("60.\.00000000", o) \
 221                     or o in ["1.00000000", "100.00000000", "mds0_sessionmap"]:
 222                 expectation = DAMAGED_ON_START
 223             else:
 224                 expectation = NO_DAMAGE
 225
 226             log.info("Expectation on corrupt header '{0}' will be '{1}'".format(
 227                 o, expectation
 228             ))
 229
 230             mutations.append(
 231                 MetadataMutation(
 232                     o,
 233                     "Corrupt omap header on {0}".format(o),
 234                     lambda o=o: self.fs.radosm(["setomapheader", o, junk]),
 235                     expectation
 236                 )
 237             )
 238
 239         results = {}
 240
 241         for mutation in mutations:
 242             log.info("Applying mutation '{0}'".format(mutation.desc))
 243
 244             # Reset MDS state
 245             self.mount_a.umount_wait(force=True)
 246             self.fs.fail()
 247             self.fs.mon_manager.raw_cluster_cmd('mds', 'repaired', '0')
 248
 249             # Reset RADOS pool state
 250             self.fs.radosm(['import', '-'], stdin=BytesIO(serialized))
 251
 252             # Inject the mutation
 253             mutation.mutate_fn()
 254
 255             # Try starting the MDS
 256             self.fs.set_joinable()
 257
 258             # How long we'll wait between starting a daemon and expecting
 259             # it to make it through startup, and potentially declare itself
 260             # damaged to the mon cluster.
 261             startup_timeout = 60
 262
 263             if mutation.expectation not in (EIO_ON_LS, DAMAGED_ON_LS, NO_DAMAGE):
 264                 if mutation.expectation == DAMAGED_ON_START:
 265                     # The MDS may pass through active before making it to damaged
 266                     try:
 267                         self.wait_until_true(lambda: self.is_marked_damaged(0), startup_timeout)
 268                     except RuntimeError:
 269                         pass
 270
 271                 # Wait for MDS to either come up or go into damaged state
 272                 try:
 273                     self.wait_until_true(lambda: self.is_marked_damaged(0) or self.fs.are_daemons_healthy(), startup_timeout)
 274                 except RuntimeError:
 275                     crashed = False
 276                     # Didn't make it to healthy or damaged, did it crash?
 277                     for daemon_id, daemon in self.fs.mds_daemons.items():
 278                         if daemon.proc and daemon.proc.finished:
 279                             crashed = True
 280                             log.error("Daemon {0} crashed!".format(daemon_id))
 281                             daemon.proc = None  # So that subsequent stop() doesn't raise error
 282                     if not crashed:
 283                         # Didn't go health, didn't go damaged, didn't crash, so what?
 284                         raise
 285                     else:
 286                         log.info("Result: Mutation '{0}' led to crash".format(mutation.desc))
 287                         results[mutation] = CRASHED
 288                         continue
 289                 if self.is_marked_damaged(0):
 290                     log.info("Result: Mutation '{0}' led to DAMAGED state".format(mutation.desc))
 291                     results[mutation] = DAMAGED_ON_START
 292                     continue
 293                 else:
 294                     log.info("Mutation '{0}' did not prevent MDS startup, attempting ls...".format(mutation.desc))
 295             else:
 296                 try:
 297                     self.wait_until_true(self.fs.are_daemons_healthy, 60)
 298                 except RuntimeError:
 299                     log.info("Result: Mutation '{0}' should have left us healthy, actually not.".format(mutation.desc))
 300                     if self.is_marked_damaged(0):
 301                         results[mutation] = DAMAGED_ON_START
 302                     else:
 303                         results[mutation] = FAILED_SERVER
 304                     continue
 305                 log.info("Daemons came up after mutation '{0}', proceeding to ls".format(mutation.desc))
 306
 307             # MDS is up, should go damaged on ls or client mount
 308             self.mount_a.mount_wait()
 309             if mutation.ls_path == ".":
 310                 proc = self.mount_a.run_shell(["ls", "-R", mutation.ls_path], wait=False)
 311             else:
 312                 proc = self.mount_a.stat(mutation.ls_path, wait=False)
 313
 314             if mutation.expectation == DAMAGED_ON_LS:
 315                 try:
 316                     self.wait_until_true(lambda: self.is_marked_damaged(0), 60)
 317                     log.info("Result: Mutation '{0}' led to DAMAGED state after ls".format(mutation.desc))
 318                     results[mutation] = DAMAGED_ON_LS
 319                 except RuntimeError:
 320                     if self.fs.are_daemons_healthy():
 321                         log.error("Result: Failed to go damaged on mutation '{0}', actually went active".format(
 322                             mutation.desc))
 323                         results[mutation] = NO_DAMAGE
 324                     else:
 325                         log.error("Result: Failed to go damaged on mutation '{0}'".format(mutation.desc))
 326                         results[mutation] = FAILED_SERVER
 327             elif mutation.expectation == READONLY:
 328                 proc = self.mount_a.run_shell(["mkdir", "foo"], wait=False)
 329                 try:
 330                     proc.wait()
 331                 except CommandFailedError:
 332                     stderr = proc.stderr.getvalue()
 333                     log.info(stderr)
 334                     if "Read-only file system".lower() in stderr.lower():
 335                         pass
 336                     else:
 337                         raise
 338             else:
 339                 try:
 340                     wait([proc], 20)
 341                     log.info("Result: Mutation '{0}' did not caused DAMAGED state".format(mutation.desc))
 342                     results[mutation] = NO_DAMAGE
 343                 except MaxWhileTries:
 344                     log.info("Result: Failed to complete client IO on mutation '{0}'".format(mutation.desc))
 345                     results[mutation] = FAILED_CLIENT
 346                 except CommandFailedError as e:
 347                     if e.exitstatus == errno.EIO:
 348                         log.info("Result: EIO on client")
 349                         results[mutation] = EIO_ON_LS
 350                     else:
 351                         log.info("Result: unexpected error {0} on client".format(e))
 352                         results[mutation] = FAILED_CLIENT
 353
 354             if mutation.expectation == EIO_ON_LS:
 355                 # EIOs mean something handled by DamageTable: assert that it has
 356                 # been populated
 357                 damage = json.loads(
 358                     self.fs.mon_manager.raw_cluster_cmd(
 359                         'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]), "damage", "ls", '--format=json-pretty'))
 360                 if len(damage) == 0:
 361                     results[mutation] = EIO_NO_DAMAGE
 362
 363         failures = [(mutation, result) for (mutation, result) in results.items() if mutation.expectation != result]
 364         if failures:
 365             log.error("{0} mutations had unexpected outcomes:".format(len(failures)))
 366             for mutation, result in failures:
 367                 log.error("  Expected '{0}' actually '{1}' from '{2}'".format(
 368                     mutation.expectation, result, mutation.desc
 369                 ))
 370             raise RuntimeError("{0} mutations had unexpected outcomes".format(len(failures)))
 371         else:
 372             log.info("All {0} mutations had expected outcomes".format(len(mutations)))
 373
 374     def test_damaged_dentry(self):
 375         # Damage to dentrys is interesting because it leaves the
 376         # directory's `complete` flag in a subtle state where
 377         # we have marked the dir complete in order that folks
 378         # can access it, but in actual fact there is a dentry
 379         # missing
 380         self.mount_a.run_shell(["mkdir", "subdir/"])
 381
 382         self.mount_a.run_shell(["touch", "subdir/file_undamaged"])
 383         self.mount_a.run_shell(["touch", "subdir/file_to_be_damaged"])
 384
 385         subdir_ino = self.mount_a.path_to_ino("subdir")
 386
 387         self.mount_a.umount_wait()
 388         for mds_name in self.fs.get_active_names():
 389             self.fs.mds_asok(["flush", "journal"], mds_name)
 390
 391         self.fs.fail()
 392
 393         # Corrupt a dentry
 394         junk = "deadbeef" * 10
 395         dirfrag_obj = "{0:x}.00000000".format(subdir_ino)
 396         self.fs.radosm(["setomapval", dirfrag_obj, "file_to_be_damaged_head", junk])
 397
 398         # Start up and try to list it
 399         self.fs.set_joinable()
 400         self.fs.wait_for_daemons()
 401
 402         self.mount_a.mount_wait()
 403         dentries = self.mount_a.ls("subdir/")
 404
 405         # The damaged guy should have disappeared
 406         self.assertEqual(dentries, ["file_undamaged"])
 407
 408         # I should get ENOENT if I try and read it normally, because
 409         # the dir is considered complete
 410         try:
 411             self.mount_a.stat("subdir/file_to_be_damaged", wait=True)
 412         except CommandFailedError as e:
 413             self.assertEqual(e.exitstatus, errno.ENOENT)
 414         else:
 415             raise AssertionError("Expected ENOENT")
 416
 417         # The fact that there is damaged should have bee recorded
 418         damage = json.loads(
 419             self.fs.mon_manager.raw_cluster_cmd(
 420                 'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 421                 "damage", "ls", '--format=json-pretty'))
 422         self.assertEqual(len(damage), 1)
 423         damage_id = damage[0]['id']
 424
 425         # If I try to create a dentry with the same name as the damaged guy
 426         # then that should be forbidden
 427         try:
 428             self.mount_a.touch("subdir/file_to_be_damaged")
 429         except CommandFailedError as e:
 430             self.assertEqual(e.exitstatus, errno.EIO)
 431         else:
 432             raise AssertionError("Expected EIO")
 433
 434         # Attempting that touch will clear the client's complete flag, now
 435         # when I stat it I'll get EIO instead of ENOENT
 436         try:
 437             self.mount_a.stat("subdir/file_to_be_damaged", wait=True)
 438         except CommandFailedError as e:
 439             if isinstance(self.mount_a, FuseMount):
 440                 self.assertEqual(e.exitstatus, errno.EIO)
 441             else:
 442                 # Old kernel client handles this case differently
 443                 self.assertIn(e.exitstatus, [errno.ENOENT, errno.EIO])
 444         else:
 445             raise AssertionError("Expected EIO")
 446
 447         nfiles = self.mount_a.getfattr("./subdir", "ceph.dir.files")
 448         self.assertEqual(nfiles, "2")
 449
 450         self.mount_a.umount_wait()
 451
 452         # Now repair the stats
 453         scrub_json = self.fs.run_scrub(["start", "/subdir", "repair"])
 454         log.info(json.dumps(scrub_json, indent=2))
 455
 456         self.assertNotEqual(scrub_json, None)
 457         self.assertEqual(scrub_json["return_code"], 0)
 458         self.assertEqual(self.fs.wait_until_scrub_complete(tag=scrub_json["scrub_tag"]), True)
 459
 460         # Check that the file count is now correct
 461         self.mount_a.mount_wait()
 462         nfiles = self.mount_a.getfattr("./subdir", "ceph.dir.files")
 463         self.assertEqual(nfiles, "1")
 464
 465         # Clean up the omap object
 466         self.fs.radosm(["setomapval", dirfrag_obj, "file_to_be_damaged_head", junk])
 467
 468         # Clean up the damagetable entry
 469         self.fs.mon_manager.raw_cluster_cmd(
 470             'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 471             "damage", "rm", "{did}".format(did=damage_id))
 472
 473         # Now I should be able to create a file with the same name as the
 474         # damaged guy if I want.
 475         self.mount_a.touch("subdir/file_to_be_damaged")
 476
 477     def test_open_ino_errors(self):
 478         """
 479         That errors encountered during opening inos are properly propagated
 480         """
 481
 482         self.mount_a.run_shell(["mkdir", "dir1"])
 483         self.mount_a.run_shell(["touch", "dir1/file1"])
 484         self.mount_a.run_shell(["mkdir", "dir2"])
 485         self.mount_a.run_shell(["touch", "dir2/file2"])
 486         self.mount_a.run_shell(["mkdir", "testdir"])
 487         self.mount_a.run_shell(["ln", "dir1/file1", "testdir/hardlink1"])
 488         self.mount_a.run_shell(["ln", "dir2/file2", "testdir/hardlink2"])
 489
 490         file1_ino = self.mount_a.path_to_ino("dir1/file1")
 491         file2_ino = self.mount_a.path_to_ino("dir2/file2")
 492         dir2_ino = self.mount_a.path_to_ino("dir2")
 493
 494         # Ensure everything is written to backing store
 495         self.mount_a.umount_wait()
 496         self.fs.mds_asok(["flush", "journal"])
 497
 498         # Drop everything from the MDS cache
 499         self.fs.fail()
 500         self.fs.journal_tool(['journal', 'reset'], 0)
 501         self.fs.set_joinable()
 502         self.fs.wait_for_daemons()
 503
 504         self.mount_a.mount_wait()
 505
 506         # Case 1: un-decodeable backtrace
 507
 508         # Validate that the backtrace is present and decodable
 509         self.fs.read_backtrace(file1_ino)
 510         # Go corrupt the backtrace of alpha/target (used for resolving
 511         # bravo/hardlink).
 512         self.fs._write_data_xattr(file1_ino, "parent", "rhubarb")
 513
 514         # Check that touching the hardlink gives EIO
 515         ran = self.mount_a.run_shell(["stat", "testdir/hardlink1"], wait=False)
 516         try:
 517             ran.wait()
 518         except CommandFailedError:
 519             self.assertTrue("Input/output error" in ran.stderr.getvalue())
 520
 521         # Check that an entry is created in the damage table
 522         damage = json.loads(
 523             self.fs.mon_manager.raw_cluster_cmd(
 524                 'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 525                 "damage", "ls", '--format=json-pretty'))
 526         self.assertEqual(len(damage), 1)
 527         self.assertEqual(damage[0]['damage_type'], "backtrace")
 528         self.assertEqual(damage[0]['ino'], file1_ino)
 529
 530         self.fs.mon_manager.raw_cluster_cmd(
 531             'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 532             "damage", "rm", str(damage[0]['id']))
 533
 534
 535         # Case 2: missing dirfrag for the target inode
 536
 537         self.fs.radosm(["rm", "{0:x}.00000000".format(dir2_ino)])
 538
 539         # Check that touching the hardlink gives EIO
 540         ran = self.mount_a.run_shell(["stat", "testdir/hardlink2"], wait=False)
 541         try:
 542             ran.wait()
 543         except CommandFailedError:
 544             self.assertTrue("Input/output error" in ran.stderr.getvalue())
 545
 546         # Check that an entry is created in the damage table
 547         damage = json.loads(
 548             self.fs.mon_manager.raw_cluster_cmd(
 549                 'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 550                 "damage", "ls", '--format=json-pretty'))
 551         self.assertEqual(len(damage), 2)
 552         if damage[0]['damage_type'] == "backtrace" :
 553             self.assertEqual(damage[0]['ino'], file2_ino)
 554             self.assertEqual(damage[1]['damage_type'], "dir_frag")
 555             self.assertEqual(damage[1]['ino'], dir2_ino)
 556         else:
 557             self.assertEqual(damage[0]['damage_type'], "dir_frag")
 558             self.assertEqual(damage[0]['ino'], dir2_ino)
 559             self.assertEqual(damage[1]['damage_type'], "backtrace")
 560             self.assertEqual(damage[1]['ino'], file2_ino)
 561
 562         for entry in damage:
 563             self.fs.mon_manager.raw_cluster_cmd(
 564                 'tell', 'mds.{0}'.format(self.fs.get_active_names()[0]),
 565                 "damage", "rm", str(entry['id']))
 566
 567     def test_dentry_first_existing(self):
 568         """
 569         That the MDS won't abort when the dentry is already known to be damaged.
 570         """
 571
 572         def verify_corrupt():
 573             info = self.fs.read_cache("/a", 0)
 574             log.debug('%s', info)
 575             self.assertEqual(len(info), 1)
 576             dirfrags = info[0]['dirfrags']
 577             self.assertEqual(len(dirfrags), 1)
 578             dentries = dirfrags[0]['dentries']
 579             self.assertEqual([dn['path'] for dn in dentries if dn['is_primary']], ['a/c'])
 580             self.assertEqual(dentries[0]['snap_first'], 18446744073709551606) # SNAP_HEAD
 581
 582         self.mount_a.run_shell_payload("mkdir -p a/b")
 583         self.fs.flush()
 584         self.config_set("mds", "mds_abort_on_newly_corrupt_dentry", False)
 585         self.config_set("mds", "mds_inject_rename_corrupt_dentry_first", "1.0")
 586         time.sleep(5) # for conf to percolate
 587         self.mount_a.run_shell_payload("mv a/b a/c; sync .")
 588         self.mount_a.umount()
 589         verify_corrupt()
 590         self.fs.fail()
 591         self.config_rm("mds", "mds_inject_rename_corrupt_dentry_first")
 592         self.config_set("mds", "mds_abort_on_newly_corrupt_dentry", False)
 593         self.fs.set_joinable()
 594         status = self.fs.status()
 595         self.fs.flush()
 596         self.assertFalse(self.fs.status().hadfailover(status))
 597         verify_corrupt()
 598
 599     def test_dentry_first_preflush(self):
 600         """
 601         That the MDS won't write a dentry with new damage to CDentry::first
 602         to the journal.
 603         """
 604
 605         rank0 = self.fs.get_rank()
 606         self.fs.rank_freeze(True, rank=0)
 607         self.mount_a.run_shell_payload("mkdir -p a/{b,c}/d")
 608         self.fs.flush()
 609         self.config_set("mds", "mds_inject_rename_corrupt_dentry_first", "1.0")
 610         time.sleep(5) # for conf to percolate
 611         p = self.mount_a.run_shell_payload("timeout 60 mv a/b a/z", wait=False)
 612         self.wait_until_true(lambda: "laggy_since" in self.fs.get_rank(), timeout=self.fs.beacon_timeout)
 613         self.config_rm("mds", "mds_inject_rename_corrupt_dentry_first")
 614         self.fs.rank_freeze(False, rank=0)
 615         self.delete_mds_coredump(rank0['name'])
 616         self.fs.mds_restart(rank0['name'])
 617         self.fs.wait_for_daemons()
 618         p.wait()
 619         self.mount_a.run_shell_payload("stat a/ && find a/")
 620         self.fs.flush()
 621
 622     def test_dentry_first_precommit(self):
 623         """
 624         That the MDS won't write a dentry with new damage to CDentry::first
 625         to the directory object.
 626         """
 627
 628         fscid = self.fs.id
 629         self.mount_a.run_shell_payload("mkdir -p a/{b,c}/d; sync .")
 630         self.mount_a.umount() # allow immediate scatter write back
 631         self.fs.flush()
 632         # now just twiddle some inode metadata on a regular file
 633         self.mount_a.mount_wait()
 634         self.mount_a.run_shell_payload("chmod 711 a/b/d; sync .")
 635         self.mount_a.umount() # avoid journaling session related things
 636         # okay, now cause the dentry to get damaged after loading from the journal
 637         self.fs.fail()
 638         self.config_set("mds", "mds_inject_journal_corrupt_dentry_first", "1.0")
 639         time.sleep(5) # for conf to percolate
 640         self.fs.set_joinable()
 641         self.fs.wait_for_daemons()
 642         rank0 = self.fs.get_rank()
 643         self.fs.rank_freeze(True, rank=0)
 644         # so now we want to trigger commit but this will crash, so:
 645         c = ['--connect-timeout=60', 'tell', f"mds.{fscid}:0", "flush", "journal"]
 646         p = self.ceph_cluster.mon_manager.run_cluster_cmd(args=c, wait=False, timeoutcmd=30)
 647         self.wait_until_true(lambda: "laggy_since" in self.fs.get_rank(), timeout=self.fs.beacon_timeout)
 648         self.config_rm("mds", "mds_inject_journal_corrupt_dentry_first")
 649         self.fs.rank_freeze(False, rank=0)
 650         self.delete_mds_coredump(rank0['name'])
 651         self.fs.mds_restart(rank0['name'])
 652         self.fs.wait_for_daemons()
 653         try:
 654             p.wait()
 655         except CommandFailedError as e:
 656             print(e)
 657         else:
 658             self.fail("flush journal should fail!")
 659         self.mount_a.mount_wait()
 660         self.mount_a.run_shell_payload("stat a/ && find a/")
 661         self.fs.flush()