+++ /dev/null
-
-from StringIO import StringIO
-import json
-import time
-import logging
-from textwrap import dedent
-
-from teuthology import misc
-from teuthology.contextutil import MaxWhileTries
-from teuthology.orchestra import run
-from teuthology.orchestra.run import CommandFailedError
-from .mount import CephFSMount
-
-log = logging.getLogger(__name__)
-
-
-class FuseMount(CephFSMount):
- def __init__(self, client_config, test_dir, client_id, client_remote):
- super(FuseMount, self).__init__(test_dir, client_id, client_remote)
-
- self.client_config = client_config if client_config else {}
- self.fuse_daemon = None
- self._fuse_conn = None
-
- def mount(self, mount_path=None, mount_fs_name=None):
- try:
- return self._mount(mount_path, mount_fs_name)
- except RuntimeError:
- # Catch exceptions by the mount() logic (i.e. not remote command
- # failures) and ensure the mount is not left half-up.
- # Otherwise we might leave a zombie mount point that causes
- # anyone traversing cephtest/ to get hung up on.
- log.warn("Trying to clean up after failed mount")
- self.umount_wait(force=True)
- raise
-
- def _mount(self, mount_path, mount_fs_name):
- log.info("Client client.%s config is %s" % (self.client_id, self.client_config))
-
- daemon_signal = 'kill'
- if self.client_config.get('coverage') or self.client_config.get('valgrind') is not None:
- daemon_signal = 'term'
-
- log.info('Mounting ceph-fuse client.{id} at {remote} {mnt}...'.format(
- id=self.client_id, remote=self.client_remote, mnt=self.mountpoint))
-
- self.client_remote.run(
- args=[
- 'mkdir',
- '--',
- self.mountpoint,
- ],
- )
-
- run_cmd = [
- 'sudo',
- 'adjust-ulimits',
- 'ceph-coverage',
- '{tdir}/archive/coverage'.format(tdir=self.test_dir),
- 'daemon-helper',
- daemon_signal,
- ]
-
- fuse_cmd = ['ceph-fuse', "-f"]
-
- if mount_path is not None:
- fuse_cmd += ["--client_mountpoint={0}".format(mount_path)]
-
- if mount_fs_name is not None:
- fuse_cmd += ["--client_mds_namespace={0}".format(mount_fs_name)]
-
- fuse_cmd += [
- '--name', 'client.{id}'.format(id=self.client_id),
- # TODO ceph-fuse doesn't understand dash dash '--',
- self.mountpoint,
- ]
-
- if self.client_config.get('valgrind') is not None:
- run_cmd = misc.get_valgrind_args(
- self.test_dir,
- 'client.{id}'.format(id=self.client_id),
- run_cmd,
- self.client_config.get('valgrind'),
- )
-
- run_cmd.extend(fuse_cmd)
-
- def list_connections():
- self.client_remote.run(
- args=["sudo", "mount", "-t", "fusectl", "/sys/fs/fuse/connections", "/sys/fs/fuse/connections"],
- check_status=False
- )
- p = self.client_remote.run(
- args=["ls", "/sys/fs/fuse/connections"],
- stdout=StringIO(),
- check_status=False
- )
- if p.exitstatus != 0:
- return []
-
- ls_str = p.stdout.getvalue().strip()
- if ls_str:
- return [int(n) for n in ls_str.split("\n")]
- else:
- return []
-
- # Before starting ceph-fuse process, note the contents of
- # /sys/fs/fuse/connections
- pre_mount_conns = list_connections()
- log.info("Pre-mount connections: {0}".format(pre_mount_conns))
-
- proc = self.client_remote.run(
- args=run_cmd,
- logger=log.getChild('ceph-fuse.{id}'.format(id=self.client_id)),
- stdin=run.PIPE,
- wait=False,
- )
- self.fuse_daemon = proc
-
- # Wait for the connection reference to appear in /sys
- mount_wait = self.client_config.get('mount_wait', 0)
- if mount_wait > 0:
- log.info("Fuse mount waits {0} seconds before checking /sys/".format(mount_wait))
- time.sleep(mount_wait)
- timeout = int(self.client_config.get('mount_timeout', 30))
- waited = 0
-
- post_mount_conns = list_connections()
- while len(post_mount_conns) <= len(pre_mount_conns):
- if self.fuse_daemon.finished:
- # Did mount fail? Raise the CommandFailedError instead of
- # hitting the "failed to populate /sys/" timeout
- self.fuse_daemon.wait()
- time.sleep(1)
- waited += 1
- if waited > timeout:
- raise RuntimeError("Fuse mount failed to populate /sys/ after {0} seconds".format(
- waited
- ))
- else:
- post_mount_conns = list_connections()
-
- log.info("Post-mount connections: {0}".format(post_mount_conns))
-
- # Record our fuse connection number so that we can use it when
- # forcing an unmount
- new_conns = list(set(post_mount_conns) - set(pre_mount_conns))
- if len(new_conns) == 0:
- raise RuntimeError("New fuse connection directory not found ({0})".format(new_conns))
- elif len(new_conns) > 1:
- raise RuntimeError("Unexpectedly numerous fuse connections {0}".format(new_conns))
- else:
- self._fuse_conn = new_conns[0]
-
- def is_mounted(self):
- proc = self.client_remote.run(
- args=[
- 'stat',
- '--file-system',
- '--printf=%T\n',
- '--',
- self.mountpoint,
- ],
- stdout=StringIO(),
- stderr=StringIO(),
- wait=False
- )
- try:
- proc.wait()
- except CommandFailedError:
- if ("endpoint is not connected" in proc.stderr.getvalue()
- or "Software caused connection abort" in proc.stderr.getvalue()):
- # This happens is fuse is killed without unmount
- log.warn("Found stale moutn point at {0}".format(self.mountpoint))
- return True
- else:
- # This happens if the mount directory doesn't exist
- log.info('mount point does not exist: %s', self.mountpoint)
- return False
-
- fstype = proc.stdout.getvalue().rstrip('\n')
- if fstype == 'fuseblk':
- log.info('ceph-fuse is mounted on %s', self.mountpoint)
- return True
- else:
- log.debug('ceph-fuse not mounted, got fs type {fstype!r}'.format(
- fstype=fstype))
- return False
-
- def wait_until_mounted(self):
- """
- Check to make sure that fuse is mounted on mountpoint. If not,
- sleep for 5 seconds and check again.
- """
-
- while not self.is_mounted():
- # Even if it's not mounted, it should at least
- # be running: catch simple failures where it has terminated.
- assert not self.fuse_daemon.poll()
-
- time.sleep(5)
-
- # Now that we're mounted, set permissions so that the rest of the test will have
- # unrestricted access to the filesystem mount.
- self.client_remote.run(
- args=['sudo', 'chmod', '1777', self.mountpoint])
-
- def _mountpoint_exists(self):
- return self.client_remote.run(args=["ls", "-d", self.mountpoint], check_status=False).exitstatus == 0
-
- def umount(self):
- try:
- log.info('Running fusermount -u on {name}...'.format(name=self.client_remote.name))
- self.client_remote.run(
- args=[
- 'sudo',
- 'fusermount',
- '-u',
- self.mountpoint,
- ],
- )
- except run.CommandFailedError:
- log.info('Failed to unmount ceph-fuse on {name}, aborting...'.format(name=self.client_remote.name))
-
- self.client_remote.run(args=[
- 'sudo',
- run.Raw('PATH=/usr/sbin:$PATH'),
- 'lsof',
- run.Raw(';'),
- 'ps',
- 'auxf',
- ])
-
- # abort the fuse mount, killing all hung processes
- if self._fuse_conn:
- self.run_python(dedent("""
- import os
- path = "/sys/fs/fuse/connections/{0}/abort"
- if os.path.exists(path):
- open(path, "w").write("1")
- """).format(self._fuse_conn))
- self._fuse_conn = None
-
- stderr = StringIO()
- try:
- # make sure its unmounted
- self.client_remote.run(
- args=[
- 'sudo',
- 'umount',
- '-l',
- '-f',
- self.mountpoint,
- ],
- stderr=stderr
- )
- except CommandFailedError:
- if self.is_mounted():
- raise
-
- assert not self.is_mounted()
- self._fuse_conn = None
-
- def umount_wait(self, force=False, require_clean=False):
- """
- :param force: Complete cleanly even if the MDS is offline
- """
- if force:
- assert not require_clean # mutually exclusive
-
- # When we expect to be forcing, kill the ceph-fuse process directly.
- # This should avoid hitting the more aggressive fallback killing
- # in umount() which can affect other mounts too.
- self.fuse_daemon.stdin.close()
-
- # However, we will still hit the aggressive wait if there is an ongoing
- # mount -o remount (especially if the remount is stuck because MDSs
- # are unavailable)
-
- self.umount()
-
- try:
- if self.fuse_daemon:
- # Permit a timeout, so that we do not block forever
- run.wait([self.fuse_daemon], 900)
- except MaxWhileTries:
- log.error("process failed to terminate after unmount. This probably"
- "indicates a bug within ceph-fuse.")
- raise
- except CommandFailedError:
- if require_clean:
- raise
-
- self.cleanup()
-
- def cleanup(self):
- """
- Remove the mount point.
-
- Prerequisite: the client is not mounted.
- """
- stderr = StringIO()
- try:
- self.client_remote.run(
- args=[
- 'rmdir',
- '--',
- self.mountpoint,
- ],
- stderr=stderr
- )
- except CommandFailedError:
- if "No such file or directory" in stderr.getvalue():
- pass
- else:
- raise
-
- def kill(self):
- """
- Terminate the client without removing the mount point.
- """
- self.fuse_daemon.stdin.close()
- try:
- self.fuse_daemon.wait()
- except CommandFailedError:
- pass
-
- def kill_cleanup(self):
- """
- Follow up ``kill`` to get to a clean unmounted state.
- """
- self.umount()
- self.cleanup()
-
- def teardown(self):
- """
- Whatever the state of the mount, get it gone.
- """
- super(FuseMount, self).teardown()
-
- self.umount()
-
- if self.fuse_daemon and not self.fuse_daemon.finished:
- self.fuse_daemon.stdin.close()
- try:
- self.fuse_daemon.wait()
- except CommandFailedError:
- pass
-
- # Indiscriminate, unlike the touchier cleanup()
- self.client_remote.run(
- args=[
- 'rm',
- '-rf',
- self.mountpoint,
- ],
- )
-
- def _asok_path(self):
- return "/var/run/ceph/ceph-client.{0}.*.asok".format(self.client_id)
-
- @property
- def _prefix(self):
- return ""
-
- def admin_socket(self, args):
- pyscript = """
-import glob
-import re
-import os
-import subprocess
-
-def find_socket(client_name):
- asok_path = "{asok_path}"
- files = glob.glob(asok_path)
-
- # Given a non-glob path, it better be there
- if "*" not in asok_path:
- assert(len(files) == 1)
- return files[0]
-
- for f in files:
- pid = re.match(".*\.(\d+)\.asok$", f).group(1)
- if os.path.exists("/proc/{{0}}".format(pid)):
- return f
- raise RuntimeError("Client socket {{0}} not found".format(client_name))
-
-print find_socket("{client_name}")
-""".format(
- asok_path=self._asok_path(),
- client_name="client.{0}".format(self.client_id))
-
- # Find the admin socket
- p = self.client_remote.run(args=[
- 'python', '-c', pyscript
- ], stdout=StringIO())
- asok_path = p.stdout.getvalue().strip()
- log.info("Found client admin socket at {0}".format(asok_path))
-
- # Query client ID from admin socket
- p = self.client_remote.run(
- args=['sudo', self._prefix + 'ceph', '--admin-daemon', asok_path] + args,
- stdout=StringIO())
- return json.loads(p.stdout.getvalue())
-
- def get_global_id(self):
- """
- Look up the CephFS client ID for this mount
- """
-
- return self.admin_socket(['mds_sessions'])['id']
-
- def get_osd_epoch(self):
- """
- Return 2-tuple of osd_epoch, osd_epoch_barrier
- """
- status = self.admin_socket(['status'])
- return status['osd_epoch'], status['osd_epoch_barrier']
-
- def get_dentry_count(self):
- """
- Return 2-tuple of dentry_count, dentry_pinned_count
- """
- status = self.admin_socket(['status'])
- return status['dentry_count'], status['dentry_pinned_count']
-
- def set_cache_size(self, size):
- return self.admin_socket(['config', 'set', 'client_cache_size', str(size)])
Code Review / stor4nfv.git / blobdiff