+++ /dev/null
-
-"""
-Test our tools for recovering metadata from the data pool
-"""
-import json
-
-import logging
-import os
-from textwrap import dedent
-import traceback
-from collections import namedtuple, defaultdict
-
-from teuthology.orchestra.run import CommandFailedError
-from tasks.cephfs.cephfs_test_case import CephFSTestCase, for_teuthology
-
-log = logging.getLogger(__name__)
-
-
-ValidationError = namedtuple("ValidationError", ["exception", "backtrace"])
-
-
-class Workload(object):
- def __init__(self, filesystem, mount):
- self._mount = mount
- self._filesystem = filesystem
- self._initial_state = None
-
- # Accumulate backtraces for every failed validation, and return them. Backtraces
- # are rather verbose, but we only see them when something breaks, and they
- # let us see which check failed without having to decorate each check with
- # a string
- self._errors = []
-
- def assert_equal(self, a, b):
- try:
- if a != b:
- raise AssertionError("{0} != {1}".format(a, b))
- except AssertionError as e:
- self._errors.append(
- ValidationError(e, traceback.format_exc(3))
- )
-
- def write(self):
- """
- Write the workload files to the mount
- """
- raise NotImplementedError()
-
- def validate(self):
- """
- Read from the mount and validate that the workload files are present (i.e. have
- survived or been reconstructed from the test scenario)
- """
- raise NotImplementedError()
-
- def damage(self):
- """
- Damage the filesystem pools in ways that will be interesting to recover from. By
- default just wipe everything in the metadata pool
- """
- # Delete every object in the metadata pool
- objects = self._filesystem.rados(["ls"]).split("\n")
- for o in objects:
- self._filesystem.rados(["rm", o])
-
- def flush(self):
- """
- Called after client unmount, after write: flush whatever you want
- """
- self._filesystem.mds_asok(["flush", "journal"])
-
-
-class SimpleWorkload(Workload):
- """
- Single file, single directory, check that it gets recovered and so does its size
- """
- 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 validate(self):
- self._mount.run_shell(["ls", "subdir"])
- st = self._mount.stat("subdir/sixmegs")
- self.assert_equal(st['st_size'], self._initial_state['st_size'])
- return self._errors
-
-
-class MovedFile(Workload):
- def write(self):
- # Create a file whose backtrace disagrees with his eventual position
- # in the metadata. We will see that he gets reconstructed in his
- # original position according to his backtrace.
- self._mount.run_shell(["mkdir", "subdir_alpha"])
- self._mount.run_shell(["mkdir", "subdir_bravo"])
- self._mount.write_n_mb("subdir_alpha/sixmegs", 6)
- self._filesystem.mds_asok(["flush", "journal"])
- self._mount.run_shell(["mv", "subdir_alpha/sixmegs", "subdir_bravo/sixmegs"])
- self._initial_state = self._mount.stat("subdir_bravo/sixmegs")
-
- def flush(self):
- pass
-
- def validate(self):
- self.assert_equal(self._mount.ls(), ["subdir_alpha"])
- st = self._mount.stat("subdir_alpha/sixmegs")
- self.assert_equal(st['st_size'], self._initial_state['st_size'])
- return self._errors
-
-
-class BacktracelessFile(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 validate(self):
- ino_name = "%x" % self._initial_state["st_ino"]
-
- # The inode should be linked into lost+found because we had no path for it
- self.assert_equal(self._mount.ls(), ["lost+found"])
- self.assert_equal(self._mount.ls("lost+found"), [ino_name])
- st = self._mount.stat("lost+found/{ino_name}".format(ino_name=ino_name))
-
- # We might not have got the name or path, but we should still get the size
- self.assert_equal(st['st_size'], self._initial_state['st_size'])
-
- return self._errors
-
-
-class StripedStashedLayout(Workload):
- def __init__(self, fs, m):
- super(StripedStashedLayout, self).__init__(fs, m)
-
- # Nice small stripes so we can quickly do our writes+validates
- self.sc = 4
- self.ss = 65536
- self.os = 262144
-
- self.interesting_sizes = [
- # Exactly stripe_count objects will exist
- self.os * self.sc,
- # Fewer than stripe_count objects will exist
- self.os * self.sc / 2,
- self.os * (self.sc - 1) + self.os / 2,
- self.os * (self.sc - 1) + self.os / 2 - 1,
- self.os * (self.sc + 1) + self.os / 2,
- self.os * (self.sc + 1) + self.os / 2 + 1,
- # More than stripe_count objects will exist
- self.os * self.sc + self.os * self.sc / 2
- ]
-
- def write(self):
- # Create a dir with a striped layout set on it
- self._mount.run_shell(["mkdir", "stripey"])
-
- self._mount.setfattr("./stripey", "ceph.dir.layout",
- "stripe_unit={ss} stripe_count={sc} object_size={os} pool={pool}".format(
- ss=self.ss, os=self.os, sc=self.sc,
- pool=self._filesystem.get_data_pool_name()
- ))
-
- # Write files, then flush metadata so that its layout gets written into an xattr
- for i, n_bytes in enumerate(self.interesting_sizes):
- self._mount.write_test_pattern("stripey/flushed_file_{0}".format(i), n_bytes)
- # This is really just validating the validator
- self._mount.validate_test_pattern("stripey/flushed_file_{0}".format(i), n_bytes)
- self._filesystem.mds_asok(["flush", "journal"])
-
- # Write another file in the same way, but this time don't flush the metadata,
- # so that it won't have the layout xattr
- self._mount.write_test_pattern("stripey/unflushed_file", 1024 * 512)
- self._mount.validate_test_pattern("stripey/unflushed_file", 1024 * 512)
-
- self._initial_state = {
- "unflushed_ino": self._mount.path_to_ino("stripey/unflushed_file")
- }
-
- def flush(self):
- # Pass because we already selectively flushed during write
- pass
-
- def validate(self):
- # The first files should have been recovered into its original location
- # with the correct layout: read back correct data
- for i, n_bytes in enumerate(self.interesting_sizes):
- try:
- self._mount.validate_test_pattern("stripey/flushed_file_{0}".format(i), n_bytes)
- except CommandFailedError as e:
- self._errors.append(
- ValidationError("File {0} (size {1}): {2}".format(i, n_bytes, e), traceback.format_exc(3))
- )
-
- # The unflushed file should have been recovered into lost+found without
- # the correct layout: read back junk
- ino_name = "%x" % self._initial_state["unflushed_ino"]
- self.assert_equal(self._mount.ls("lost+found"), [ino_name])
- try:
- self._mount.validate_test_pattern(os.path.join("lost+found", ino_name), 1024 * 512)
- except CommandFailedError:
- pass
- else:
- self._errors.append(
- ValidationError("Unexpectedly valid data in unflushed striped file", "")
- )
-
- return self._errors
-
-
-class ManyFilesWorkload(Workload):
- def __init__(self, filesystem, mount, file_count):
- super(ManyFilesWorkload, self).__init__(filesystem, mount)
- self.file_count = file_count
-
- def write(self):
- self._mount.run_shell(["mkdir", "subdir"])
- for n in range(0, self.file_count):
- self._mount.write_test_pattern("subdir/{0}".format(n), 6 * 1024 * 1024)
-
- def validate(self):
- for n in range(0, self.file_count):
- try:
- self._mount.validate_test_pattern("subdir/{0}".format(n), 6 * 1024 * 1024)
- except CommandFailedError as e:
- self._errors.append(
- ValidationError("File {0}: {1}".format(n, e), traceback.format_exc(3))
- )
-
- return self._errors
-
-
-class MovedDir(Workload):
- def write(self):
- # Create a nested dir that we will then move. Two files with two different
- # backtraces referring to the moved dir, claiming two different locations for
- # it. We will see that only one backtrace wins and the dir ends up with
- # single linkage.
- self._mount.run_shell(["mkdir", "-p", "grandmother/parent"])
- self._mount.write_n_mb("grandmother/parent/orig_pos_file", 1)
- self._filesystem.mds_asok(["flush", "journal"])
- self._mount.run_shell(["mkdir", "grandfather"])
- self._mount.run_shell(["mv", "grandmother/parent", "grandfather"])
- self._mount.write_n_mb("grandfather/parent/new_pos_file", 2)
- self._filesystem.mds_asok(["flush", "journal"])
-
- self._initial_state = (
- self._mount.stat("grandfather/parent/orig_pos_file"),
- self._mount.stat("grandfather/parent/new_pos_file")
- )
-
- def validate(self):
- root_files = self._mount.ls()
- self.assert_equal(len(root_files), 1)
- self.assert_equal(root_files[0] in ["grandfather", "grandmother"], True)
- winner = root_files[0]
- st_opf = self._mount.stat("{0}/parent/orig_pos_file".format(winner))
- st_npf = self._mount.stat("{0}/parent/new_pos_file".format(winner))
-
- self.assert_equal(st_opf['st_size'], self._initial_state[0]['st_size'])
- self.assert_equal(st_npf['st_size'], self._initial_state[1]['st_size'])
-
-
-class MissingZerothObject(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 damage(self):
- super(MissingZerothObject, self).damage()
- zeroth_id = "{0:x}.00000000".format(self._initial_state['st_ino'])
- self._filesystem.rados(["rm", zeroth_id], pool=self._filesystem.get_data_pool_name())
-
- def validate(self):
- st = self._mount.stat("lost+found/{0:x}".format(self._initial_state['st_ino']))
- self.assert_equal(st['st_size'], self._initial_state['st_size'])
-
-
-class NonDefaultLayout(Workload):
- """
- Check that the reconstruction copes with files that have a different
- object size in their layout
- """
- def write(self):
- self._mount.run_shell(["touch", "datafile"])
- self._mount.setfattr("./datafile", "ceph.file.layout.object_size", "8388608")
- self._mount.run_shell(["dd", "if=/dev/urandom", "of=./datafile", "bs=1M", "count=32"])
- self._initial_state = self._mount.stat("datafile")
-
- def validate(self):
- # Check we got the layout reconstructed properly
- object_size = int(self._mount.getfattr(
- "./datafile", "ceph.file.layout.object_size"))
- self.assert_equal(object_size, 8388608)
-
- # Check we got the file size reconstructed properly
- st = self._mount.stat("datafile")
- self.assert_equal(st['st_size'], self._initial_state['st_size'])
-
-
-class TestDataScan(CephFSTestCase):
- MDSS_REQUIRED = 2
-
- def is_marked_damaged(self, rank):
- mds_map = self.fs.get_mds_map()
- return rank in mds_map['damaged']
-
- def _rebuild_metadata(self, workload, workers=1):
- """
- That when all objects in metadata pool are removed, we can rebuild a metadata pool
- based on the contents of a data pool, and a client can see and read our files.
- """
-
- # First, inject some files
-
- workload.write()
-
- # Unmount the client and flush the journal: the tool should also cope with
- # situations where there is dirty metadata, but we'll test that separately
- self.mount_a.umount_wait()
- workload.flush()
-
- # Stop the MDS
- self.fs.mds_stop()
- self.fs.mds_fail()
-
- # After recovery, we need the MDS to not be strict about stats (in production these options
- # are off by default, but in QA we need to explicitly disable them)
- self.fs.set_ceph_conf('mds', 'mds verify scatter', False)
- self.fs.set_ceph_conf('mds', 'mds debug scatterstat', False)
-
- # Apply any data damage the workload wants
- workload.damage()
-
- # Reset the MDS map in case multiple ranks were in play: recovery procedure
- # only understands how to rebuild metadata under rank 0
- self.fs.mon_manager.raw_cluster_cmd('fs', 'reset', self.fs.name,
- '--yes-i-really-mean-it')
-
- self.fs.mds_restart()
-
- def get_state(mds_id):
- info = self.mds_cluster.get_mds_info(mds_id)
- return info['state'] if info is not None else None
-
- self.wait_until_true(lambda: self.is_marked_damaged(0), 60)
- for mds_id in self.fs.mds_ids:
- self.wait_until_equal(
- lambda: get_state(mds_id),
- "up:standby",
- timeout=60)
-
- self.fs.table_tool([self.fs.name + ":0", "reset", "session"])
- self.fs.table_tool([self.fs.name + ":0", "reset", "snap"])
- self.fs.table_tool([self.fs.name + ":0", "reset", "inode"])
-
- # Run the recovery procedure
- if False:
- with self.assertRaises(CommandFailedError):
- # Normal reset should fail when no objects are present, we'll use --force instead
- self.fs.journal_tool(["journal", "reset"])
-
- self.fs.journal_tool(["journal", "reset", "--force"])
- self.fs.data_scan(["init"])
- self.fs.data_scan(["scan_extents", self.fs.get_data_pool_name()], worker_count=workers)
- self.fs.data_scan(["scan_inodes", self.fs.get_data_pool_name()], worker_count=workers)
-
- # Mark the MDS repaired
- self.fs.mon_manager.raw_cluster_cmd('mds', 'repaired', '0')
-
- # Start the MDS
- self.fs.mds_restart()
- self.fs.wait_for_daemons()
- log.info(str(self.mds_cluster.status()))
-
- # Mount a client
- self.mount_a.mount()
- self.mount_a.wait_until_mounted()
-
- # See that the files are present and correct
- errors = workload.validate()
- if errors:
- log.error("Validation errors found: {0}".format(len(errors)))
- for e in errors:
- log.error(e.exception)
- log.error(e.backtrace)
- raise AssertionError("Validation failed, first error: {0}\n{1}".format(
- errors[0].exception, errors[0].backtrace
- ))
-
- def test_rebuild_simple(self):
- self._rebuild_metadata(SimpleWorkload(self.fs, self.mount_a))
-
- def test_rebuild_moved_file(self):
- self._rebuild_metadata(MovedFile(self.fs, self.mount_a))
-
- def test_rebuild_backtraceless(self):
- self._rebuild_metadata(BacktracelessFile(self.fs, self.mount_a))
-
- def test_rebuild_moved_dir(self):
- self._rebuild_metadata(MovedDir(self.fs, self.mount_a))
-
- def test_rebuild_missing_zeroth(self):
- self._rebuild_metadata(MissingZerothObject(self.fs, self.mount_a))
-
- def test_rebuild_nondefault_layout(self):
- self._rebuild_metadata(NonDefaultLayout(self.fs, self.mount_a))
-
- def test_stashed_layout(self):
- self._rebuild_metadata(StripedStashedLayout(self.fs, self.mount_a))
-
- def _dirfrag_keys(self, object_id):
- keys_str = self.fs.rados(["listomapkeys", object_id])
- if keys_str:
- return keys_str.split("\n")
- else:
- return []
-
- def test_fragmented_injection(self):
- """
- That when injecting a dentry into a fragmented directory, we put it in the right fragment.
- """
-
- self.fs.set_allow_dirfrags(True)
-
- file_count = 100
- file_names = ["%s" % n for n in range(0, file_count)]
-
- # Create a directory of `file_count` files, each named after its
- # decimal number and containing the string of its decimal number
- self.mount_a.run_python(dedent("""
- import os
- path = os.path.join("{path}", "subdir")
- os.mkdir(path)
- for n in range(0, {file_count}):
- open(os.path.join(path, "%s" % n), 'w').write("%s" % n)
- """.format(
- path=self.mount_a.mountpoint,
- file_count=file_count
- )))
-
- dir_ino = self.mount_a.path_to_ino("subdir")
-
- # Only one MDS should be active!
- self.assertEqual(len(self.fs.get_active_names()), 1)
-
- # Ensure that one directory is fragmented
- mds_id = self.fs.get_active_names()[0]
- self.fs.mds_asok(["dirfrag", "split", "/subdir", "0/0", "1"], mds_id)
-
- # Flush journal and stop MDS
- self.mount_a.umount_wait()
- self.fs.mds_asok(["flush", "journal"], mds_id)
- self.fs.mds_stop()
- self.fs.mds_fail()
-
- # Pick a dentry and wipe out its key
- # Because I did a 1 bit split, I know one frag will be named <inode>.01000000
- frag_obj_id = "{0:x}.01000000".format(dir_ino)
- keys = self._dirfrag_keys(frag_obj_id)
- victim_key = keys[7] # arbitrary choice
- log.info("victim_key={0}".format(victim_key))
- victim_dentry = victim_key.split("_head")[0]
- self.fs.rados(["rmomapkey", frag_obj_id, victim_key])
-
- # Start filesystem back up, observe that the file appears to be gone in an `ls`
- self.fs.mds_restart()
- self.fs.wait_for_daemons()
- self.mount_a.mount()
- self.mount_a.wait_until_mounted()
- files = self.mount_a.run_shell(["ls", "subdir/"]).stdout.getvalue().strip().split("\n")
- self.assertListEqual(sorted(files), sorted(list(set(file_names) - set([victim_dentry]))))
-
- # Stop the filesystem
- self.mount_a.umount_wait()
- self.fs.mds_stop()
- self.fs.mds_fail()
-
- # Run data-scan, observe that it inserts our dentry back into the correct fragment
- # by checking the omap now has the dentry's key again
- self.fs.data_scan(["scan_extents", self.fs.get_data_pool_name()])
- self.fs.data_scan(["scan_inodes", self.fs.get_data_pool_name()])
- self.assertIn(victim_key, self._dirfrag_keys(frag_obj_id))
-
- # Start the filesystem and check that the dentry we deleted is now once again visible
- # and points to the correct file data.
- self.fs.mds_restart()
- self.fs.wait_for_daemons()
- self.mount_a.mount()
- self.mount_a.wait_until_mounted()
- out = self.mount_a.run_shell(["cat", "subdir/{0}".format(victim_dentry)]).stdout.getvalue().strip()
- self.assertEqual(out, victim_dentry)
-
- # Finally, close the loop by checking our injected dentry survives a merge
- mds_id = self.fs.get_active_names()[0]
- self.mount_a.ls("subdir") # Do an ls to ensure both frags are in cache so the merge will work
- self.fs.mds_asok(["dirfrag", "merge", "/subdir", "0/0"], mds_id)
- self.fs.mds_asok(["flush", "journal"], mds_id)
- frag_obj_id = "{0:x}.00000000".format(dir_ino)
- keys = self._dirfrag_keys(frag_obj_id)
- self.assertListEqual(sorted(keys), sorted(["%s_head" % f for f in file_names]))
-
- @for_teuthology
- def test_parallel_execution(self):
- self._rebuild_metadata(ManyFilesWorkload(self.fs, self.mount_a, 25), workers=7)
-
- def test_pg_files(self):
- """
- That the pg files command tells us which files are associated with
- a particular PG
- """
- file_count = 20
- self.mount_a.run_shell(["mkdir", "mydir"])
- self.mount_a.create_n_files("mydir/myfile", file_count)
-
- # Some files elsewhere in the system that we will ignore
- # to check that the tool is filtering properly
- self.mount_a.run_shell(["mkdir", "otherdir"])
- self.mount_a.create_n_files("otherdir/otherfile", file_count)
-
- pgs_to_files = defaultdict(list)
- # Rough (slow) reimplementation of the logic
- for i in range(0, file_count):
- file_path = "mydir/myfile_{0}".format(i)
- ino = self.mount_a.path_to_ino(file_path)
- obj = "{0:x}.{1:08x}".format(ino, 0)
- pgid = json.loads(self.fs.mon_manager.raw_cluster_cmd(
- "osd", "map", self.fs.get_data_pool_name(), obj,
- "--format=json-pretty"
- ))['pgid']
- pgs_to_files[pgid].append(file_path)
- log.info("{0}: {1}".format(file_path, pgid))
-
- pg_count = self.fs.get_pgs_per_fs_pool()
- for pg_n in range(0, pg_count):
- pg_str = "{0}.{1}".format(self.fs.get_data_pool_id(), pg_n)
- out = self.fs.data_scan(["pg_files", "mydir", pg_str])
- lines = [l for l in out.split("\n") if l]
- log.info("{0}: {1}".format(pg_str, lines))
- self.assertSetEqual(set(lines), set(pgs_to_files[pg_str]))
-
- def test_scan_links(self):
- """
- The scan_links command fixes linkage errors
- """
- self.mount_a.run_shell(["mkdir", "testdir1"])
- self.mount_a.run_shell(["mkdir", "testdir2"])
- dir1_ino = self.mount_a.path_to_ino("testdir1")
- dir2_ino = self.mount_a.path_to_ino("testdir2")
- dirfrag1_oid = "{0:x}.00000000".format(dir1_ino)
- dirfrag2_oid = "{0:x}.00000000".format(dir2_ino)
-
- self.mount_a.run_shell(["touch", "testdir1/file1"])
- self.mount_a.run_shell(["ln", "testdir1/file1", "testdir1/link1"])
- self.mount_a.run_shell(["ln", "testdir1/file1", "testdir2/link2"])
-
- mds_id = self.fs.get_active_names()[0]
- self.fs.mds_asok(["flush", "journal"], mds_id)
-
- dirfrag1_keys = self._dirfrag_keys(dirfrag1_oid)
-
- # introduce duplicated primary link
- file1_key = "file1_head"
- self.assertIn(file1_key, dirfrag1_keys)
- file1_omap_data = self.fs.rados(["getomapval", dirfrag1_oid, file1_key, '-'])
- self.fs.rados(["setomapval", dirfrag2_oid, file1_key], stdin_data=file1_omap_data)
- self.assertIn(file1_key, self._dirfrag_keys(dirfrag2_oid))
-
- # remove a remote link, make inode link count incorrect
- link1_key = 'link1_head'
- self.assertIn(link1_key, dirfrag1_keys)
- self.fs.rados(["rmomapkey", dirfrag1_oid, link1_key])
-
- # increase good primary link's version
- self.mount_a.run_shell(["touch", "testdir1/file1"])
- self.mount_a.umount_wait()
-
- self.fs.mds_asok(["flush", "journal"], mds_id)
- self.fs.mds_stop()
- self.fs.mds_fail()
-
- # repair linkage errors
- self.fs.data_scan(["scan_links"])
-
- # primary link in testdir2 was deleted?
- self.assertNotIn(file1_key, self._dirfrag_keys(dirfrag2_oid))
-
- self.fs.mds_restart()
- self.fs.wait_for_daemons()
-
- self.mount_a.mount()
- self.mount_a.wait_until_mounted()
-
- # link count was adjusted?
- file1_nlink = self.mount_a.path_to_nlink("testdir1/file1")
- self.assertEqual(file1_nlink, 2)
Code Review / stor4nfv.git / blobdiff