import json import time import logging from textwrap import dedent import datetime import gevent from teuthology.orchestra.run import CommandFailedError, Raw from tasks.cephfs.cephfs_test_case import CephFSTestCase, for_teuthology log = logging.getLogger(__name__) class TestStrays(CephFSTestCase): MDSS_REQUIRED = 2 OPS_THROTTLE = 1 FILES_THROTTLE = 2 # Range of different file sizes used in throttle test's workload throttle_workload_size_range = 16 @for_teuthology def test_ops_throttle(self): self._test_throttling(self.OPS_THROTTLE) @for_teuthology def test_files_throttle(self): self._test_throttling(self.FILES_THROTTLE) def test_dir_deletion(self): """ That when deleting a bunch of dentries and the containing directory, everything gets purged. Catches cases where the client might e.g. fail to trim the unlinked dir from its cache. """ file_count = 1000 create_script = dedent(""" import os mount_path = "{mount_path}" subdir = "delete_me" size = {size} file_count = {file_count} os.mkdir(os.path.join(mount_path, subdir)) for i in xrange(0, file_count): filename = "{{0}}_{{1}}.bin".format(i, size) f = open(os.path.join(mount_path, subdir, filename), 'w') f.write(size * 'x') f.close() """.format( mount_path=self.mount_a.mountpoint, size=1024, file_count=file_count )) self.mount_a.run_python(create_script) # That the dirfrag object is created self.fs.mds_asok(["flush", "journal"]) dir_ino = self.mount_a.path_to_ino("delete_me") self.assertTrue(self.fs.dirfrag_exists(dir_ino, 0)) # Remove everything self.mount_a.run_shell(["rm", "-rf", "delete_me"]) self.fs.mds_asok(["flush", "journal"]) # That all the removed files get created as strays strays = self.get_mdc_stat("strays_created") self.assertEqual(strays, file_count + 1) # That the strays all get enqueued for purge self.wait_until_equal( lambda: self.get_mdc_stat("strays_enqueued"), strays, timeout=600 ) # That all the purge operations execute self.wait_until_equal( lambda: self.get_stat("purge_queue", "pq_executed"), strays, timeout=600 ) # That finally, the directory metadata object is gone self.assertFalse(self.fs.dirfrag_exists(dir_ino, 0)) # That finally, the data objects are all gone self.await_data_pool_empty() def _test_throttling(self, throttle_type): self.data_log = [] try: return self._do_test_throttling(throttle_type) except: for l in self.data_log: log.info(",".join([l_.__str__() for l_ in l])) raise def _do_test_throttling(self, throttle_type): """ That the mds_max_purge_ops setting is respected """ def set_throttles(files, ops): """ Helper for updating ops/files limits, and calculating effective ops_per_pg setting to give the same ops limit. """ self.set_conf('mds', 'mds_max_purge_files', "%d" % files) self.set_conf('mds', 'mds_max_purge_ops', "%d" % ops) pgs = self.fs.mon_manager.get_pool_property( self.fs.get_data_pool_name(), "pg_num" ) ops_per_pg = float(ops) / pgs self.set_conf('mds', 'mds_max_purge_ops_per_pg', "%s" % ops_per_pg) # Test conditions depend on what we're going to be exercising. # * Lift the threshold on whatever throttle we are *not* testing, so # that the throttle of interest is the one that will be the bottleneck # * Create either many small files (test file count throttling) or fewer # large files (test op throttling) if throttle_type == self.OPS_THROTTLE: set_throttles(files=100000000, ops=16) size_unit = 1024 * 1024 # big files, generate lots of ops file_multiplier = 100 elif throttle_type == self.FILES_THROTTLE: # The default value of file limit is pretty permissive, so to avoid # the test running too fast, create lots of files and set the limit # pretty low. set_throttles(ops=100000000, files=6) size_unit = 1024 # small, numerous files file_multiplier = 200 else: raise NotImplemented(throttle_type) # Pick up config changes self.fs.mds_fail_restart() self.fs.wait_for_daemons() create_script = dedent(""" import os mount_path = "{mount_path}" subdir = "delete_me" size_unit = {size_unit} file_multiplier = {file_multiplier} os.mkdir(os.path.join(mount_path, subdir)) for i in xrange(0, file_multiplier): for size in xrange(0, {size_range}*size_unit, size_unit): filename = "{{0}}_{{1}}.bin".format(i, size / size_unit) f = open(os.path.join(mount_path, subdir, filename), 'w') f.write(size * 'x') f.close() """.format( mount_path=self.mount_a.mountpoint, size_unit=size_unit, file_multiplier=file_multiplier, size_range=self.throttle_workload_size_range )) self.mount_a.run_python(create_script) # We will run the deletion in the background, to reduce the risk of it completing before # we have started monitoring the stray statistics. def background(): self.mount_a.run_shell(["rm", "-rf", "delete_me"]) self.fs.mds_asok(["flush", "journal"]) background_thread = gevent.spawn(background) total_inodes = file_multiplier * self.throttle_workload_size_range + 1 mds_max_purge_ops = int(self.fs.get_config("mds_max_purge_ops", 'mds')) mds_max_purge_files = int(self.fs.get_config("mds_max_purge_files", 'mds')) # During this phase we look for the concurrent ops to exceed half # the limit (a heuristic) and not exceed the limit (a correctness # condition). purge_timeout = 600 elapsed = 0 files_high_water = 0 ops_high_water = 0 while True: stats = self.fs.mds_asok(['perf', 'dump']) mdc_stats = stats['mds_cache'] pq_stats = stats['purge_queue'] if elapsed >= purge_timeout: raise RuntimeError("Timeout waiting for {0} inodes to purge, stats:{1}".format(total_inodes, mdc_stats)) num_strays = mdc_stats['num_strays'] num_strays_purging = pq_stats['pq_executing'] num_purge_ops = pq_stats['pq_executing_ops'] self.data_log.append([datetime.datetime.now(), num_strays, num_strays_purging, num_purge_ops]) files_high_water = max(files_high_water, num_strays_purging) ops_high_water = max(ops_high_water, num_purge_ops) total_strays_created = mdc_stats['strays_created'] total_strays_purged = pq_stats['pq_executed'] if total_strays_purged == total_inodes: log.info("Complete purge in {0} seconds".format(elapsed)) break elif total_strays_purged > total_inodes: raise RuntimeError("Saw more strays than expected, mdc stats: {0}".format(mdc_stats)) else: if throttle_type == self.OPS_THROTTLE: # 11 is filer_max_purge_ops plus one for the backtrace: # limit is allowed to be overshot by this much. if num_purge_ops > mds_max_purge_ops + 11: raise RuntimeError("num_purge_ops violates threshold {0}/{1}".format( num_purge_ops, mds_max_purge_ops )) elif throttle_type == self.FILES_THROTTLE: if num_strays_purging > mds_max_purge_files: raise RuntimeError("num_strays_purging violates threshold {0}/{1}".format( num_strays_purging, mds_max_purge_files )) else: raise NotImplemented(throttle_type) log.info("Waiting for purge to complete {0}/{1}, {2}/{3}".format( num_strays_purging, num_strays, total_strays_purged, total_strays_created )) time.sleep(1) elapsed += 1 background_thread.join() # Check that we got up to a respectable rate during the purge. This is totally # racy, but should be safeish unless the cluster is pathologically slow, or # insanely fast such that the deletions all pass before we have polled the # statistics. if throttle_type == self.OPS_THROTTLE: if ops_high_water < mds_max_purge_ops / 2: raise RuntimeError("Ops in flight high water is unexpectedly low ({0} / {1})".format( ops_high_water, mds_max_purge_ops )) elif throttle_type == self.FILES_THROTTLE: if files_high_water < mds_max_purge_files / 2: raise RuntimeError("Files in flight high water is unexpectedly low ({0} / {1})".format( ops_high_water, mds_max_purge_files )) # Sanity check all MDC stray stats stats = self.fs.mds_asok(['perf', 'dump']) mdc_stats = stats['mds_cache'] pq_stats = stats['purge_queue'] self.assertEqual(mdc_stats['num_strays'], 0) self.assertEqual(mdc_stats['num_strays_delayed'], 0) self.assertEqual(pq_stats['pq_executing'], 0) self.assertEqual(pq_stats['pq_executing_ops'], 0) self.assertEqual(mdc_stats['strays_created'], total_inodes) self.assertEqual(mdc_stats['strays_enqueued'], total_inodes) self.assertEqual(pq_stats['pq_executed'], total_inodes) def get_mdc_stat(self, name, mds_id=None): return self.get_stat("mds_cache", name, mds_id) def get_stat(self, subsys, name, mds_id=None): return self.fs.mds_asok(['perf', 'dump', subsys, name], mds_id=mds_id)[subsys][name] def _wait_for_counter(self, subsys, counter, expect_val, timeout=60, mds_id=None): self.wait_until_equal( lambda: self.get_stat(subsys, counter, mds_id), expect_val=expect_val, timeout=timeout, reject_fn=lambda x: x > expect_val ) def test_open_inode(self): """ That the case of a dentry unlinked while a client holds an inode open is handled correctly. The inode should be moved into a stray dentry, while the original dentry and directory should be purged. The inode's data should be purged when the client eventually closes it. """ mount_a_client_id = self.mount_a.get_global_id() # Write some bytes to a file size_mb = 8 # Hold the file open p = self.mount_a.open_background("open_file") self.mount_a.write_n_mb("open_file", size_mb) open_file_ino = self.mount_a.path_to_ino("open_file") self.assertEqual(self.get_session(mount_a_client_id)['num_caps'], 2) # Unlink the dentry self.mount_a.run_shell(["rm", "-f", "open_file"]) # Wait to see the stray count increment self.wait_until_equal( lambda: self.get_mdc_stat("num_strays"), expect_val=1, timeout=60, reject_fn=lambda x: x > 1) # See that while the stray count has incremented, none have passed # on to the purge queue self.assertEqual(self.get_mdc_stat("strays_created"), 1) self.assertEqual(self.get_mdc_stat("strays_enqueued"), 0) # See that the client still holds 2 caps self.assertEqual(self.get_session(mount_a_client_id)['num_caps'], 2) # See that the data objects remain in the data pool self.assertTrue(self.fs.data_objects_present(open_file_ino, size_mb * 1024 * 1024)) # Now close the file self.mount_a.kill_background(p) # Wait to see the client cap count decrement self.wait_until_equal( lambda: self.get_session(mount_a_client_id)['num_caps'], expect_val=1, timeout=60, reject_fn=lambda x: x > 2 or x < 1 ) # Wait to see the purge counter increment, stray count go to zero self._wait_for_counter("mds_cache", "strays_enqueued", 1) self.wait_until_equal( lambda: self.get_mdc_stat("num_strays"), expect_val=0, timeout=6, reject_fn=lambda x: x > 1 ) self._wait_for_counter("purge_queue", "pq_executed", 1) # See that the data objects no longer exist self.assertTrue(self.fs.data_objects_absent(open_file_ino, size_mb * 1024 * 1024)) self.await_data_pool_empty() def test_hardlink_reintegration(self): """ That removal of primary dentry of hardlinked inode results in reintegration of inode into the previously-remote dentry, rather than lingering as a stray indefinitely. """ # Write some bytes to file_a size_mb = 8 self.mount_a.write_n_mb("file_a", size_mb) ino = self.mount_a.path_to_ino("file_a") # Create a hardlink named file_b self.mount_a.run_shell(["ln", "file_a", "file_b"]) self.assertEqual(self.mount_a.path_to_ino("file_b"), ino) # Flush journal self.fs.mds_asok(['flush', 'journal']) # See that backtrace for the file points to the file_a path pre_unlink_bt = self.fs.read_backtrace(ino) self.assertEqual(pre_unlink_bt['ancestors'][0]['dname'], "file_a") # Unlink file_a self.mount_a.run_shell(["rm", "-f", "file_a"]) # See that a stray was created self.assertEqual(self.get_mdc_stat("num_strays"), 1) self.assertEqual(self.get_mdc_stat("strays_created"), 1) # Wait, see that data objects are still present (i.e. that the # stray did not advance to purging given time) time.sleep(30) self.assertTrue(self.fs.data_objects_present(ino, size_mb * 1024 * 1024)) self.assertEqual(self.get_mdc_stat("strays_enqueued"), 0) # See that before reintegration, the inode's backtrace points to a stray dir self.fs.mds_asok(['flush', 'journal']) self.assertTrue(self.get_backtrace_path(ino).startswith("stray")) # Do a metadata operation on the remaining link (mv is heavy handed, but # others like touch may be satisfied from caps without poking MDS) self.mount_a.run_shell(["mv", "file_b", "file_c"]) # See the reintegration counter increment # This should happen as a result of the eval_remote call on # responding to a client request. self._wait_for_counter("mds_cache", "strays_reintegrated", 1) # Flush the journal self.fs.mds_asok(['flush', 'journal']) # See that the backtrace for the file points to the remaining link's path post_reint_bt = self.fs.read_backtrace(ino) self.assertEqual(post_reint_bt['ancestors'][0]['dname'], "file_c") # See that the number of strays in existence is zero self.assertEqual(self.get_mdc_stat("num_strays"), 0) # Now really delete it self.mount_a.run_shell(["rm", "-f", "file_c"]) self._wait_for_counter("mds_cache", "strays_enqueued", 1) self._wait_for_counter("purge_queue", "pq_executed", 1) self.assert_purge_idle() self.assertTrue(self.fs.data_objects_absent(ino, size_mb * 1024 * 1024)) # We caused the inode to go stray twice self.assertEqual(self.get_mdc_stat("strays_created"), 2) # One time we reintegrated it self.assertEqual(self.get_mdc_stat("strays_reintegrated"), 1) # Then the second time we purged it self.assertEqual(self.get_mdc_stat("strays_enqueued"), 1) def test_mv_hardlink_cleanup(self): """ That when doing a rename from A to B, and B has hardlinks, then we make a stray for B which is then reintegrated into one of his hardlinks. """ # Create file_a, file_b, and a hardlink to file_b size_mb = 8 self.mount_a.write_n_mb("file_a", size_mb) file_a_ino = self.mount_a.path_to_ino("file_a") self.mount_a.write_n_mb("file_b", size_mb) file_b_ino = self.mount_a.path_to_ino("file_b") self.mount_a.run_shell(["ln", "file_b", "linkto_b"]) self.assertEqual(self.mount_a.path_to_ino("linkto_b"), file_b_ino) # mv file_a file_b self.mount_a.run_shell(["mv", "file_a", "file_b"]) self.fs.mds_asok(['flush', 'journal']) # Initially, linkto_b will still be a remote inode pointing to a newly created # stray from when file_b was unlinked due to the 'mv'. No data objects should # have been deleted, as both files still have linkage. self.assertEqual(self.get_mdc_stat("num_strays"), 1) self.assertEqual(self.get_mdc_stat("strays_created"), 1) self.assertTrue(self.get_backtrace_path(file_b_ino).startswith("stray")) self.assertTrue(self.fs.data_objects_present(file_a_ino, size_mb * 1024 * 1024)) self.assertTrue(self.fs.data_objects_present(file_b_ino, size_mb * 1024 * 1024)) # Trigger reintegration and wait for it to happen self.assertEqual(self.get_mdc_stat("strays_reintegrated"), 0) self.mount_a.run_shell(["mv", "linkto_b", "file_c"]) self._wait_for_counter("mds_cache", "strays_reintegrated", 1) self.fs.mds_asok(['flush', 'journal']) post_reint_bt = self.fs.read_backtrace(file_b_ino) self.assertEqual(post_reint_bt['ancestors'][0]['dname'], "file_c") self.assertEqual(self.get_mdc_stat("num_strays"), 0) def _setup_two_ranks(self): # Set up two MDSs self.fs.set_allow_multimds(True) self.fs.set_max_mds(2) # See that we have two active MDSs self.wait_until_equal(lambda: len(self.fs.get_active_names()), 2, 30, reject_fn=lambda v: v > 2 or v < 1) active_mds_names = self.fs.get_active_names() rank_0_id = active_mds_names[0] rank_1_id = active_mds_names[1] log.info("Ranks 0 and 1 are {0} and {1}".format( rank_0_id, rank_1_id)) # Get rid of other MDS daemons so that it's easier to know which # daemons to expect in which ranks after restarts for unneeded_mds in set(self.mds_cluster.mds_ids) - {rank_0_id, rank_1_id}: self.mds_cluster.mds_stop(unneeded_mds) self.mds_cluster.mds_fail(unneeded_mds) return rank_0_id, rank_1_id def _force_migrate(self, from_id, to_id, path, watch_ino): """ :param from_id: MDS id currently containing metadata :param to_id: MDS id to move it to :param path: Filesystem path (string) to move :param watch_ino: Inode number to look for at destination to confirm move :return: None """ result = self.fs.mds_asok(["export", "dir", path, "1"], from_id) self.assertEqual(result["return_code"], 0) # Poll the MDS cache dump to watch for the export completing migrated = False migrate_timeout = 60 migrate_elapsed = 0 while not migrated: data = self.fs.mds_asok(["dump", "cache"], to_id) for inode_data in data: if inode_data['ino'] == watch_ino: log.debug("Found ino in cache: {0}".format(json.dumps(inode_data, indent=2))) if inode_data['is_auth'] is True: migrated = True break if not migrated: if migrate_elapsed > migrate_timeout: raise RuntimeError("Migration hasn't happened after {0}s!".format(migrate_elapsed)) else: migrate_elapsed += 1 time.sleep(1) def _is_stopped(self, rank): mds_map = self.fs.get_mds_map() return rank not in [i['rank'] for i in mds_map['info'].values()] def test_purge_on_shutdown(self): """ That when an MDS rank is shut down, its purge queue is drained in the process. """ rank_0_id, rank_1_id = self._setup_two_ranks() self.set_conf("mds.{0}".format(rank_1_id), 'mds_max_purge_files', "0") self.mds_cluster.mds_fail_restart(rank_1_id) self.fs.wait_for_daemons() file_count = 5 self.mount_a.create_n_files("delete_me/file", file_count) self._force_migrate(rank_0_id, rank_1_id, "/delete_me", self.mount_a.path_to_ino("delete_me/file_0")) self.mount_a.run_shell(["rm", "-rf", Raw("delete_me/*")]) self.mount_a.umount_wait() # See all the strays go into purge queue self._wait_for_counter("mds_cache", "strays_created", file_count, mds_id=rank_1_id) self._wait_for_counter("mds_cache", "strays_enqueued", file_count, mds_id=rank_1_id) self.assertEqual(self.get_stat("mds_cache", "num_strays", mds_id=rank_1_id), 0) # See nothing get purged from the purge queue (yet) time.sleep(10) self.assertEqual(self.get_stat("purge_queue", "pq_executed", mds_id=rank_1_id), 0) # Shut down rank 1 self.fs.set_max_mds(1) self.fs.deactivate(1) # It shouldn't proceed past stopping because its still not allowed # to purge time.sleep(10) self.assertEqual(self.get_stat("purge_queue", "pq_executed", mds_id=rank_1_id), 0) self.assertFalse(self._is_stopped(1)) # Permit the daemon to start purging again self.fs.mon_manager.raw_cluster_cmd('tell', 'mds.{0}'.format(rank_1_id), 'injectargs', "--mds_max_purge_files 100") # It should now proceed through shutdown self.wait_until_true( lambda: self._is_stopped(1), timeout=60 ) # ...and in the process purge all that data self.await_data_pool_empty() def test_migration_on_shutdown(self): """ That when an MDS rank is shut down, any non-purgeable strays get migrated to another rank. """ rank_0_id, rank_1_id = self._setup_two_ranks() # Create a non-purgeable stray in a ~mds1 stray directory # by doing a hard link and deleting the original file self.mount_a.run_shell(["mkdir", "mydir"]) self.mount_a.run_shell(["touch", "mydir/original"]) self.mount_a.run_shell(["ln", "mydir/original", "mydir/linkto"]) self._force_migrate(rank_0_id, rank_1_id, "/mydir", self.mount_a.path_to_ino("mydir/original")) self.mount_a.run_shell(["rm", "-f", "mydir/original"]) self.mount_a.umount_wait() self._wait_for_counter("mds_cache", "strays_created", 1, mds_id=rank_1_id) # Shut down rank 1 self.fs.mon_manager.raw_cluster_cmd_result('mds', 'set', "max_mds", "1") self.fs.mon_manager.raw_cluster_cmd_result('mds', 'deactivate', "1") # Wait til we get to a single active MDS mdsmap state self.wait_until_true(lambda: self._is_stopped(1), timeout=120) # See that the stray counter on rank 0 has incremented self.assertEqual(self.get_mdc_stat("strays_created", rank_0_id), 1) def assert_backtrace(self, ino, expected_path): """ Assert that the backtrace in the data pool for an inode matches an expected /foo/bar path. """ expected_elements = expected_path.strip("/").split("/") bt = self.fs.read_backtrace(ino) actual_elements = list(reversed([dn['dname'] for dn in bt['ancestors']])) self.assertListEqual(expected_elements, actual_elements) def get_backtrace_path(self, ino): bt = self.fs.read_backtrace(ino) elements = reversed([dn['dname'] for dn in bt['ancestors']]) return "/".join(elements) def assert_purge_idle(self): """ Assert that the MDS perf counters indicate no strays exist and no ongoing purge activity. Sanity check for when PurgeQueue should be idle. """ mdc_stats = self.fs.mds_asok(['perf', 'dump', "mds_cache"])['mds_cache'] pq_stats = self.fs.mds_asok(['perf', 'dump', "purge_queue"])['purge_queue'] self.assertEqual(mdc_stats["num_strays"], 0) self.assertEqual(mdc_stats["num_strays_delayed"], 0) self.assertEqual(pq_stats["pq_executing"], 0) self.assertEqual(pq_stats["pq_executing_ops"], 0) def test_mv_cleanup(self): """ That when doing a rename from A to B, and B has no hardlinks, then we make a stray for B and purge him. """ # Create file_a and file_b, write some to both size_mb = 8 self.mount_a.write_n_mb("file_a", size_mb) file_a_ino = self.mount_a.path_to_ino("file_a") self.mount_a.write_n_mb("file_b", size_mb) file_b_ino = self.mount_a.path_to_ino("file_b") self.fs.mds_asok(['flush', 'journal']) self.assert_backtrace(file_a_ino, "file_a") self.assert_backtrace(file_b_ino, "file_b") # mv file_a file_b self.mount_a.run_shell(['mv', 'file_a', 'file_b']) # See that stray counter increments self.assertEqual(self.get_mdc_stat("strays_created"), 1) # Wait for purge counter to increment self._wait_for_counter("mds_cache", "strays_enqueued", 1) self._wait_for_counter("purge_queue", "pq_executed", 1) self.assert_purge_idle() # file_b should have been purged self.assertTrue(self.fs.data_objects_absent(file_b_ino, size_mb * 1024 * 1024)) # Backtrace should have updated from file_a to file_b self.fs.mds_asok(['flush', 'journal']) self.assert_backtrace(file_a_ino, "file_b") # file_a's data should still exist self.assertTrue(self.fs.data_objects_present(file_a_ino, size_mb * 1024 * 1024)) def _pool_df(self, pool_name): """ Return a dict like { "kb_used": 0, "bytes_used": 0, "max_avail": 19630292406, "objects": 0 } :param pool_name: Which pool (must exist) """ out = self.fs.mon_manager.raw_cluster_cmd("df", "--format=json-pretty") for p in json.loads(out)['pools']: if p['name'] == pool_name: return p['stats'] raise RuntimeError("Pool '{0}' not found".format(pool_name)) def await_data_pool_empty(self): self.wait_until_true( lambda: self._pool_df( self.fs.get_data_pool_name() )['objects'] == 0, timeout=60) def test_snapshot_remove(self): """ That removal of a snapshot that references a now-unlinked file results in purging on the stray for the file. """ # Enable snapshots self.fs.mon_manager.raw_cluster_cmd("mds", "set", "allow_new_snaps", "true", "--yes-i-really-mean-it") # Create a dir with a file in it size_mb = 8 self.mount_a.run_shell(["mkdir", "snapdir"]) self.mount_a.run_shell(["mkdir", "snapdir/subdir"]) self.mount_a.write_test_pattern("snapdir/subdir/file_a", size_mb * 1024 * 1024) file_a_ino = self.mount_a.path_to_ino("snapdir/subdir/file_a") # Snapshot the dir self.mount_a.run_shell(["mkdir", "snapdir/.snap/snap1"]) # Cause the head revision to deviate from the snapshot self.mount_a.write_n_mb("snapdir/subdir/file_a", size_mb) # Flush the journal so that backtraces, dirfrag objects will actually be written self.fs.mds_asok(["flush", "journal"]) # Unlink the file self.mount_a.run_shell(["rm", "-f", "snapdir/subdir/file_a"]) self.mount_a.run_shell(["rmdir", "snapdir/subdir"]) # Unmount the client because when I come back to check the data is still # in the file I don't want to just see what's in the page cache. self.mount_a.umount_wait() self.assertEqual(self.get_mdc_stat("strays_created"), 2) # FIXME: at this stage we see a purge and the stray count drops to # zero, but there's actually still a stray, so at the very # least the StrayManager stats code is slightly off self.mount_a.mount() # See that the data from the snapshotted revision of the file is still present # and correct self.mount_a.validate_test_pattern("snapdir/.snap/snap1/subdir/file_a", size_mb * 1024 * 1024) # Remove the snapshot self.mount_a.run_shell(["rmdir", "snapdir/.snap/snap1"]) self.mount_a.umount_wait() # Purging file_a doesn't happen until after we've flushed the journal, because # it is referenced by the snapshotted subdir, and the snapshot isn't really # gone until the journal references to it are gone self.fs.mds_asok(["flush", "journal"]) # See that a purge happens now self._wait_for_counter("mds_cache", "strays_enqueued", 2) self._wait_for_counter("purge_queue", "pq_executed", 2) self.assertTrue(self.fs.data_objects_absent(file_a_ino, size_mb * 1024 * 1024)) self.await_data_pool_empty() def test_fancy_layout(self): """ purge stray file with fancy layout """ file_name = "fancy_layout_file" self.mount_a.run_shell(["touch", file_name]) file_layout = "stripe_unit=1048576 stripe_count=4 object_size=8388608" self.mount_a.setfattr(file_name, "ceph.file.layout", file_layout) # 35MB requires 7 objects size_mb = 35 self.mount_a.write_n_mb(file_name, size_mb) self.mount_a.run_shell(["rm", "-f", file_name]) self.fs.mds_asok(["flush", "journal"]) # can't use self.fs.data_objects_absent here, it does not support fancy layout self.await_data_pool_empty() def test_dirfrag_limit(self): """ That the directory fragment size cannot exceed mds_bal_fragment_size_max (using a limit of 50 in all configurations). That fragmentation (forced) will allow more entries to be created. That unlinking fails when the stray directory fragment becomes too large and that unlinking may continue once those strays are purged. """ self.fs.set_allow_dirfrags(True) LOW_LIMIT = 50 for mds in self.fs.get_daemon_names(): self.fs.mds_asok(["config", "set", "mds_bal_fragment_size_max", str(LOW_LIMIT)], mds) try: 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=LOW_LIMIT+1 ))) except CommandFailedError: pass # ENOSPAC else: raise RuntimeError("fragment size exceeded") # Now test that we can go beyond the limit if we fragment the directory self.mount_a.run_python(dedent(""" import os path = os.path.join("{path}", "subdir2") os.mkdir(path) for n in range(0, {file_count}): open(os.path.join(path, "%s" % n), 'w').write("%s" % n) dfd = os.open(path, os.O_DIRECTORY) os.fsync(dfd) """.format( path=self.mount_a.mountpoint, file_count=LOW_LIMIT ))) # Ensure that subdir2 is fragmented mds_id = self.fs.get_active_names()[0] self.fs.mds_asok(["dirfrag", "split", "/subdir2", "0/0", "1"], mds_id) # remount+flush (release client caps) self.mount_a.umount_wait() self.fs.mds_asok(["flush", "journal"], mds_id) self.mount_a.mount() self.mount_a.wait_until_mounted() # Create 50% more files than the current fragment limit self.mount_a.run_python(dedent(""" import os path = os.path.join("{path}", "subdir2") for n in range({file_count}, ({file_count}*3)//2): open(os.path.join(path, "%s" % n), 'w').write("%s" % n) """.format( path=self.mount_a.mountpoint, file_count=LOW_LIMIT ))) # Now test the stray directory size is limited and recovers strays_before = self.get_mdc_stat("strays_created") try: self.mount_a.run_python(dedent(""" import os path = os.path.join("{path}", "subdir3") os.mkdir(path) for n in range({file_count}): fpath = os.path.join(path, "%s" % n) f = open(fpath, 'w') f.write("%s" % n) f.close() os.unlink(fpath) """.format( path=self.mount_a.mountpoint, file_count=LOW_LIMIT*10 # 10 stray directories, should collide before this count ))) except CommandFailedError: pass # ENOSPAC else: raise RuntimeError("fragment size exceeded") strays_after = self.get_mdc_stat("strays_created") self.assertGreaterEqual(strays_after-strays_before, LOW_LIMIT) self._wait_for_counter("mds_cache", "strays_enqueued", strays_after) self._wait_for_counter("purge_queue", "pq_executed", strays_after) self.mount_a.run_python(dedent(""" import os path = os.path.join("{path}", "subdir4") os.mkdir(path) for n in range({file_count}): fpath = os.path.join(path, "%s" % n) f = open(fpath, 'w') f.write("%s" % n) f.close() os.unlink(fpath) """.format( path=self.mount_a.mountpoint, file_count=LOW_LIMIT ))) def test_purge_queue_upgrade(self): """ That when starting on a system with no purge queue in the metadata pool, we silently create one. :return: """ self.mds_cluster.mds_stop() self.mds_cluster.mds_fail() self.fs.rados(["rm", "500.00000000"]) self.mds_cluster.mds_restart() self.fs.wait_for_daemons() def test_purge_queue_op_rate(self): """ A busy purge queue is meant to aggregate operations sufficiently that our RADOS ops to the metadata pool are not O(files). Check that that is so. :return: """ # For low rates of deletion, the rate of metadata ops actually # will be o(files), so to see the desired behaviour we have to give # the system a significant quantity, i.e. an order of magnitude # more than the number of files it will purge at one time. max_purge_files = 2 self.set_conf('mds', 'mds_bal_frag', 'false') self.set_conf('mds', 'mds_max_purge_files', "%d" % max_purge_files) self.fs.mds_fail_restart() self.fs.wait_for_daemons() phase_1_files = 256 phase_2_files = 512 self.mount_a.run_shell(["mkdir", "phase1"]) self.mount_a.create_n_files("phase1/file", phase_1_files) self.mount_a.run_shell(["mkdir", "phase2"]) self.mount_a.create_n_files("phase2/file", phase_2_files) def unlink_and_count_ops(path, expected_deletions): initial_ops = self.get_stat("objecter", "op") initial_pq_executed = self.get_stat("purge_queue", "pq_executed") self.mount_a.run_shell(["rm", "-rf", path]) self._wait_for_counter( "purge_queue", "pq_executed", initial_pq_executed + expected_deletions ) final_ops = self.get_stat("objecter", "op") # Calculation of the *overhead* operations, i.e. do not include # the operations where we actually delete files. return final_ops - initial_ops - expected_deletions self.fs.mds_asok(['flush', 'journal']) phase1_ops = unlink_and_count_ops("phase1/", phase_1_files + 1) self.fs.mds_asok(['flush', 'journal']) phase2_ops = unlink_and_count_ops("phase2/", phase_2_files + 1) log.info("Phase 1: {0}".format(phase1_ops)) log.info("Phase 2: {0}".format(phase2_ops)) # The success criterion is that deleting double the number # of files doesn't generate double the number of overhead ops # -- this comparison is a rough approximation of that rule. self.assertTrue(phase2_ops < phase1_ops * 1.25) # Finally, check that our activity did include properly quiescing # the queue (i.e. call to Journaler::write_head in the right place), # by restarting the MDS and checking that it doesn't try re-executing # any of the work we did. self.fs.mds_asok(['flush', 'journal']) # flush to ensure no strays # hanging around self.fs.mds_fail_restart() self.fs.wait_for_daemons() time.sleep(10) self.assertEqual(self.get_stat("purge_queue", "pq_executed"), 0)