ceph/qa/tasks/cephfs/fuse_mount.py

   1 from StringIO import StringIO
   2 import json
   3 import time
   4 import logging
   5 from textwrap import dedent
   6
   7 from teuthology import misc
   8 from teuthology.contextutil import MaxWhileTries
   9 from teuthology.orchestra import run
  10 from teuthology.orchestra.run import CommandFailedError
  11 from .mount import CephFSMount
  12
  13 log = logging.getLogger(__name__)
  14
  15
  16 class FuseMount(CephFSMount):
  17     def __init__(self, ctx, client_config, test_dir, client_id, client_remote):
  18         super(FuseMount, self).__init__(ctx, test_dir, client_id, client_remote)
  19
  20         self.client_config = client_config if client_config else {}
  21         self.fuse_daemon = None
  22         self._fuse_conn = None
  23         self.id = None
  24         self.inst = None
  25         self.addr = None
  26
  27     def mount(self, mount_path=None, mount_fs_name=None, mountpoint=None, mount_options=[]):
  28         if mountpoint is not None:
  29             self.mountpoint = mountpoint
  30         self.setupfs(name=mount_fs_name)
  31
  32         try:
  33             return self._mount(mount_path, mount_fs_name, mount_options)
  34         except RuntimeError:
  35             # Catch exceptions by the mount() logic (i.e. not remote command
  36             # failures) and ensure the mount is not left half-up.
  37             # Otherwise we might leave a zombie mount point that causes
  38             # anyone traversing cephtest/ to get hung up on.
  39             log.warn("Trying to clean up after failed mount")
  40             self.umount_wait(force=True)
  41             raise
  42
  43     def _mount(self, mount_path, mount_fs_name, mount_options):
  44         log.info("Client client.%s config is %s" % (self.client_id, self.client_config))
  45
  46         daemon_signal = 'kill'
  47         if self.client_config.get('coverage') or self.client_config.get('valgrind') is not None:
  48             daemon_signal = 'term'
  49
  50         log.info('Mounting ceph-fuse client.{id} at {remote} {mnt}...'.format(
  51             id=self.client_id, remote=self.client_remote, mnt=self.mountpoint))
  52
  53         self.client_remote.run(args=['mkdir', '-p', self.mountpoint],
  54                                timeout=(15*60), cwd=self.test_dir)
  55
  56         run_cmd = [
  57             'sudo',
  58             'adjust-ulimits',
  59             'ceph-coverage',
  60             '{tdir}/archive/coverage'.format(tdir=self.test_dir),
  61             'daemon-helper',
  62             daemon_signal,
  63         ]
  64
  65         fuse_cmd = ['ceph-fuse', "-f"]
  66
  67         if mount_path is not None:
  68             fuse_cmd += ["--client_mountpoint={0}".format(mount_path)]
  69
  70         if mount_fs_name is not None:
  71             fuse_cmd += ["--client_fs={0}".format(mount_fs_name)]
  72
  73         fuse_cmd += mount_options
  74
  75         fuse_cmd += [
  76             '--name', 'client.{id}'.format(id=self.client_id),
  77             # TODO ceph-fuse doesn't understand dash dash '--',
  78             self.mountpoint,
  79         ]
  80
  81         cwd = self.test_dir
  82         if self.client_config.get('valgrind') is not None:
  83             run_cmd = misc.get_valgrind_args(
  84                 self.test_dir,
  85                 'client.{id}'.format(id=self.client_id),
  86                 run_cmd,
  87                 self.client_config.get('valgrind'),
  88             )
  89             cwd = None # misc.get_valgrind_args chdir for us
  90
  91         run_cmd.extend(fuse_cmd)
  92
  93         def list_connections():
  94             self.client_remote.run(
  95                 args=["sudo", "mount", "-t", "fusectl", "/sys/fs/fuse/connections", "/sys/fs/fuse/connections"],
  96                 check_status=False,
  97                 timeout=(15*60)
  98             )
  99             p = self.client_remote.run(
 100                 args=["ls", "/sys/fs/fuse/connections"],
 101                 stdout=StringIO(),
 102                 check_status=False,
 103                 timeout=(15*60)
 104             )
 105             if p.exitstatus != 0:
 106                 return []
 107
 108             ls_str = p.stdout.getvalue().strip()
 109             if ls_str:
 110                 return [int(n) for n in ls_str.split("\n")]
 111             else:
 112                 return []
 113
 114         # Before starting ceph-fuse process, note the contents of
 115         # /sys/fs/fuse/connections
 116         pre_mount_conns = list_connections()
 117         log.info("Pre-mount connections: {0}".format(pre_mount_conns))
 118
 119         proc = self.client_remote.run(
 120             args=run_cmd,
 121             cwd=cwd,
 122             logger=log.getChild('ceph-fuse.{id}'.format(id=self.client_id)),
 123             stdin=run.PIPE,
 124             wait=False,
 125         )
 126         self.fuse_daemon = proc
 127
 128         # Wait for the connection reference to appear in /sys
 129         mount_wait = self.client_config.get('mount_wait', 0)
 130         if mount_wait > 0:
 131             log.info("Fuse mount waits {0} seconds before checking /sys/".format(mount_wait))
 132             time.sleep(mount_wait)
 133         timeout = int(self.client_config.get('mount_timeout', 30))
 134         waited = 0
 135
 136         post_mount_conns = list_connections()
 137         while len(post_mount_conns) <= len(pre_mount_conns):
 138             if self.fuse_daemon.finished:
 139                 # Did mount fail?  Raise the CommandFailedError instead of
 140                 # hitting the "failed to populate /sys/" timeout
 141                 self.fuse_daemon.wait()
 142             time.sleep(1)
 143             waited += 1
 144             if waited > timeout:
 145                 raise RuntimeError("Fuse mount failed to populate /sys/ after {0} seconds".format(
 146                     waited
 147                 ))
 148             else:
 149                 post_mount_conns = list_connections()
 150
 151         log.info("Post-mount connections: {0}".format(post_mount_conns))
 152
 153         # Record our fuse connection number so that we can use it when
 154         # forcing an unmount
 155         new_conns = list(set(post_mount_conns) - set(pre_mount_conns))
 156         if len(new_conns) == 0:
 157             raise RuntimeError("New fuse connection directory not found ({0})".format(new_conns))
 158         elif len(new_conns) > 1:
 159             raise RuntimeError("Unexpectedly numerous fuse connections {0}".format(new_conns))
 160         else:
 161             self._fuse_conn = new_conns[0]
 162
 163         self.gather_mount_info()
 164
 165     def gather_mount_info(self):
 166         status = self.admin_socket(['status'])
 167         self.id = status['id']
 168         self.client_pid = status['metadata']['pid']
 169         try:
 170             self.inst = status['inst_str']
 171             self.addr = status['addr_str']
 172         except KeyError:
 173             sessions = self.fs.rank_asok(['session', 'ls'])
 174             for s in sessions:
 175                 if s['id'] == self.id:
 176                     self.inst = s['inst']
 177                     self.addr = self.inst.split()[1]
 178             if self.inst is None:
 179                 raise RuntimeError("cannot find client session")
 180
 181     def is_mounted(self):
 182         proc = self.client_remote.run(
 183             args=[
 184                 'stat',
 185                 '--file-system',
 186                 '--printf=%T\n',
 187                 '--',
 188                 self.mountpoint,
 189             ],
 190             cwd=self.test_dir,
 191             stdout=StringIO(),
 192             stderr=StringIO(),
 193             wait=False,
 194             timeout=(15*60)
 195         )
 196         try:
 197             proc.wait()
 198         except CommandFailedError:
 199             if ("endpoint is not connected" in proc.stderr.getvalue()
 200             or "Software caused connection abort" in proc.stderr.getvalue()):
 201                 # This happens is fuse is killed without unmount
 202                 log.warn("Found stale moutn point at {0}".format(self.mountpoint))
 203                 return True
 204             else:
 205                 # This happens if the mount directory doesn't exist
 206                 log.info('mount point does not exist: %s', self.mountpoint)
 207                 return False
 208
 209         fstype = proc.stdout.getvalue().rstrip('\n')
 210         if fstype == 'fuseblk':
 211             log.info('ceph-fuse is mounted on %s', self.mountpoint)
 212             return True
 213         else:
 214             log.debug('ceph-fuse not mounted, got fs type {fstype!r}'.format(
 215                 fstype=fstype))
 216             return False
 217
 218     def wait_until_mounted(self):
 219         """
 220         Check to make sure that fuse is mounted on mountpoint.  If not,
 221         sleep for 5 seconds and check again.
 222         """
 223
 224         while not self.is_mounted():
 225             # Even if it's not mounted, it should at least
 226             # be running: catch simple failures where it has terminated.
 227             assert not self.fuse_daemon.poll()
 228
 229             time.sleep(5)
 230
 231         # Now that we're mounted, set permissions so that the rest of the test will have
 232         # unrestricted access to the filesystem mount.
 233         try:
 234             stderr = StringIO()
 235             self.client_remote.run(args=['sudo', 'chmod', '1777', self.mountpoint], timeout=(15*60), cwd=self.test_dir, stderr=stderr)
 236         except run.CommandFailedError:
 237             stderr = stderr.getvalue()
 238             if "Read-only file system".lower() in stderr.lower():
 239                 pass
 240             else:
 241                 raise
 242
 243     def _mountpoint_exists(self):
 244         return self.client_remote.run(args=["ls", "-d", self.mountpoint], check_status=False, cwd=self.test_dir, timeout=(15*60)).exitstatus == 0
 245
 246     def umount(self):
 247         try:
 248             log.info('Running fusermount -u on {name}...'.format(name=self.client_remote.name))
 249             self.client_remote.run(
 250                 args=[
 251                     'sudo',
 252                     'fusermount',
 253                     '-u',
 254                     self.mountpoint,
 255                 ],
 256                 cwd=self.test_dir,
 257                 timeout=(30*60),
 258             )
 259         except run.CommandFailedError:
 260             log.info('Failed to unmount ceph-fuse on {name}, aborting...'.format(name=self.client_remote.name))
 261
 262             self.client_remote.run(args=[
 263                 'sudo',
 264                 run.Raw('PATH=/usr/sbin:$PATH'),
 265                 'lsof',
 266                 run.Raw(';'),
 267                 'ps',
 268                 'auxf',
 269             ], timeout=(60*15))
 270
 271             # abort the fuse mount, killing all hung processes
 272             if self._fuse_conn:
 273                 self.run_python(dedent("""
 274                 import os
 275                 path = "/sys/fs/fuse/connections/{0}/abort"
 276                 if os.path.exists(path):
 277                     open(path, "w").write("1")
 278                 """).format(self._fuse_conn))
 279                 self._fuse_conn = None
 280
 281             stderr = StringIO()
 282             try:
 283                 # make sure its unmounted
 284                 self.client_remote.run(
 285                     args=[
 286                         'sudo',
 287                         'umount',
 288                         '-l',
 289                         '-f',
 290                         self.mountpoint,
 291                     ],
 292                     stderr=stderr,
 293                     timeout=(60*15)
 294                 )
 295             except CommandFailedError:
 296                 if self.is_mounted():
 297                     raise
 298
 299         assert not self.is_mounted()
 300         self._fuse_conn = None
 301         self.id = None
 302         self.inst = None
 303         self.addr = None
 304
 305     def umount_wait(self, force=False, require_clean=False, timeout=900):
 306         """
 307         :param force: Complete cleanly even if the MDS is offline
 308         """
 309         if not (self.is_mounted() and self.fuse_daemon):
 310             log.debug('ceph-fuse client.{id} is not mounted at {remote} {mnt}'.format(id=self.client_id,
 311                                                                                       remote=self.client_remote,
 312                                                                                       mnt=self.mountpoint))
 313             return
 314
 315         if force:
 316             assert not require_clean  # mutually exclusive
 317
 318             # When we expect to be forcing, kill the ceph-fuse process directly.
 319             # This should avoid hitting the more aggressive fallback killing
 320             # in umount() which can affect other mounts too.
 321             self.fuse_daemon.stdin.close()
 322
 323             # However, we will still hit the aggressive wait if there is an ongoing
 324             # mount -o remount (especially if the remount is stuck because MDSs
 325             # are unavailable)
 326
 327         self.umount()
 328
 329         try:
 330             # Permit a timeout, so that we do not block forever
 331             run.wait([self.fuse_daemon], timeout)
 332         except MaxWhileTries:
 333             log.error("process failed to terminate after unmount. This probably"
 334                       " indicates a bug within ceph-fuse.")
 335             raise
 336         except CommandFailedError:
 337             if require_clean:
 338                 raise
 339
 340         self.cleanup()
 341
 342     def cleanup(self):
 343         """
 344         Remove the mount point.
 345
 346         Prerequisite: the client is not mounted.
 347         """
 348         stderr = StringIO()
 349         try:
 350             self.client_remote.run(
 351                 args=[
 352                     'rmdir',
 353                     '--',
 354                     self.mountpoint,
 355                 ],
 356                 cwd=self.test_dir,
 357                 stderr=stderr,
 358                 timeout=(60*5),
 359                 check_status=False,
 360             )
 361         except CommandFailedError:
 362             if "No such file or directory" in stderr.getvalue():
 363                 pass
 364             else:
 365                 raise
 366
 367     def kill(self):
 368         """
 369         Terminate the client without removing the mount point.
 370         """
 371         log.info('Killing ceph-fuse connection on {name}...'.format(name=self.client_remote.name))
 372         self.fuse_daemon.stdin.close()
 373         try:
 374             self.fuse_daemon.wait()
 375         except CommandFailedError:
 376             pass
 377
 378     def kill_cleanup(self):
 379         """
 380         Follow up ``kill`` to get to a clean unmounted state.
 381         """
 382         log.info('Cleaning up killed ceph-fuse connection')
 383         self.umount()
 384         self.cleanup()
 385
 386     def teardown(self):
 387         """
 388         Whatever the state of the mount, get it gone.
 389         """
 390         super(FuseMount, self).teardown()
 391
 392         self.umount()
 393
 394         if self.fuse_daemon and not self.fuse_daemon.finished:
 395             self.fuse_daemon.stdin.close()
 396             try:
 397                 self.fuse_daemon.wait()
 398             except CommandFailedError:
 399                 pass
 400
 401         # Indiscriminate, unlike the touchier cleanup()
 402         self.client_remote.run(
 403             args=[
 404                 'rm',
 405                 '-rf',
 406                 self.mountpoint,
 407             ],
 408             cwd=self.test_dir,
 409             timeout=(60*5)
 410         )
 411
 412     def _asok_path(self):
 413         return "/var/run/ceph/ceph-client.{0}.*.asok".format(self.client_id)
 414
 415     @property
 416     def _prefix(self):
 417         return ""
 418
 419     def admin_socket(self, args):
 420         pyscript = """
 421 import glob
 422 import re
 423 import os
 424 import subprocess
 425
 426 def find_socket(client_name):
 427         asok_path = "{asok_path}"
 428         files = glob.glob(asok_path)
 429
 430         # Given a non-glob path, it better be there
 431         if "*" not in asok_path:
 432             assert(len(files) == 1)
 433             return files[0]
 434
 435         for f in files:
 436                 pid = re.match(".*\.(\d+)\.asok$", f).group(1)
 437                 if os.path.exists("/proc/{{0}}".format(pid)):
 438                         return f
 439         raise RuntimeError("Client socket {{0}} not found".format(client_name))
 440
 441 print(find_socket("{client_name}"))
 442 """.format(
 443             asok_path=self._asok_path(),
 444             client_name="client.{0}".format(self.client_id))
 445
 446         # Find the admin socket
 447         p = self.client_remote.run(args=[
 448             'sudo', 'python3', '-c', pyscript
 449         ], stdout=StringIO(), timeout=(15*60))
 450         asok_path = p.stdout.getvalue().strip()
 451         log.info("Found client admin socket at {0}".format(asok_path))
 452
 453         # Query client ID from admin socket
 454         p = self.client_remote.run(
 455             args=['sudo', self._prefix + 'ceph', '--admin-daemon', asok_path] + args,
 456             stdout=StringIO(), timeout=(15*60))
 457         return json.loads(p.stdout.getvalue())
 458
 459     def get_global_id(self):
 460         """
 461         Look up the CephFS client ID for this mount
 462         """
 463         return self.admin_socket(['mds_sessions'])['id']
 464
 465     def get_global_inst(self):
 466         """
 467         Look up the CephFS client instance for this mount
 468         """
 469         return self.inst
 470
 471     def get_global_addr(self):
 472         """
 473         Look up the CephFS client addr for this mount
 474         """
 475         return self.addr
 476
 477     def get_client_pid(self):
 478         """
 479         return pid of ceph-fuse process
 480         """
 481         status = self.admin_socket(['status'])
 482         return status['metadata']['pid']
 483
 484     def get_osd_epoch(self):
 485         """
 486         Return 2-tuple of osd_epoch, osd_epoch_barrier
 487         """
 488         status = self.admin_socket(['status'])
 489         return status['osd_epoch'], status['osd_epoch_barrier']
 490
 491     def get_dentry_count(self):
 492         """
 493         Return 2-tuple of dentry_count, dentry_pinned_count
 494         """
 495         status = self.admin_socket(['status'])
 496         return status['dentry_count'], status['dentry_pinned_count']
 497
 498     def set_cache_size(self, size):
 499         return self.admin_socket(['config', 'set', 'client_cache_size', str(size)])