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.
8 from textwrap
import dedent
9 from tasks
.ceph_test_case
import TestTimeoutError
10 from tasks
.cephfs
.cephfs_test_case
import CephFSTestCase
, needs_trimming
11 from tasks
.cephfs
.fuse_mount
import FuseMount
15 log
= logging
.getLogger(__name__
)
18 # Arbitrary timeouts for operations involving restarting
19 # an MDS or waiting for it to come up
20 MDS_RESTART_GRACE
= 60
22 # Hardcoded values from Server::recall_client_state
23 CAP_RECALL_RATIO
= 0.8
27 class TestClientLimits(CephFSTestCase
):
30 def _test_client_pin(self
, use_subdir
, open_files
):
32 When a client pins an inode in its cache, for example because the file is held open,
33 it should reject requests from the MDS to trim these caps. The MDS should complain
34 to the user that it is unable to enforce its cache size limits because of this
37 :param use_subdir: whether to put test files in a subdir or use root
40 # Set MDS cache memory limit to a low value that will make the MDS to
41 # ask the client to trim the caps.
42 cache_memory_limit
= "1K"
44 self
.config_set('mds', 'mds_cache_memory_limit', cache_memory_limit
)
45 self
.config_set('mds', 'mds_recall_max_caps', int(open_files
/2))
46 self
.config_set('mds', 'mds_recall_warning_threshold', open_files
)
48 mds_min_caps_per_client
= int(self
.config_get('mds', "mds_min_caps_per_client"))
49 self
.config_set('mds', 'mds_min_caps_working_set', mds_min_caps_per_client
)
50 mds_max_caps_per_client
= int(self
.config_get('mds', "mds_max_caps_per_client"))
51 mds_recall_warning_decay_rate
= float(self
.config_get('mds', "mds_recall_warning_decay_rate"))
52 self
.assertGreaterEqual(open_files
, mds_min_caps_per_client
)
54 mount_a_client_id
= self
.mount_a
.get_global_id()
55 path
= "subdir" if use_subdir
else "."
56 open_proc
= self
.mount_a
.open_n_background(path
, open_files
)
58 # Client should now hold:
59 # `open_files` caps for the open files
62 self
.wait_until_equal(lambda: self
.get_session(mount_a_client_id
)['num_caps'],
63 open_files
+ (2 if use_subdir
else 1),
65 reject_fn
=lambda x
: x
> open_files
+ 2)
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)
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)
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 self
.mount_a
._kill
_background
(open_proc
)
80 # The remaining caps should comply with the numbers sent from MDS in SESSION_RECALL message,
81 # which depend on the caps outstanding, cache size and overall ratio
83 num_caps
= self
.get_session(mount_a_client_id
)['num_caps']
84 if num_caps
<= mds_min_caps_per_client
:
86 elif num_caps
<= mds_max_caps_per_client
:
91 self
.wait_until_true(expected_caps
, timeout
=60)
94 def test_client_pin_root(self
):
95 self
._test
_client
_pin
(False, 400)
98 def test_client_pin(self
):
99 self
._test
_client
_pin
(True, 800)
102 def test_client_pin_mincaps(self
):
103 self
._test
_client
_pin
(True, 200)
105 def test_client_min_caps_working_set(self
):
107 When a client has inodes pinned in its cache (open files), that the MDS
108 will not warn about the client not responding to cache pressure when
109 the number of caps is below mds_min_caps_working_set.
112 # Set MDS cache memory limit to a low value that will make the MDS to
113 # ask the client to trim the caps.
114 cache_memory_limit
= "1K"
117 self
.config_set('mds', 'mds_cache_memory_limit', cache_memory_limit
)
118 self
.config_set('mds', 'mds_recall_max_caps', int(open_files
/2))
119 self
.config_set('mds', 'mds_recall_warning_threshold', open_files
)
120 self
.config_set('mds', 'mds_min_caps_working_set', open_files
*2)
122 mds_min_caps_per_client
= int(self
.config_get('mds', "mds_min_caps_per_client"))
123 mds_recall_warning_decay_rate
= float(self
.config_get('mds', "mds_recall_warning_decay_rate"))
124 self
.assertGreaterEqual(open_files
, mds_min_caps_per_client
)
126 mount_a_client_id
= self
.mount_a
.get_global_id()
127 self
.mount_a
.open_n_background("subdir", open_files
)
129 # Client should now hold:
130 # `open_files` caps for the open files
133 self
.wait_until_equal(lambda: self
.get_session(mount_a_client_id
)['num_caps'],
136 reject_fn
=lambda x
: x
> open_files
+ 2)
138 # We can also test that the MDS health warning for oversized
139 # cache is functioning as intended.
140 self
.wait_for_health("MDS_CACHE_OVERSIZED", mds_recall_warning_decay_rate
*2)
143 # MDS should not be happy about that but it's not sending
144 # MDS_CLIENT_RECALL warnings because the client's caps are below
145 # mds_min_caps_working_set.
146 self
.wait_for_health("MDS_CLIENT_RECALL", mds_recall_warning_decay_rate
*2)
147 except TestTimeoutError
:
150 raise RuntimeError("expected no client recall warning")
152 def test_cap_acquisition_throttle_readdir(self
):
154 Mostly readdir acquires caps faster than the mds recalls, so the cap
155 acquisition via readdir is throttled by retrying the readdir after
156 a fraction of second (0.5) by default when throttling condition is met.
159 max_caps_per_client
= 500
160 cap_acquisition_throttle
= 250
162 self
.config_set('mds', 'mds_max_caps_per_client', max_caps_per_client
)
163 self
.config_set('mds', 'mds_session_cap_acquisition_throttle', cap_acquisition_throttle
)
165 # Create 1500 files split across 6 directories, 250 each.
166 for i
in range(1, 7):
167 self
.mount_a
.create_n_files("dir{0}/file".format(i
), cap_acquisition_throttle
, sync
=True)
169 mount_a_client_id
= self
.mount_a
.get_global_id()
172 self
.mount_a
.run_shell_payload("find | wc")
174 # validate cap_acquisition decay counter after readdir to exceed throttle count i.e 250
175 cap_acquisition_value
= self
.get_session(mount_a_client_id
)['cap_acquisition']['value']
176 self
.assertGreaterEqual(cap_acquisition_value
, cap_acquisition_throttle
)
178 # validate the throttle condition to be hit atleast once
179 cap_acquisition_throttle_hit_count
= self
.perf_dump()['mds_server']['cap_acquisition_throttle']
180 self
.assertGreaterEqual(cap_acquisition_throttle_hit_count
, 1)
182 def test_client_release_bug(self
):
184 When a client has a bug (which we will simulate) preventing it from releasing caps,
185 the MDS should notice that releases are not being sent promptly, and generate a health
186 metric to that effect.
189 # The debug hook to inject the failure only exists in the fuse client
190 if not isinstance(self
.mount_a
, FuseMount
):
191 self
.skipTest("Require FUSE client to inject client release failure")
193 self
.set_conf('client.{0}'.format(self
.mount_a
.client_id
), 'client inject release failure', 'true')
194 self
.mount_a
.teardown()
195 self
.mount_a
.mount_wait()
196 mount_a_client_id
= self
.mount_a
.get_global_id()
198 # Client A creates a file. He will hold the write caps on the file, and later (simulated bug) fail
199 # to comply with the MDSs request to release that cap
200 self
.mount_a
.run_shell(["touch", "file1"])
202 # Client B tries to stat the file that client A created
203 rproc
= self
.mount_b
.write_background("file1")
205 # After session_timeout, we should see a health warning (extra lag from
207 session_timeout
= self
.fs
.get_var("session_timeout")
208 self
.wait_for_health("MDS_CLIENT_LATE_RELEASE", session_timeout
+ 10)
210 # Client B should still be stuck
211 self
.assertFalse(rproc
.finished
)
215 self
.mount_a
.kill_cleanup()
217 # Client B should complete
218 self
.fs
.mds_asok(['session', 'evict', "%s" % mount_a_client_id
])
221 def test_client_oldest_tid(self
):
223 When a client does not advance its oldest tid, the MDS should notice that
224 and generate health warnings.
227 # num of requests client issues
230 # The debug hook to inject the failure only exists in the fuse client
231 if not isinstance(self
.mount_a
, FuseMount
):
232 self
.skipTest("Require FUSE client to inject client release failure")
234 self
.set_conf('client', 'client inject fixed oldest tid', 'true')
235 self
.mount_a
.teardown()
236 self
.mount_a
.mount_wait()
238 self
.fs
.mds_asok(['config', 'set', 'mds_max_completed_requests', '{0}'.format(max_requests
)])
240 # Create lots of files
241 self
.mount_a
.create_n_files("testdir/file1", max_requests
+ 100)
243 # Create a few files synchronously. This makes sure previous requests are completed
244 self
.mount_a
.create_n_files("testdir/file2", 5, True)
246 # Wait for the health warnings. Assume mds can handle 10 request per second at least
247 self
.wait_for_health("MDS_CLIENT_OLDEST_TID", max_requests
// 10)
249 def _test_client_cache_size(self
, mount_subdir
):
251 check if client invalidate kernel dcache according to its cache size config
254 # The debug hook to inject the failure only exists in the fuse client
255 if not isinstance(self
.mount_a
, FuseMount
):
256 self
.skipTest("Require FUSE client to inject client release failure")
259 # fuse assigns a fix inode number (1) to root inode. But in mounting into
260 # subdir case, the actual inode number of root is not 1. This mismatch
261 # confuses fuse_lowlevel_notify_inval_entry() when invalidating dentries
263 self
.mount_a
.run_shell(["mkdir", "subdir"])
264 self
.mount_a
.umount_wait()
265 self
.set_conf('client', 'client mountpoint', '/subdir')
266 self
.mount_a
.mount_wait()
267 root_ino
= self
.mount_a
.path_to_ino(".")
268 self
.assertEqual(root_ino
, 1);
270 dir_path
= os
.path
.join(self
.mount_a
.mountpoint
, "testdir")
272 mkdir_script
= dedent("""
275 for n in range(0, {num_dirs}):
276 os.mkdir("{path}/dir{{0}}".format(n))
280 self
.mount_a
.run_python(mkdir_script
.format(path
=dir_path
, num_dirs
=num_dirs
))
281 self
.mount_a
.run_shell(["sync"])
283 dentry_count
, dentry_pinned_count
= self
.mount_a
.get_dentry_count()
284 self
.assertGreaterEqual(dentry_count
, num_dirs
)
285 self
.assertGreaterEqual(dentry_pinned_count
, num_dirs
)
287 cache_size
= num_dirs
// 10
288 self
.mount_a
.set_cache_size(cache_size
)
291 dentry_count
, dentry_pinned_count
= self
.mount_a
.get_dentry_count()
292 log
.info("waiting, dentry_count, dentry_pinned_count: {0}, {1}".format(
293 dentry_count
, dentry_pinned_count
295 if dentry_count
> cache_size
or dentry_pinned_count
> cache_size
:
300 self
.wait_until_true(trimmed
, 30)
303 def test_client_cache_size(self
):
304 self
._test
_client
_cache
_size
(False)
305 self
._test
_client
_cache
_size
(True)
307 def test_client_max_caps(self
):
309 That the MDS will not let a client sit above mds_max_caps_per_client caps.
312 mds_min_caps_per_client
= int(self
.config_get('mds', "mds_min_caps_per_client"))
313 mds_max_caps_per_client
= 2*mds_min_caps_per_client
314 self
.config_set('mds', 'mds_max_caps_per_client', mds_max_caps_per_client
)
316 self
.mount_a
.create_n_files("foo/", 3*mds_max_caps_per_client
, sync
=True)
318 mount_a_client_id
= self
.mount_a
.get_global_id()
320 num_caps
= self
.get_session(mount_a_client_id
)['num_caps']
321 if num_caps
<= mds_max_caps_per_client
:
326 self
.wait_until_true(expected_caps
, timeout
=60)