--- /dev/null
+"""
+Copyright (C) 2015 Red Hat, Inc.
+
+LGPL2. See file COPYING.
+"""
+
+from contextlib import contextmanager
+import errno
+import fcntl
+import json
+import logging
+import os
+import re
+import struct
+import sys
+import threading
+import time
+import uuid
+
+from ceph_argparse import json_command
+
+import cephfs
+import rados
+
+
+class RadosError(Exception):
+ """
+ Something went wrong talking to Ceph with librados
+ """
+ pass
+
+
+RADOS_TIMEOUT = 10
+
+log = logging.getLogger(__name__)
+
+
+# Reserved volume group name which we use in paths for volumes
+# that are not assigned to a group (i.e. created with group=None)
+NO_GROUP_NAME = "_nogroup"
+
+# Filename extensions for meta files.
+META_FILE_EXT = ".meta"
+
+class VolumePath(object):
+ """
+ Identify a volume's path as group->volume
+ The Volume ID is a unique identifier, but this is a much more
+ helpful thing to pass around.
+ """
+ def __init__(self, group_id, volume_id):
+ self.group_id = group_id
+ self.volume_id = volume_id
+ assert self.group_id != NO_GROUP_NAME
+ assert self.volume_id != "" and self.volume_id is not None
+
+ def __str__(self):
+ return "{0}/{1}".format(self.group_id, self.volume_id)
+
+
+class ClusterTimeout(Exception):
+ """
+ Exception indicating that we timed out trying to talk to the Ceph cluster,
+ either to the mons, or to any individual daemon that the mons indicate ought
+ to be up but isn't responding to us.
+ """
+ pass
+
+
+class ClusterError(Exception):
+ """
+ Exception indicating that the cluster returned an error to a command that
+ we thought should be successful based on our last knowledge of the cluster
+ state.
+ """
+ def __init__(self, action, result_code, result_str):
+ self._action = action
+ self._result_code = result_code
+ self._result_str = result_str
+
+ def __str__(self):
+ return "Error {0} (\"{1}\") while {2}".format(
+ self._result_code, self._result_str, self._action)
+
+
+class RankEvicter(threading.Thread):
+ """
+ Thread for evicting client(s) from a particular MDS daemon instance.
+
+ This is more complex than simply sending a command, because we have to
+ handle cases where MDS daemons might not be fully up yet, and/or might
+ be transiently unresponsive to commands.
+ """
+ class GidGone(Exception):
+ pass
+
+ POLL_PERIOD = 5
+
+ def __init__(self, volume_client, client_spec, rank, gid, mds_map, ready_timeout):
+ """
+ :param client_spec: list of strings, used as filter arguments to "session evict"
+ pass ["id=123"] to evict a single client with session id 123.
+ """
+ self.rank = rank
+ self.gid = gid
+ self._mds_map = mds_map
+ self._client_spec = client_spec
+ self._volume_client = volume_client
+ self._ready_timeout = ready_timeout
+ self._ready_waited = 0
+
+ self.success = False
+ self.exception = None
+
+ super(RankEvicter, self).__init__()
+
+ def _ready_to_evict(self):
+ if self._mds_map['up'].get("mds_{0}".format(self.rank), None) != self.gid:
+ log.info("Evicting {0} from {1}/{2}: rank no longer associated with gid, done.".format(
+ self._client_spec, self.rank, self.gid
+ ))
+ raise RankEvicter.GidGone()
+
+ info = self._mds_map['info']["gid_{0}".format(self.gid)]
+ log.debug("_ready_to_evict: state={0}".format(info['state']))
+ return info['state'] in ["up:active", "up:clientreplay"]
+
+ def _wait_for_ready(self):
+ """
+ Wait for that MDS rank to reach an active or clientreplay state, and
+ not be laggy.
+ """
+ while not self._ready_to_evict():
+ if self._ready_waited > self._ready_timeout:
+ raise ClusterTimeout()
+
+ time.sleep(self.POLL_PERIOD)
+ self._ready_waited += self.POLL_PERIOD
+
+ self._mds_map = self._volume_client._rados_command("mds dump", {})
+
+ def _evict(self):
+ """
+ Run the eviction procedure. Return true on success, false on errors.
+ """
+
+ # Wait til the MDS is believed by the mon to be available for commands
+ try:
+ self._wait_for_ready()
+ except self.GidGone:
+ return True
+
+ # Then send it an evict
+ ret = errno.ETIMEDOUT
+ while ret == errno.ETIMEDOUT:
+ log.debug("mds_command: {0}, {1}".format(
+ "%s" % self.gid, ["session", "evict"] + self._client_spec
+ ))
+ ret, outb, outs = self._volume_client.fs.mds_command(
+ "%s" % self.gid,
+ [json.dumps({
+ "prefix": "session evict",
+ "filters": self._client_spec
+ })], "")
+ log.debug("mds_command: complete {0} {1}".format(ret, outs))
+
+ # If we get a clean response, great, it's gone from that rank.
+ if ret == 0:
+ return True
+ elif ret == errno.ETIMEDOUT:
+ # Oh no, the MDS went laggy (that's how libcephfs knows to emit this error)
+ self._mds_map = self._volume_client._rados_command("mds dump", {})
+ try:
+ self._wait_for_ready()
+ except self.GidGone:
+ return True
+ else:
+ raise ClusterError("Sending evict to mds.{0}".format(self.gid), ret, outs)
+
+ def run(self):
+ try:
+ self._evict()
+ except Exception as e:
+ self.success = False
+ self.exception = e
+ else:
+ self.success = True
+
+
+class EvictionError(Exception):
+ pass
+
+
+class CephFSVolumeClientError(Exception):
+ """
+ Something went wrong talking to Ceph using CephFSVolumeClient.
+ """
+ pass
+
+
+CEPHFSVOLUMECLIENT_VERSION_HISTORY = """
+
+ CephFSVolumeClient Version History:
+
+ * 1 - Initial version
+ * 2 - Added get_object, put_object, delete_object methods to CephFSVolumeClient
+
+"""
+
+
+class CephFSVolumeClient(object):
+ """
+ Combine libcephfs and librados interfaces to implement a
+ 'Volume' concept implemented as a cephfs directory and
+ client capabilities which restrict mount access to this
+ directory.
+
+ Additionally, volumes may be in a 'Group'. Conveniently,
+ volumes are a lot like manila shares, and groups are a lot
+ like manila consistency groups.
+
+ Refer to volumes with VolumePath, which specifies the
+ volume and group IDs (both strings). The group ID may
+ be None.
+
+ In general, functions in this class are allowed raise rados.Error
+ or cephfs.Error exceptions in unexpected situations.
+ """
+
+ # Current version
+ version = 2
+
+ # Where shall we create our volumes?
+ POOL_PREFIX = "fsvolume_"
+ DEFAULT_VOL_PREFIX = "/volumes"
+ DEFAULT_NS_PREFIX = "fsvolumens_"
+
+ def __init__(self, auth_id, conf_path, cluster_name, volume_prefix=None, pool_ns_prefix=None):
+ self.fs = None
+ self.rados = None
+ self.connected = False
+ self.conf_path = conf_path
+ self.cluster_name = cluster_name
+ self.auth_id = auth_id
+ self.volume_prefix = volume_prefix if volume_prefix else self.DEFAULT_VOL_PREFIX
+ self.pool_ns_prefix = pool_ns_prefix if pool_ns_prefix else self.DEFAULT_NS_PREFIX
+ # For flock'ing in cephfs, I want a unique ID to distinguish me
+ # from any other manila-share services that are loading this module.
+ # We could use pid, but that's unnecessary weak: generate a
+ # UUID
+ self._id = struct.unpack(">Q", uuid.uuid1().get_bytes()[0:8])[0]
+
+ # TODO: version the on-disk structures
+
+ def recover(self):
+ # Scan all auth keys to see if they're dirty: if they are, they have
+ # state that might not have propagated to Ceph or to the related
+ # volumes yet.
+
+ # Important: we *always* acquire locks in the order auth->volume
+ # That means a volume can never be dirty without the auth key
+ # we're updating it with being dirty at the same time.
+
+ # First list the auth IDs that have potentially dirty on-disk metadata
+ log.debug("Recovering from partial auth updates (if any)...")
+
+ try:
+ dir_handle = self.fs.opendir(self.volume_prefix)
+ except cephfs.ObjectNotFound:
+ log.debug("Nothing to recover. No auth meta files.")
+ return
+
+ d = self.fs.readdir(dir_handle)
+ auth_ids = []
+
+ if not d:
+ log.debug("Nothing to recover. No auth meta files.")
+
+ while d:
+ # Identify auth IDs from auth meta filenames. The auth meta files
+ # are named as, "$<auth_id><meta filename extension>"
+ regex = "^\$(.*){0}$".format(re.escape(META_FILE_EXT))
+ match = re.search(regex, d.d_name)
+ if match:
+ auth_ids.append(match.group(1))
+
+ d = self.fs.readdir(dir_handle)
+
+ self.fs.closedir(dir_handle)
+
+ # Key points based on ordering:
+ # * Anything added in VMeta is already added in AMeta
+ # * Anything added in Ceph is already added in VMeta
+ # * Anything removed in VMeta is already removed in Ceph
+ # * Anything removed in AMeta is already removed in VMeta
+
+ # Deauthorization: because I only update metadata AFTER the
+ # update of the next level down, I have the same ordering of
+ # -> things which exist in the AMeta should also exist
+ # in the VMeta, should also exist in Ceph, and the same
+ # recovery procedure that gets me consistent after crashes
+ # during authorization will also work during deauthorization
+
+ # Now for each auth ID, check for dirty flag and apply updates
+ # if dirty flag is found
+ for auth_id in auth_ids:
+ with self._auth_lock(auth_id):
+ auth_meta = self._auth_metadata_get(auth_id)
+ if not auth_meta or not auth_meta['volumes']:
+ # Clean up auth meta file
+ self.fs.unlink(self._auth_metadata_path(auth_id))
+ continue
+ if not auth_meta['dirty']:
+ continue
+ self._recover_auth_meta(auth_id, auth_meta)
+
+ log.debug("Recovered from partial auth updates (if any).")
+
+ def _recover_auth_meta(self, auth_id, auth_meta):
+ """
+ Call me after locking the auth meta file.
+ """
+ remove_volumes = []
+
+ for volume, volume_data in auth_meta['volumes'].items():
+ if not volume_data['dirty']:
+ continue
+
+ (group_id, volume_id) = volume.split('/')
+ group_id = group_id if group_id is not 'None' else None
+ volume_path = VolumePath(group_id, volume_id)
+ access_level = volume_data['access_level']
+
+ with self._volume_lock(volume_path):
+ vol_meta = self._volume_metadata_get(volume_path)
+
+ # No VMeta update indicates that there was no auth update
+ # in Ceph either. So it's safe to remove corresponding
+ # partial update in AMeta.
+ if not vol_meta or auth_id not in vol_meta['auths']:
+ remove_volumes.append(volume)
+ continue
+
+ want_auth = {
+ 'access_level': access_level,
+ 'dirty': False,
+ }
+ # VMeta update looks clean. Ceph auth update must have been
+ # clean.
+ if vol_meta['auths'][auth_id] == want_auth:
+ continue
+
+ readonly = True if access_level is 'r' else False
+ self._authorize_volume(volume_path, auth_id, readonly)
+
+ # Recovered from partial auth updates for the auth ID's access
+ # to a volume.
+ auth_meta['volumes'][volume]['dirty'] = False
+ self._auth_metadata_set(auth_id, auth_meta)
+
+ for volume in remove_volumes:
+ del auth_meta['volumes'][volume]
+
+ if not auth_meta['volumes']:
+ # Clean up auth meta file
+ self.fs.unlink(self._auth_metadata_path(auth_id))
+ return
+
+ # Recovered from all partial auth updates for the auth ID.
+ auth_meta['dirty'] = False
+ self._auth_metadata_set(auth_id, auth_meta)
+
+
+ def evict(self, auth_id, timeout=30, volume_path=None):
+ """
+ Evict all clients based on the authorization ID and optionally based on
+ the volume path mounted. Assumes that the authorization key has been
+ revoked prior to calling this function.
+
+ This operation can throw an exception if the mon cluster is unresponsive, or
+ any individual MDS daemon is unresponsive for longer than the timeout passed in.
+ """
+
+ client_spec = ["auth_name={0}".format(auth_id), ]
+ if volume_path:
+ client_spec.append("client_metadata.root={0}".
+ format(self._get_path(volume_path)))
+
+ log.info("evict clients with {0}".format(', '.join(client_spec)))
+
+ mds_map = self._rados_command("mds dump", {})
+
+ up = {}
+ for name, gid in mds_map['up'].items():
+ # Quirk of the MDSMap JSON dump: keys in the up dict are like "mds_0"
+ assert name.startswith("mds_")
+ up[int(name[4:])] = gid
+
+ # For all MDS ranks held by a daemon
+ # Do the parallelism in python instead of using "tell mds.*", because
+ # the latter doesn't give us per-mds output
+ threads = []
+ for rank, gid in up.items():
+ thread = RankEvicter(self, client_spec, rank, gid, mds_map,
+ timeout)
+ thread.start()
+ threads.append(thread)
+
+ for t in threads:
+ t.join()
+
+ log.info("evict: joined all")
+
+ for t in threads:
+ if not t.success:
+ msg = ("Failed to evict client with {0} from mds {1}/{2}: {3}".
+ format(', '.join(client_spec), t.rank, t.gid, t.exception)
+ )
+ log.error(msg)
+ raise EvictionError(msg)
+
+ def _get_path(self, volume_path):
+ """
+ Determine the path within CephFS where this volume will live
+ :return: absolute path (string)
+ """
+ return os.path.join(
+ self.volume_prefix,
+ volume_path.group_id if volume_path.group_id is not None else NO_GROUP_NAME,
+ volume_path.volume_id)
+
+ def _get_group_path(self, group_id):
+ if group_id is None:
+ raise ValueError("group_id may not be None")
+
+ return os.path.join(
+ self.volume_prefix,
+ group_id
+ )
+
+ def connect(self, premount_evict = None):
+ """
+
+ :param premount_evict: Optional auth_id to evict before mounting the filesystem: callers
+ may want to use this to specify their own auth ID if they expect
+ to be a unique instance and don't want to wait for caps to time
+ out after failure of another instance of themselves.
+ """
+ log.debug("Connecting to RADOS with config {0}...".format(self.conf_path))
+ self.rados = rados.Rados(
+ name="client.{0}".format(self.auth_id),
+ clustername=self.cluster_name,
+ conffile=self.conf_path,
+ conf={}
+ )
+ self.rados.connect()
+
+ log.debug("Connection to RADOS complete")
+
+ log.debug("Connecting to cephfs...")
+ self.fs = cephfs.LibCephFS(rados_inst=self.rados)
+ log.debug("CephFS initializing...")
+ self.fs.init()
+ if premount_evict is not None:
+ log.debug("Premount eviction of {0} starting".format(premount_evict))
+ self.evict(premount_evict)
+ log.debug("Premount eviction of {0} completes".format(premount_evict))
+ log.debug("CephFS mounting...")
+ self.fs.mount()
+ log.debug("Connection to cephfs complete")
+
+ # Recover from partial auth updates due to a previous
+ # crash.
+ self.recover()
+
+ def get_mon_addrs(self):
+ log.info("get_mon_addrs")
+ result = []
+ mon_map = self._rados_command("mon dump")
+ for mon in mon_map['mons']:
+ ip_port = mon['addr'].split("/")[0]
+ result.append(ip_port)
+
+ return result
+
+ def disconnect(self):
+ log.info("disconnect")
+ if self.fs:
+ log.debug("Disconnecting cephfs...")
+ self.fs.shutdown()
+ self.fs = None
+ log.debug("Disconnecting cephfs complete")
+
+ if self.rados:
+ log.debug("Disconnecting rados...")
+ self.rados.shutdown()
+ self.rados = None
+ log.debug("Disconnecting rados complete")
+
+ def __del__(self):
+ self.disconnect()
+
+ def _get_pool_id(self, osd_map, pool_name):
+ # Maybe borrow the OSDMap wrapper class from calamari if more helpers
+ # like this are needed.
+ for pool in osd_map['pools']:
+ if pool['pool_name'] == pool_name:
+ return pool['pool']
+
+ return None
+
+ def _create_volume_pool(self, pool_name):
+ """
+ Idempotently create a pool for use as a CephFS data pool, with the given name
+
+ :return The ID of the created pool
+ """
+ osd_map = self._rados_command('osd dump', {})
+
+ existing_id = self._get_pool_id(osd_map, pool_name)
+ if existing_id is not None:
+ log.info("Pool {0} already exists".format(pool_name))
+ return existing_id
+
+ osd_count = len(osd_map['osds'])
+
+ # We can't query the actual cluster config remotely, but since this is
+ # just a heuristic we'll assume that the ceph.conf we have locally reflects
+ # that in use in the rest of the cluster.
+ pg_warn_max_per_osd = int(self.rados.conf_get('mon_max_pg_per_osd'))
+
+ other_pgs = 0
+ for pool in osd_map['pools']:
+ if not pool['pool_name'].startswith(self.POOL_PREFIX):
+ other_pgs += pool['pg_num']
+
+ # A basic heuristic for picking pg_num: work out the max number of
+ # PGs we can have without tripping a warning, then subtract the number
+ # of PGs already created by non-manila pools, then divide by ten. That'll
+ # give you a reasonable result on a system where you have "a few" manila
+ # shares.
+ pg_num = ((pg_warn_max_per_osd * osd_count) - other_pgs) / 10
+ # TODO Alternatively, respect an override set by the user.
+
+ self._rados_command(
+ 'osd pool create',
+ {
+ 'pool': pool_name,
+ 'pg_num': pg_num
+ }
+ )
+
+ osd_map = self._rados_command('osd dump', {})
+ pool_id = self._get_pool_id(osd_map, pool_name)
+
+ if pool_id is None:
+ # If the pool isn't there, that's either a ceph bug, or it's some outside influence
+ # removing it right after we created it.
+ log.error("OSD map doesn't contain expected pool '{0}':\n{1}".format(
+ pool_name, json.dumps(osd_map, indent=2)
+ ))
+ raise RuntimeError("Pool '{0}' not present in map after creation".format(pool_name))
+ else:
+ return pool_id
+
+ def create_group(self, group_id):
+ # Prevent craftily-named volume groups from colliding with the meta
+ # files.
+ if group_id.endswith(META_FILE_EXT):
+ raise ValueError("group ID cannot end with '{0}'.".format(
+ META_FILE_EXT))
+ path = self._get_group_path(group_id)
+ self._mkdir_p(path)
+
+ def destroy_group(self, group_id):
+ path = self._get_group_path(group_id)
+ try:
+ self.fs.stat(self.volume_prefix)
+ except cephfs.ObjectNotFound:
+ pass
+ else:
+ self.fs.rmdir(path)
+
+ def _mkdir_p(self, path):
+ try:
+ self.fs.stat(path)
+ except cephfs.ObjectNotFound:
+ pass
+ else:
+ return
+
+ parts = path.split(os.path.sep)
+
+ for i in range(1, len(parts) + 1):
+ subpath = os.path.join(*parts[0:i])
+ try:
+ self.fs.stat(subpath)
+ except cephfs.ObjectNotFound:
+ self.fs.mkdir(subpath, 0o755)
+
+ def create_volume(self, volume_path, size=None, data_isolated=False):
+ """
+ Set up metadata, pools and auth for a volume.
+
+ This function is idempotent. It is safe to call this again
+ for an already-created volume, even if it is in use.
+
+ :param volume_path: VolumePath instance
+ :param size: In bytes, or None for no size limit
+ :param data_isolated: If true, create a separate OSD pool for this volume
+ :return:
+ """
+ path = self._get_path(volume_path)
+ log.info("create_volume: {0}".format(path))
+
+ self._mkdir_p(path)
+
+ if size is not None:
+ self.fs.setxattr(path, 'ceph.quota.max_bytes', size.__str__(), 0)
+
+ # data_isolated means create a separate pool for this volume
+ if data_isolated:
+ pool_name = "{0}{1}".format(self.POOL_PREFIX, volume_path.volume_id)
+ log.info("create_volume: {0}, create pool {1} as data_isolated =True.".format(volume_path, pool_name))
+ pool_id = self._create_volume_pool(pool_name)
+ mds_map = self._rados_command("mds dump", {})
+ if pool_id not in mds_map['data_pools']:
+ self._rados_command("mds add_data_pool", {
+ 'pool': pool_name
+ })
+ self.fs.setxattr(path, 'ceph.dir.layout.pool', pool_name, 0)
+
+ # enforce security isolation, use seperate namespace for this volume
+ namespace = "{0}{1}".format(self.pool_ns_prefix, volume_path.volume_id)
+ log.info("create_volume: {0}, using rados namespace {1} to isolate data.".format(volume_path, namespace))
+ self.fs.setxattr(path, 'ceph.dir.layout.pool_namespace', namespace, 0)
+
+ # Create a volume meta file, if it does not already exist, to store
+ # data about auth ids having access to the volume
+ fd = self.fs.open(self._volume_metadata_path(volume_path),
+ os.O_CREAT, 0o755)
+ self.fs.close(fd)
+
+ return {
+ 'mount_path': path
+ }
+
+ def delete_volume(self, volume_path, data_isolated=False):
+ """
+ Make a volume inaccessible to guests. This function is
+ idempotent. This is the fast part of tearing down a volume: you must
+ also later call purge_volume, which is the slow part.
+
+ :param volume_path: Same identifier used in create_volume
+ :return:
+ """
+
+ path = self._get_path(volume_path)
+ log.info("delete_volume: {0}".format(path))
+
+ # Create the trash folder if it doesn't already exist
+ trash = os.path.join(self.volume_prefix, "_deleting")
+ self._mkdir_p(trash)
+
+ # We'll move it to here
+ trashed_volume = os.path.join(trash, volume_path.volume_id)
+
+ # Move the volume's data to the trash folder
+ try:
+ self.fs.stat(path)
+ except cephfs.ObjectNotFound:
+ log.warning("Trying to delete volume '{0}' but it's already gone".format(
+ path))
+ else:
+ self.fs.rename(path, trashed_volume)
+
+ # Delete the volume meta file, if it's not already deleted
+ vol_meta_path = self._volume_metadata_path(volume_path)
+ try:
+ self.fs.unlink(vol_meta_path)
+ except cephfs.ObjectNotFound:
+ pass
+
+ def purge_volume(self, volume_path, data_isolated=False):
+ """
+ Finish clearing up a volume that was previously passed to delete_volume. This
+ function is idempotent.
+ """
+
+ trash = os.path.join(self.volume_prefix, "_deleting")
+ trashed_volume = os.path.join(trash, volume_path.volume_id)
+
+ try:
+ self.fs.stat(trashed_volume)
+ except cephfs.ObjectNotFound:
+ log.warning("Trying to purge volume '{0}' but it's already been purged".format(
+ trashed_volume))
+ return
+
+ def rmtree(root_path):
+ log.debug("rmtree {0}".format(root_path))
+ dir_handle = self.fs.opendir(root_path)
+ d = self.fs.readdir(dir_handle)
+ while d:
+ if d.d_name not in [".", ".."]:
+ # Do not use os.path.join because it is sensitive
+ # to string encoding, we just pass through dnames
+ # as byte arrays
+ d_full = "{0}/{1}".format(root_path, d.d_name)
+ if d.is_dir():
+ rmtree(d_full)
+ else:
+ self.fs.unlink(d_full)
+
+ d = self.fs.readdir(dir_handle)
+ self.fs.closedir(dir_handle)
+
+ self.fs.rmdir(root_path)
+
+ rmtree(trashed_volume)
+
+ if data_isolated:
+ pool_name = "{0}{1}".format(self.POOL_PREFIX, volume_path.volume_id)
+ osd_map = self._rados_command("osd dump", {})
+ pool_id = self._get_pool_id(osd_map, pool_name)
+ mds_map = self._rados_command("mds dump", {})
+ if pool_id in mds_map['data_pools']:
+ self._rados_command("mds remove_data_pool", {
+ 'pool': pool_name
+ })
+ self._rados_command("osd pool delete",
+ {
+ "pool": pool_name,
+ "pool2": pool_name,
+ "sure": "--yes-i-really-really-mean-it"
+ })
+
+ def _get_ancestor_xattr(self, path, attr):
+ """
+ Helper for reading layout information: if this xattr is missing
+ on the requested path, keep checking parents until we find it.
+ """
+ try:
+ result = self.fs.getxattr(path, attr)
+ if result == "":
+ # Annoying! cephfs gives us empty instead of an error when attr not found
+ raise cephfs.NoData()
+ else:
+ return result
+ except cephfs.NoData:
+ if path == "/":
+ raise
+ else:
+ return self._get_ancestor_xattr(os.path.split(path)[0], attr)
+
+ def _check_compat_version(self, compat_version):
+ if self.version < compat_version:
+ msg = ("The current version of CephFSVolumeClient, version {0} "
+ "does not support the required feature. Need version {1} "
+ "or greater".format(self.version, compat_version)
+ )
+ log.error(msg)
+ raise CephFSVolumeClientError(msg)
+
+ def _metadata_get(self, path):
+ """
+ Return a deserialized JSON object, or None
+ """
+ fd = self.fs.open(path, "r")
+ # TODO iterate instead of assuming file < 4MB
+ read_bytes = self.fs.read(fd, 0, 4096 * 1024)
+ self.fs.close(fd)
+ if read_bytes:
+ return json.loads(read_bytes)
+ else:
+ return None
+
+ def _metadata_set(self, path, data):
+ serialized = json.dumps(data)
+ fd = self.fs.open(path, "w")
+ try:
+ self.fs.write(fd, serialized, 0)
+ self.fs.fsync(fd, 0)
+ finally:
+ self.fs.close(fd)
+
+ def _lock(self, path):
+ @contextmanager
+ def fn():
+ while(1):
+ fd = self.fs.open(path, os.O_CREAT, 0o755)
+ self.fs.flock(fd, fcntl.LOCK_EX, self._id)
+
+ # The locked file will be cleaned up sometime. It could be
+ # unlinked e.g., by an another manila share instance, before
+ # lock was applied on it. Perform checks to ensure that this
+ # does not happen.
+ try:
+ statbuf = self.fs.stat(path)
+ except cephfs.ObjectNotFound:
+ self.fs.close(fd)
+ continue
+
+ fstatbuf = self.fs.fstat(fd)
+ if statbuf.st_ino == fstatbuf.st_ino:
+ break
+
+ try:
+ yield
+ finally:
+ self.fs.flock(fd, fcntl.LOCK_UN, self._id)
+ self.fs.close(fd)
+
+ return fn()
+
+ def _auth_metadata_path(self, auth_id):
+ return os.path.join(self.volume_prefix, "${0}{1}".format(
+ auth_id, META_FILE_EXT))
+
+ def _auth_lock(self, auth_id):
+ return self._lock(self._auth_metadata_path(auth_id))
+
+ def _auth_metadata_get(self, auth_id):
+ """
+ Call me with the metadata locked!
+
+ Check whether a auth metadata structure can be decoded by the current
+ version of CephFSVolumeClient.
+
+ Return auth metadata that the current version of CephFSVolumeClient
+ can decode.
+ """
+ auth_metadata = self._metadata_get(self._auth_metadata_path(auth_id))
+
+ if auth_metadata:
+ self._check_compat_version(auth_metadata['compat_version'])
+
+ return auth_metadata
+
+ def _auth_metadata_set(self, auth_id, data):
+ """
+ Call me with the metadata locked!
+
+ Fsync the auth metadata.
+
+ Add two version attributes to the auth metadata,
+ 'compat_version', the minimum CephFSVolumeClient version that can
+ decode the metadata, and 'version', the CephFSVolumeClient version
+ that encoded the metadata.
+ """
+ data['compat_version'] = 1
+ data['version'] = self.version
+ return self._metadata_set(self._auth_metadata_path(auth_id), data)
+
+ def _volume_metadata_path(self, volume_path):
+ return os.path.join(self.volume_prefix, "_{0}:{1}{2}".format(
+ volume_path.group_id if volume_path.group_id else "",
+ volume_path.volume_id,
+ META_FILE_EXT
+ ))
+
+ def _volume_lock(self, volume_path):
+ """
+ Return a ContextManager which locks the authorization metadata for
+ a particular volume, and persists a flag to the metadata indicating
+ that it is currently locked, so that we can detect dirty situations
+ during recovery.
+
+ This lock isn't just to make access to the metadata safe: it's also
+ designed to be used over the two-step process of checking the
+ metadata and then responding to an authorization request, to
+ ensure that at the point we respond the metadata hasn't changed
+ in the background. It's key to how we avoid security holes
+ resulting from races during that problem ,
+ """
+ return self._lock(self._volume_metadata_path(volume_path))
+
+ def _volume_metadata_get(self, volume_path):
+ """
+ Call me with the metadata locked!
+
+ Check whether a volume metadata structure can be decoded by the current
+ version of CephFSVolumeClient.
+
+ Return a volume_metadata structure that the current version of
+ CephFSVolumeClient can decode.
+ """
+ volume_metadata = self._metadata_get(self._volume_metadata_path(volume_path))
+
+ if volume_metadata:
+ self._check_compat_version(volume_metadata['compat_version'])
+
+ return volume_metadata
+
+ def _volume_metadata_set(self, volume_path, data):
+ """
+ Call me with the metadata locked!
+
+ Add two version attributes to the volume metadata,
+ 'compat_version', the minimum CephFSVolumeClient version that can
+ decode the metadata and 'version', the CephFSVolumeClient version
+ that encoded the metadata.
+ """
+ data['compat_version'] = 1
+ data['version'] = self.version
+ return self._metadata_set(self._volume_metadata_path(volume_path), data)
+
+ def authorize(self, volume_path, auth_id, readonly=False, tenant_id=None):
+ """
+ Get-or-create a Ceph auth identity for `auth_id` and grant them access
+ to
+ :param volume_path:
+ :param auth_id:
+ :param readonly:
+ :param tenant_id: Optionally provide a stringizable object to
+ restrict any created cephx IDs to other callers
+ passing the same tenant ID.
+ :return:
+ """
+
+ with self._auth_lock(auth_id):
+ # Existing meta, or None, to be updated
+ auth_meta = self._auth_metadata_get(auth_id)
+
+ # volume data to be inserted
+ volume_path_str = str(volume_path)
+ volume = {
+ volume_path_str : {
+ # The access level at which the auth_id is authorized to
+ # access the volume.
+ 'access_level': 'r' if readonly else 'rw',
+ 'dirty': True,
+ }
+ }
+ if auth_meta is None:
+ sys.stderr.write("Creating meta for ID {0} with tenant {1}\n".format(
+ auth_id, tenant_id
+ ))
+ log.debug("Authorize: no existing meta")
+ auth_meta = {
+ 'dirty': True,
+ 'tenant_id': tenant_id.__str__() if tenant_id else None,
+ 'volumes': volume
+ }
+
+ # Note: this is *not* guaranteeing that the key doesn't already
+ # exist in Ceph: we are allowing VolumeClient tenants to
+ # 'claim' existing Ceph keys. In order to prevent VolumeClient
+ # tenants from reading e.g. client.admin keys, you need to
+ # have configured your VolumeClient user (e.g. Manila) to
+ # have mon auth caps that prevent it from accessing those keys
+ # (e.g. limit it to only access keys with a manila.* prefix)
+ else:
+ # Disallow tenants to share auth IDs
+ if auth_meta['tenant_id'].__str__() != tenant_id.__str__():
+ msg = "auth ID: {0} is already in use".format(auth_id)
+ log.error(msg)
+ raise CephFSVolumeClientError(msg)
+
+ if auth_meta['dirty']:
+ self._recover_auth_meta(auth_id, auth_meta)
+
+ log.debug("Authorize: existing tenant {tenant}".format(
+ tenant=auth_meta['tenant_id']
+ ))
+ auth_meta['dirty'] = True
+ auth_meta['volumes'].update(volume)
+
+ self._auth_metadata_set(auth_id, auth_meta)
+
+ with self._volume_lock(volume_path):
+ key = self._authorize_volume(volume_path, auth_id, readonly)
+
+ auth_meta['dirty'] = False
+ auth_meta['volumes'][volume_path_str]['dirty'] = False
+ self._auth_metadata_set(auth_id, auth_meta)
+
+ if tenant_id:
+ return {
+ 'auth_key': key
+ }
+ else:
+ # Caller wasn't multi-tenant aware: be safe and don't give
+ # them a key
+ return {
+ 'auth_key': None
+ }
+
+ def _authorize_volume(self, volume_path, auth_id, readonly):
+ vol_meta = self._volume_metadata_get(volume_path)
+
+ access_level = 'r' if readonly else 'rw'
+ auth = {
+ auth_id: {
+ 'access_level': access_level,
+ 'dirty': True,
+ }
+ }
+
+ if vol_meta is None:
+ vol_meta = {
+ 'auths': auth
+ }
+ else:
+ vol_meta['auths'].update(auth)
+ self._volume_metadata_set(volume_path, vol_meta)
+
+ key = self._authorize_ceph(volume_path, auth_id, readonly)
+
+ vol_meta['auths'][auth_id]['dirty'] = False
+ self._volume_metadata_set(volume_path, vol_meta)
+
+ return key
+
+ def _authorize_ceph(self, volume_path, auth_id, readonly):
+ path = self._get_path(volume_path)
+ log.debug("Authorizing Ceph id '{0}' for path '{1}'".format(
+ auth_id, path
+ ))
+
+ # First I need to work out what the data pool is for this share:
+ # read the layout
+ pool_name = self._get_ancestor_xattr(path, "ceph.dir.layout.pool")
+ namespace = self.fs.getxattr(path, "ceph.dir.layout.pool_namespace")
+
+ # Now construct auth capabilities that give the guest just enough
+ # permissions to access the share
+ client_entity = "client.{0}".format(auth_id)
+ want_access_level = 'r' if readonly else 'rw'
+ want_mds_cap = 'allow {0} path={1}'.format(want_access_level, path)
+ want_osd_cap = 'allow {0} pool={1} namespace={2}'.format(
+ want_access_level, pool_name, namespace)
+
+ try:
+ existing = self._rados_command(
+ 'auth get',
+ {
+ 'entity': client_entity
+ }
+ )
+ # FIXME: rados raising Error instead of ObjectNotFound in auth get failure
+ except rados.Error:
+ caps = self._rados_command(
+ 'auth get-or-create',
+ {
+ 'entity': client_entity,
+ 'caps': [
+ 'mds', want_mds_cap,
+ 'osd', want_osd_cap,
+ 'mon', 'allow r']
+ })
+ else:
+ # entity exists, update it
+ cap = existing[0]
+
+ # Construct auth caps that if present might conflict with the desired
+ # auth caps.
+ unwanted_access_level = 'r' if want_access_level is 'rw' else 'rw'
+ unwanted_mds_cap = 'allow {0} path={1}'.format(unwanted_access_level, path)
+ unwanted_osd_cap = 'allow {0} pool={1} namespace={2}'.format(
+ unwanted_access_level, pool_name, namespace)
+
+ def cap_update(orig, want, unwanted):
+ # Updates the existing auth caps such that there is a single
+ # occurrence of wanted auth caps and no occurrence of
+ # conflicting auth caps.
+
+ if not orig:
+ return want
+
+ cap_tokens = set(orig.split(","))
+
+ cap_tokens.discard(unwanted)
+ cap_tokens.add(want)
+
+ return ",".join(cap_tokens)
+
+ osd_cap_str = cap_update(cap['caps'].get('osd', ""), want_osd_cap, unwanted_osd_cap)
+ mds_cap_str = cap_update(cap['caps'].get('mds', ""), want_mds_cap, unwanted_mds_cap)
+
+ caps = self._rados_command(
+ 'auth caps',
+ {
+ 'entity': client_entity,
+ 'caps': [
+ 'mds', mds_cap_str,
+ 'osd', osd_cap_str,
+ 'mon', cap['caps'].get('mon', 'allow r')]
+ })
+ caps = self._rados_command(
+ 'auth get',
+ {
+ 'entity': client_entity
+ }
+ )
+
+ # Result expected like this:
+ # [
+ # {
+ # "entity": "client.foobar",
+ # "key": "AQBY0\/pViX\/wBBAAUpPs9swy7rey1qPhzmDVGQ==",
+ # "caps": {
+ # "mds": "allow *",
+ # "mon": "allow *"
+ # }
+ # }
+ # ]
+ assert len(caps) == 1
+ assert caps[0]['entity'] == client_entity
+ return caps[0]['key']
+
+ def deauthorize(self, volume_path, auth_id):
+ with self._auth_lock(auth_id):
+ # Existing meta, or None, to be updated
+ auth_meta = self._auth_metadata_get(auth_id)
+
+ volume_path_str = str(volume_path)
+ if (auth_meta is None) or (not auth_meta['volumes']):
+ log.warn("deauthorized called for already-removed auth"
+ "ID '{auth_id}' for volume ID '{volume}'".format(
+ auth_id=auth_id, volume=volume_path.volume_id
+ ))
+ # Clean up the auth meta file of an auth ID
+ self.fs.unlink(self._auth_metadata_path(auth_id))
+ return
+
+ if volume_path_str not in auth_meta['volumes']:
+ log.warn("deauthorized called for already-removed auth"
+ "ID '{auth_id}' for volume ID '{volume}'".format(
+ auth_id=auth_id, volume=volume_path.volume_id
+ ))
+ return
+
+ if auth_meta['dirty']:
+ self._recover_auth_meta(auth_id, auth_meta)
+
+ auth_meta['dirty'] = True
+ auth_meta['volumes'][volume_path_str]['dirty'] = True
+ self._auth_metadata_set(auth_id, auth_meta)
+
+ self._deauthorize_volume(volume_path, auth_id)
+
+ # Filter out the volume we're deauthorizing
+ del auth_meta['volumes'][volume_path_str]
+
+ # Clean up auth meta file
+ if not auth_meta['volumes']:
+ self.fs.unlink(self._auth_metadata_path(auth_id))
+ return
+
+ auth_meta['dirty'] = False
+ self._auth_metadata_set(auth_id, auth_meta)
+
+ def _deauthorize_volume(self, volume_path, auth_id):
+ with self._volume_lock(volume_path):
+ vol_meta = self._volume_metadata_get(volume_path)
+
+ if (vol_meta is None) or (auth_id not in vol_meta['auths']):
+ log.warn("deauthorized called for already-removed auth"
+ "ID '{auth_id}' for volume ID '{volume}'".format(
+ auth_id=auth_id, volume=volume_path.volume_id
+ ))
+ return
+
+ vol_meta['auths'][auth_id]['dirty'] = True
+ self._volume_metadata_set(volume_path, vol_meta)
+
+ self._deauthorize(volume_path, auth_id)
+
+ # Remove the auth_id from the metadata *after* removing it
+ # from ceph, so that if we crashed here, we would actually
+ # recreate the auth ID during recovery (i.e. end up with
+ # a consistent state).
+
+ # Filter out the auth we're removing
+ del vol_meta['auths'][auth_id]
+ self._volume_metadata_set(volume_path, vol_meta)
+
+ def _deauthorize(self, volume_path, auth_id):
+ """
+ The volume must still exist.
+ """
+ client_entity = "client.{0}".format(auth_id)
+ path = self._get_path(volume_path)
+ pool_name = self._get_ancestor_xattr(path, "ceph.dir.layout.pool")
+ namespace = self.fs.getxattr(path, "ceph.dir.layout.pool_namespace")
+
+ # The auth_id might have read-only or read-write mount access for the
+ # volume path.
+ access_levels = ('r', 'rw')
+ want_mds_caps = {'allow {0} path={1}'.format(access_level, path)
+ for access_level in access_levels}
+ want_osd_caps = {'allow {0} pool={1} namespace={2}'.format(
+ access_level, pool_name, namespace)
+ for access_level in access_levels}
+
+ try:
+ existing = self._rados_command(
+ 'auth get',
+ {
+ 'entity': client_entity
+ }
+ )
+
+ def cap_remove(orig, want):
+ cap_tokens = set(orig.split(","))
+ return ",".join(cap_tokens.difference(want))
+
+ cap = existing[0]
+ osd_cap_str = cap_remove(cap['caps'].get('osd', ""), want_osd_caps)
+ mds_cap_str = cap_remove(cap['caps'].get('mds', ""), want_mds_caps)
+ if (not osd_cap_str) and (not mds_cap_str):
+ self._rados_command('auth del', {'entity': client_entity}, decode=False)
+ else:
+ self._rados_command(
+ 'auth caps',
+ {
+ 'entity': client_entity,
+ 'caps': [
+ 'mds', mds_cap_str,
+ 'osd', osd_cap_str,
+ 'mon', cap['caps'].get('mon', 'allow r')]
+ })
+
+ # FIXME: rados raising Error instead of ObjectNotFound in auth get failure
+ except rados.Error:
+ # Already gone, great.
+ return
+
+ def get_authorized_ids(self, volume_path):
+ """
+ Expose a list of auth IDs that have access to a volume.
+
+ return: a list of (auth_id, access_level) tuples, where
+ the access_level can be 'r' , or 'rw'.
+ None if no auth ID is given access to the volume.
+ """
+ with self._volume_lock(volume_path):
+ meta = self._volume_metadata_get(volume_path)
+ auths = []
+ if not meta or not meta['auths']:
+ return None
+
+ for auth, auth_data in meta['auths'].items():
+ # Skip partial auth updates.
+ if not auth_data['dirty']:
+ auths.append((auth, auth_data['access_level']))
+
+ return auths
+
+ def _rados_command(self, prefix, args=None, decode=True):
+ """
+ Safer wrapper for ceph_argparse.json_command, which raises
+ Error exception instead of relying on caller to check return
+ codes.
+
+ Error exception can result from:
+ * Timeout
+ * Actual legitimate errors
+ * Malformed JSON output
+
+ return: Decoded object from ceph, or None if empty string returned.
+ If decode is False, return a string (the data returned by
+ ceph command)
+ """
+ if args is None:
+ args = {}
+
+ argdict = args.copy()
+ argdict['format'] = 'json'
+
+ ret, outbuf, outs = json_command(self.rados,
+ prefix=prefix,
+ argdict=argdict,
+ timeout=RADOS_TIMEOUT)
+ if ret != 0:
+ raise rados.Error(outs)
+ else:
+ if decode:
+ if outbuf:
+ try:
+ return json.loads(outbuf)
+ except (ValueError, TypeError):
+ raise RadosError("Invalid JSON output for command {0}".format(argdict))
+ else:
+ return None
+ else:
+ return outbuf
+
+ def get_used_bytes(self, volume_path):
+ return int(self.fs.getxattr(self._get_path(volume_path), "ceph.dir.rbytes"))
+
+ def set_max_bytes(self, volume_path, max_bytes):
+ self.fs.setxattr(self._get_path(volume_path), 'ceph.quota.max_bytes',
+ max_bytes.__str__() if max_bytes is not None else "0",
+ 0)
+
+ def _snapshot_path(self, dir_path, snapshot_name):
+ return os.path.join(
+ dir_path, self.rados.conf_get('client_snapdir'), snapshot_name
+ )
+
+ def _snapshot_create(self, dir_path, snapshot_name):
+ # TODO: raise intelligible exception for clusters where snaps are disabled
+ self.fs.mkdir(self._snapshot_path(dir_path, snapshot_name), 0o755)
+
+ def _snapshot_destroy(self, dir_path, snapshot_name):
+ """
+ Remove a snapshot, or do nothing if it already doesn't exist.
+ """
+ try:
+ self.fs.rmdir(self._snapshot_path(dir_path, snapshot_name))
+ except cephfs.ObjectNotFound:
+ log.warn("Snapshot was already gone: {0}".format(snapshot_name))
+
+ def create_snapshot_volume(self, volume_path, snapshot_name):
+ self._snapshot_create(self._get_path(volume_path), snapshot_name)
+
+ def destroy_snapshot_volume(self, volume_path, snapshot_name):
+ self._snapshot_destroy(self._get_path(volume_path), snapshot_name)
+
+ def create_snapshot_group(self, group_id, snapshot_name):
+ if group_id is None:
+ raise RuntimeError("Group ID may not be None")
+
+ return self._snapshot_create(self._get_group_path(group_id), snapshot_name)
+
+ def destroy_snapshot_group(self, group_id, snapshot_name):
+ if group_id is None:
+ raise RuntimeError("Group ID may not be None")
+ if snapshot_name is None:
+ raise RuntimeError("Snapshot name may not be None")
+
+ return self._snapshot_destroy(self._get_group_path(group_id), snapshot_name)
+
+ def _cp_r(self, src, dst):
+ # TODO
+ raise NotImplementedError()
+
+ def clone_volume_to_existing(self, dest_volume_path, src_volume_path, src_snapshot_name):
+ dest_fs_path = self._get_path(dest_volume_path)
+ src_snapshot_path = self._snapshot_path(self._get_path(src_volume_path), src_snapshot_name)
+
+ self._cp_r(src_snapshot_path, dest_fs_path)
+
+ def put_object(self, pool_name, object_name, data):
+ """
+ Synchronously write data to an object.
+
+ :param pool_name: name of the pool
+ :type pool_name: str
+ :param object_name: name of the object
+ :type object_name: str
+ :param data: data to write
+ :type data: bytes
+ """
+ ioctx = self.rados.open_ioctx(pool_name)
+ max_size = int(self.rados.conf_get('osd_max_write_size')) * 1024 * 1024
+ if len(data) > max_size:
+ msg = ("Data to be written to object '{0}' exceeds "
+ "{1} bytes".format(object_name, max_size))
+ log.error(msg)
+ raise CephFSVolumeClientError(msg)
+ try:
+ ioctx.write_full(object_name, data)
+ finally:
+ ioctx.close()
+
+ def get_object(self, pool_name, object_name):
+ """
+ Synchronously read data from object.
+
+ :param pool_name: name of the pool
+ :type pool_name: str
+ :param object_name: name of the object
+ :type object_name: str
+
+ :returns: bytes - data read from object
+ """
+ ioctx = self.rados.open_ioctx(pool_name)
+ max_size = int(self.rados.conf_get('osd_max_write_size')) * 1024 * 1024
+ try:
+ bytes_read = ioctx.read(object_name, max_size)
+ if ((len(bytes_read) == max_size) and
+ (ioctx.read(object_name, 1, offset=max_size))):
+ log.warning("Size of object {0} exceeds '{1}' bytes "
+ "read".format(object_name, max_size))
+ finally:
+ ioctx.close()
+ return bytes_read
+
+ def delete_object(self, pool_name, object_name):
+ ioctx = self.rados.open_ioctx(pool_name)
+ try:
+ ioctx.remove_object(object_name)
+ except rados.ObjectNotFound:
+ log.warn("Object '{0}' was already removed".format(object_name))
+ finally:
+ ioctx.close()
Code Review / stor4nfv.git / blobdiff