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