ceph/qa/tasks/cephfs/test_sessionmap.py

   1 import time
   2 import json
   3 import logging
   4
   5 from tasks.cephfs.fuse_mount import FuseMount
   6 from teuthology.exceptions import CommandFailedError
   7 from tasks.cephfs.cephfs_test_case import CephFSTestCase
   8 from teuthology.misc import sudo_write_file
   9
  10 log = logging.getLogger(__name__)
  11
  12
  13 class TestSessionMap(CephFSTestCase):
  14     CLIENTS_REQUIRED = 2
  15     MDSS_REQUIRED = 2
  16
  17     def test_tell_session_drop(self):
  18         """
  19         That when a `tell` command is sent using the python CLI,
  20         its MDS session is gone after it terminates
  21         """
  22         self.mount_a.umount_wait()
  23         self.mount_b.umount_wait()
  24
  25         status = self.fs.status()
  26         self.fs.rank_tell(["session", "ls"], status=status)
  27
  28         ls_data = self.fs.rank_asok(['session', 'ls'], status=status)
  29         self.assertEqual(len(ls_data), 0)
  30
  31     def _get_connection_count(self, status=None):
  32         perf = self.fs.rank_asok(["perf", "dump"], status=status)
  33         conn = 0
  34         for module, dump in perf.items():
  35             if "AsyncMessenger::Worker" in module:
  36                 conn += dump['msgr_active_connections']
  37         return conn
  38
  39     def test_tell_conn_close(self):
  40         """
  41         That when a `tell` command is sent using the python CLI,
  42         the conn count goes back to where it started (i.e. we aren't
  43         leaving connections open)
  44         """
  45         self.mount_a.umount_wait()
  46         self.mount_b.umount_wait()
  47
  48         status = self.fs.status()
  49         s = self._get_connection_count(status=status)
  50         self.fs.rank_tell(["session", "ls"], status=status)
  51         e = self._get_connection_count(status=status)
  52
  53         self.assertEqual(s, e)
  54
  55     def test_mount_conn_close(self):
  56         """
  57         That when a client unmounts, the thread count on the MDS goes back
  58         to what it was before the client mounted
  59         """
  60         self.mount_a.umount_wait()
  61         self.mount_b.umount_wait()
  62
  63         status = self.fs.status()
  64         s = self._get_connection_count(status=status)
  65         self.mount_a.mount_wait()
  66         self.assertGreater(self._get_connection_count(status=status), s)
  67         self.mount_a.umount_wait()
  68         e = self._get_connection_count(status=status)
  69
  70         self.assertEqual(s, e)
  71
  72     def test_version_splitting(self):
  73         """
  74         That when many sessions are updated, they are correctly
  75         split into multiple versions to obey mds_sessionmap_keys_per_op
  76         """
  77
  78         self.mount_a.umount_wait()
  79         self.mount_b.umount_wait()
  80
  81         # Configure MDS to write one OMAP key at once
  82         self.set_conf('mds', 'mds_sessionmap_keys_per_op', 1)
  83         self.fs.mds_fail_restart()
  84         status = self.fs.wait_for_daemons()
  85
  86         # Bring the clients back
  87         self.mount_a.mount_wait()
  88         self.mount_b.mount_wait()
  89
  90         # See that they've got sessions
  91         self.assert_session_count(2, mds_id=self.fs.get_rank(status=status)['name'])
  92
  93         # See that we persist their sessions
  94         self.fs.rank_asok(["flush", "journal"], rank=0, status=status)
  95         table_json = json.loads(self.fs.table_tool(["0", "show", "session"]))
  96         log.info("SessionMap: {0}".format(json.dumps(table_json, indent=2)))
  97         self.assertEqual(table_json['0']['result'], 0)
  98         self.assertEqual(len(table_json['0']['data']['sessions']), 2)
  99
 100         # Now, induce a "force_open_sessions" event by exporting a dir
 101         self.mount_a.run_shell(["mkdir", "bravo"])
 102         self.mount_a.run_shell(["touch", "bravo/file_a"])
 103         self.mount_b.run_shell(["touch", "bravo/file_b"])
 104
 105         self.fs.set_max_mds(2)
 106         status = self.fs.wait_for_daemons()
 107
 108         def get_omap_wrs():
 109             return self.fs.rank_asok(['perf', 'dump', 'objecter'], rank=1, status=status)['objecter']['omap_wr']
 110
 111         # Flush so that there are no dirty sessions on rank 1
 112         self.fs.rank_asok(["flush", "journal"], rank=1, status=status)
 113
 114         # Export so that we get a force_open to rank 1 for the two sessions from rank 0
 115         initial_omap_wrs = get_omap_wrs()
 116         self.fs.rank_asok(['export', 'dir', '/bravo', '1'], rank=0, status=status)
 117
 118         # This is the critical (if rather subtle) check: that in the process of doing an export dir,
 119         # we hit force_open_sessions, and as a result we end up writing out the sessionmap.  There
 120         # will be two sessions dirtied here, and because we have set keys_per_op to 1, we should see
 121         # a single session get written out (the first of the two, triggered by the second getting marked
 122         # dirty)
 123         # The number of writes is two per session, because the header (sessionmap version) update and
 124         # KV write both count. Also, multiply by 2 for each openfile table update.
 125         self.wait_until_true(
 126             lambda: get_omap_wrs() - initial_omap_wrs == 2*2,
 127             timeout=30  # Long enough for an export to get acked
 128         )
 129
 130         # Now end our sessions and check the backing sessionmap is updated correctly
 131         self.mount_a.umount_wait()
 132         self.mount_b.umount_wait()
 133
 134         # In-memory sessionmap check
 135         self.assert_session_count(0, mds_id=self.fs.get_rank(status=status)['name'])
 136
 137         # On-disk sessionmap check
 138         self.fs.rank_asok(["flush", "journal"], rank=0, status=status)
 139         table_json = json.loads(self.fs.table_tool(["0", "show", "session"]))
 140         log.info("SessionMap: {0}".format(json.dumps(table_json, indent=2)))
 141         self.assertEqual(table_json['0']['result'], 0)
 142         self.assertEqual(len(table_json['0']['data']['sessions']), 0)
 143
 144     def _configure_auth(self, mount, id_name, mds_caps, osd_caps=None, mon_caps=None):
 145         """
 146         Set up auth credentials for a client mount, and write out the keyring
 147         for the client to use.
 148         """
 149
 150         if osd_caps is None:
 151             osd_caps = "allow rw"
 152
 153         if mon_caps is None:
 154             mon_caps = "allow r"
 155
 156         out = self.fs.mon_manager.raw_cluster_cmd(
 157             "auth", "get-or-create", "client.{name}".format(name=id_name),
 158             "mds", mds_caps,
 159             "osd", osd_caps,
 160             "mon", mon_caps
 161         )
 162         mount.client_id = id_name
 163         sudo_write_file(mount.client_remote, mount.get_keyring_path(), out)
 164         self.set_conf("client.{name}".format(name=id_name), "keyring", mount.get_keyring_path())
 165
 166     def test_session_reject(self):
 167         if not isinstance(self.mount_a, FuseMount):
 168             self.skipTest("Requires FUSE client to inject client metadata")
 169
 170         self.mount_a.run_shell(["mkdir", "foo"])
 171         self.mount_a.run_shell(["mkdir", "foo/bar"])
 172         self.mount_a.umount_wait()
 173
 174         # Mount B will be my rejected client
 175         self.mount_b.umount_wait()
 176
 177         # Configure a client that is limited to /foo/bar
 178         self._configure_auth(self.mount_b, "badguy", "allow rw path=/foo/bar")
 179         # Check he can mount that dir and do IO
 180         self.mount_b.mount_wait(mount_path="/foo/bar")
 181         self.mount_b.create_destroy()
 182         self.mount_b.umount_wait()
 183
 184         # Configure the client to claim that its mount point metadata is /baz
 185         self.set_conf("client.badguy", "client_metadata", "root=/baz")
 186         # Try to mount the client, see that it fails
 187         with self.assert_cluster_log("client session with non-allowable root '/baz' denied"):
 188             with self.assertRaises(CommandFailedError):
 189                 self.mount_b.mount(mount_path="/foo/bar")
 190
 191     def test_session_evict_blacklisted(self):
 192         """
 193         Check that mds evicts blacklisted client
 194         """
 195         if not isinstance(self.mount_a, FuseMount):
 196             self.skipTest("Requires FUSE client to use is_blacklisted()")
 197
 198         self.fs.set_max_mds(2)
 199         status = self.fs.wait_for_daemons()
 200
 201         self.mount_a.run_shell_payload("mkdir {d0,d1} && touch {d0,d1}/file")
 202         self.mount_a.setfattr("d0", "ceph.dir.pin", "0")
 203         self.mount_a.setfattr("d1", "ceph.dir.pin", "1")
 204         self._wait_subtrees([('/d0', 0), ('/d1', 1)], status=status)
 205
 206         self.mount_a.run_shell(["touch", "d0/f0"])
 207         self.mount_a.run_shell(["touch", "d1/f0"])
 208         self.mount_b.run_shell(["touch", "d0/f1"])
 209         self.mount_b.run_shell(["touch", "d1/f1"])
 210
 211         self.assert_session_count(2, mds_id=self.fs.get_rank(rank=0, status=status)['name'])
 212         self.assert_session_count(2, mds_id=self.fs.get_rank(rank=1, status=status)['name'])
 213
 214         mount_a_client_id = self.mount_a.get_global_id()
 215         self.fs.mds_asok(['session', 'evict', "%s" % mount_a_client_id],
 216                          mds_id=self.fs.get_rank(rank=0, status=status)['name'])
 217         self.wait_until_true(lambda: self.mount_a.is_blacklisted(), timeout=30)
 218
 219         # 10 seconds should be enough for evicting client
 220         time.sleep(10)
 221         self.assert_session_count(1, mds_id=self.fs.get_rank(rank=0, status=status)['name'])
 222         self.assert_session_count(1, mds_id=self.fs.get_rank(rank=1, status=status)['name'])
 223
 224         self.mount_a.kill_cleanup()
 225         self.mount_a.mount_wait()