ceph/qa/tasks/cephfs/test_client_limits.py

   1
   2 """
   3 Exercise the MDS's behaviour when clients and the MDCache reach or
   4 exceed the limits of how many caps/inodes they should hold.
   5 """
   6
   7 import logging
   8 from textwrap import dedent
   9 from unittest import SkipTest
  10 from teuthology.orchestra.run import CommandFailedError
  11 from tasks.cephfs.cephfs_test_case import CephFSTestCase, needs_trimming
  12 from tasks.cephfs.fuse_mount import FuseMount
  13 import os
  14
  15
  16 log = logging.getLogger(__name__)
  17
  18
  19 # Arbitrary timeouts for operations involving restarting
  20 # an MDS or waiting for it to come up
  21 MDS_RESTART_GRACE = 60
  22
  23 # Hardcoded values from Server::recall_client_state
  24 CAP_RECALL_RATIO = 0.8
  25 CAP_RECALL_MIN = 100
  26
  27
  28 class TestClientLimits(CephFSTestCase):
  29     REQUIRE_KCLIENT_REMOTE = True
  30     CLIENTS_REQUIRED = 2
  31
  32     def _test_client_pin(self, use_subdir, open_files):
  33         """
  34         When a client pins an inode in its cache, for example because the file is held open,
  35         it should reject requests from the MDS to trim these caps.  The MDS should complain
  36         to the user that it is unable to enforce its cache size limits because of this
  37         objectionable client.
  38
  39         :param use_subdir: whether to put test files in a subdir or use root
  40         """
  41
  42         cache_size = open_files/2
  43
  44         self.set_conf('mds', 'mds cache size', cache_size)
  45         self.set_conf('mds', 'mds_recall_max_caps', open_files/2)
  46         self.set_conf('mds', 'mds_recall_warning_threshold', open_files)
  47         self.fs.mds_fail_restart()
  48         self.fs.wait_for_daemons()
  49
  50         mds_min_caps_per_client = int(self.fs.get_config("mds_min_caps_per_client"))
  51         mds_recall_warning_decay_rate = self.fs.get_config("mds_recall_warning_decay_rate")
  52         self.assertTrue(open_files >= mds_min_caps_per_client)
  53
  54         mount_a_client_id = self.mount_a.get_global_id()
  55         path = "subdir/mount_a" if use_subdir else "mount_a"
  56         open_proc = self.mount_a.open_n_background(path, open_files)
  57
  58         # Client should now hold:
  59         # `open_files` caps for the open files
  60         # 1 cap for root
  61         # 1 cap for subdir
  62         self.wait_until_equal(lambda: self.get_session(mount_a_client_id)['num_caps'],
  63                               open_files + (2 if use_subdir else 1),
  64                               timeout=600,
  65                               reject_fn=lambda x: x > open_files + 2)
  66
  67         # MDS should not be happy about that, as the client is failing to comply
  68         # with the SESSION_RECALL messages it is being sent
  69         self.wait_for_health("MDS_CLIENT_RECALL", mds_recall_warning_decay_rate*2)
  70
  71         # We can also test that the MDS health warning for oversized
  72         # cache is functioning as intended.
  73         self.wait_for_health("MDS_CACHE_OVERSIZED", mds_recall_warning_decay_rate*2)
  74
  75         # When the client closes the files, it should retain only as many caps as allowed
  76         # under the SESSION_RECALL policy
  77         log.info("Terminating process holding files open")
  78         open_proc.stdin.close()
  79         try:
  80             open_proc.wait()
  81         except CommandFailedError:
  82             # We killed it, so it raises an error
  83             pass
  84
  85         # The remaining caps should comply with the numbers sent from MDS in SESSION_RECALL message,
  86         # which depend on the caps outstanding, cache size and overall ratio
  87         def expected_caps():
  88             num_caps = self.get_session(mount_a_client_id)['num_caps']
  89             if num_caps <= mds_min_caps_per_client:
  90                 return True
  91             elif num_caps < cache_size:
  92                 return True
  93             else:
  94                 return False
  95
  96         self.wait_until_true(expected_caps, timeout=60)
  97
  98     @needs_trimming
  99     def test_client_pin_root(self):
 100         self._test_client_pin(False, 400)
 101
 102     @needs_trimming
 103     def test_client_pin(self):
 104         self._test_client_pin(True, 800)
 105
 106     @needs_trimming
 107     def test_client_pin_mincaps(self):
 108         self._test_client_pin(True, 200)
 109
 110     def test_client_release_bug(self):
 111         """
 112         When a client has a bug (which we will simulate) preventing it from releasing caps,
 113         the MDS should notice that releases are not being sent promptly, and generate a health
 114         metric to that effect.
 115         """
 116
 117         # The debug hook to inject the failure only exists in the fuse client
 118         if not isinstance(self.mount_a, FuseMount):
 119             raise SkipTest("Require FUSE client to inject client release failure")
 120
 121         self.set_conf('client.{0}'.format(self.mount_a.client_id), 'client inject release failure', 'true')
 122         self.mount_a.teardown()
 123         self.mount_a.mount()
 124         self.mount_a.wait_until_mounted()
 125         mount_a_client_id = self.mount_a.get_global_id()
 126
 127         # Client A creates a file.  He will hold the write caps on the file, and later (simulated bug) fail
 128         # to comply with the MDSs request to release that cap
 129         self.mount_a.run_shell(["touch", "file1"])
 130
 131         # Client B tries to stat the file that client A created
 132         rproc = self.mount_b.write_background("file1")
 133
 134         # After session_timeout, we should see a health warning (extra lag from
 135         # MDS beacon period)
 136         session_timeout = self.fs.get_var("session_timeout")
 137         self.wait_for_health("MDS_CLIENT_LATE_RELEASE", session_timeout + 10)
 138
 139         # Client B should still be stuck
 140         self.assertFalse(rproc.finished)
 141
 142         # Kill client A
 143         self.mount_a.kill()
 144         self.mount_a.kill_cleanup()
 145
 146         # Client B should complete
 147         self.fs.mds_asok(['session', 'evict', "%s" % mount_a_client_id])
 148         rproc.wait()
 149
 150     def test_client_oldest_tid(self):
 151         """
 152         When a client does not advance its oldest tid, the MDS should notice that
 153         and generate health warnings.
 154         """
 155
 156         # num of requests client issues
 157         max_requests = 1000
 158
 159         # The debug hook to inject the failure only exists in the fuse client
 160         if not isinstance(self.mount_a, FuseMount):
 161             raise SkipTest("Require FUSE client to inject client release failure")
 162
 163         self.set_conf('client', 'client inject fixed oldest tid', 'true')
 164         self.mount_a.teardown()
 165         self.mount_a.mount()
 166         self.mount_a.wait_until_mounted()
 167
 168         self.fs.mds_asok(['config', 'set', 'mds_max_completed_requests', '{0}'.format(max_requests)])
 169
 170         # Create lots of files
 171         self.mount_a.create_n_files("testdir/file1", max_requests + 100)
 172
 173         # Create a few files synchronously. This makes sure previous requests are completed
 174         self.mount_a.create_n_files("testdir/file2", 5, True)
 175
 176         # Wait for the health warnings. Assume mds can handle 10 request per second at least
 177         self.wait_for_health("MDS_CLIENT_OLDEST_TID", max_requests / 10)
 178
 179     def _test_client_cache_size(self, mount_subdir):
 180         """
 181         check if client invalidate kernel dcache according to its cache size config
 182         """
 183
 184         # The debug hook to inject the failure only exists in the fuse client
 185         if not isinstance(self.mount_a, FuseMount):
 186             raise SkipTest("Require FUSE client to inject client release failure")
 187
 188         if mount_subdir:
 189             # fuse assigns a fix inode number (1) to root inode. But in mounting into
 190             # subdir case, the actual inode number of root is not 1. This mismatch
 191             # confuses fuse_lowlevel_notify_inval_entry() when invalidating dentries
 192             # in root directory.
 193             self.mount_a.run_shell(["mkdir", "subdir"])
 194             self.mount_a.umount_wait()
 195             self.set_conf('client', 'client mountpoint', '/subdir')
 196             self.mount_a.mount()
 197             self.mount_a.wait_until_mounted()
 198             root_ino = self.mount_a.path_to_ino(".")
 199             self.assertEqual(root_ino, 1);
 200
 201         dir_path = os.path.join(self.mount_a.mountpoint, "testdir")
 202
 203         mkdir_script = dedent("""
 204             import os
 205             os.mkdir("{path}")
 206             for n in range(0, {num_dirs}):
 207                 os.mkdir("{path}/dir{{0}}".format(n))
 208             """)
 209
 210         num_dirs = 1000
 211         self.mount_a.run_python(mkdir_script.format(path=dir_path, num_dirs=num_dirs))
 212         self.mount_a.run_shell(["sync"])
 213
 214         dentry_count, dentry_pinned_count = self.mount_a.get_dentry_count()
 215         self.assertGreaterEqual(dentry_count, num_dirs)
 216         self.assertGreaterEqual(dentry_pinned_count, num_dirs)
 217
 218         cache_size = num_dirs / 10
 219         self.mount_a.set_cache_size(cache_size)
 220
 221         def trimmed():
 222             dentry_count, dentry_pinned_count = self.mount_a.get_dentry_count()
 223             log.info("waiting, dentry_count, dentry_pinned_count: {0}, {1}".format(
 224                 dentry_count, dentry_pinned_count
 225             ))
 226             if dentry_count > cache_size or dentry_pinned_count > cache_size:
 227                 return False
 228
 229             return True
 230
 231         self.wait_until_true(trimmed, 30)
 232
 233     @needs_trimming
 234     def test_client_cache_size(self):
 235         self._test_client_cache_size(False)
 236         self._test_client_cache_size(True)
 237
 238     def test_client_max_caps(self):
 239         """
 240         That the MDS will not let a client sit above mds_max_caps_per_client caps.
 241         """
 242
 243         mds_min_caps_per_client = int(self.fs.get_config("mds_min_caps_per_client"))
 244         mds_max_caps_per_client = 2*mds_min_caps_per_client
 245         self.set_conf('mds', 'mds_max_caps_per_client', mds_max_caps_per_client)
 246         self.fs.mds_fail_restart()
 247         self.fs.wait_for_daemons()
 248
 249         self.mount_a.create_n_files("foo/", 3*mds_max_caps_per_client, sync=True)
 250
 251         mount_a_client_id = self.mount_a.get_global_id()
 252         def expected_caps():
 253             num_caps = self.get_session(mount_a_client_id)['num_caps']
 254             if num_caps <= mds_max_caps_per_client:
 255                 return True
 256             else:
 257                 return False
 258
 259         self.wait_until_true(expected_caps, timeout=60)