2 Copyright (C) 2015 Red Hat, Inc.
4 LGPL-2.1 or LGPL-3.0. See file COPYING.
7 from contextlib
import contextmanager
20 from ceph_argparse
import json_command
27 Helper method that returns byte representation of the given parameter.
29 if isinstance(param
, str):
30 return param
.encode('utf-8')
34 return str(param
).encode('utf-8')
36 class RadosError(Exception):
38 Something went wrong talking to Ceph with librados
45 log
= logging
.getLogger(__name__
)
47 # Reserved volume group name which we use in paths for volumes
48 # that are not assigned to a group (i.e. created with group=None)
49 NO_GROUP_NAME
= "_nogroup"
51 # Filename extensions for meta files.
52 META_FILE_EXT
= ".meta"
54 class VolumePath(object):
56 Identify a volume's path as group->volume
57 The Volume ID is a unique identifier, but this is a much more
58 helpful thing to pass around.
60 def __init__(self
, group_id
, volume_id
):
61 self
.group_id
= group_id
62 self
.volume_id
= volume_id
63 assert self
.group_id
!= NO_GROUP_NAME
64 assert self
.volume_id
!= "" and self
.volume_id
is not None
67 return "{0}/{1}".format(self
.group_id
, self
.volume_id
)
70 class ClusterTimeout(Exception):
72 Exception indicating that we timed out trying to talk to the Ceph cluster,
73 either to the mons, or to any individual daemon that the mons indicate ought
74 to be up but isn't responding to us.
79 class ClusterError(Exception):
81 Exception indicating that the cluster returned an error to a command that
82 we thought should be successful based on our last knowledge of the cluster
85 def __init__(self
, action
, result_code
, result_str
):
87 self
._result
_code
= result_code
88 self
._result
_str
= result_str
91 return "Error {0} (\"{1}\") while {2}".format(
92 self
._result
_code
, self
._result
_str
, self
._action
)
95 class RankEvicter(threading
.Thread
):
97 Thread for evicting client(s) from a particular MDS daemon instance.
99 This is more complex than simply sending a command, because we have to
100 handle cases where MDS daemons might not be fully up yet, and/or might
101 be transiently unresponsive to commands.
103 class GidGone(Exception):
108 def __init__(self
, volume_client
, client_spec
, rank
, gid
, mds_map
, ready_timeout
):
110 :param client_spec: list of strings, used as filter arguments to "session evict"
111 pass ["id=123"] to evict a single client with session id 123.
115 self
._mds
_map
= mds_map
116 self
._client
_spec
= client_spec
117 self
._volume
_client
= volume_client
118 self
._ready
_timeout
= ready_timeout
119 self
._ready
_waited
= 0
122 self
.exception
= None
124 super(RankEvicter
, self
).__init
__()
126 def _ready_to_evict(self
):
127 if self
._mds
_map
['up'].get("mds_{0}".format(self
.rank
), None) != self
.gid
:
128 log
.info("Evicting {0} from {1}/{2}: rank no longer associated with gid, done.".format(
129 self
._client
_spec
, self
.rank
, self
.gid
131 raise RankEvicter
.GidGone()
133 info
= self
._mds
_map
['info']["gid_{0}".format(self
.gid
)]
134 log
.debug("_ready_to_evict: state={0}".format(info
['state']))
135 return info
['state'] in ["up:active", "up:clientreplay"]
137 def _wait_for_ready(self
):
139 Wait for that MDS rank to reach an active or clientreplay state, and
142 while not self
._ready
_to
_evict
():
143 if self
._ready
_waited
> self
._ready
_timeout
:
144 raise ClusterTimeout()
146 time
.sleep(self
.POLL_PERIOD
)
147 self
._ready
_waited
+= self
.POLL_PERIOD
149 self
._mds
_map
= self
._volume
_client
.get_mds_map()
153 Run the eviction procedure. Return true on success, false on errors.
156 # Wait til the MDS is believed by the mon to be available for commands
158 self
._wait
_for
_ready
()
162 # Then send it an evict
163 ret
= errno
.ETIMEDOUT
164 while ret
== errno
.ETIMEDOUT
:
165 log
.debug("mds_command: {0}, {1}".format(
166 "%s" % self
.gid
, ["session", "evict"] + self
._client
_spec
168 ret
, outb
, outs
= self
._volume
_client
.fs
.mds_command(
171 "prefix": "session evict",
172 "filters": self
._client
_spec
174 log
.debug("mds_command: complete {0} {1}".format(ret
, outs
))
176 # If we get a clean response, great, it's gone from that rank.
179 elif ret
== errno
.ETIMEDOUT
:
180 # Oh no, the MDS went laggy (that's how libcephfs knows to emit this error)
181 self
._mds
_map
= self
._volume
_client
.get_mds_map()
183 self
._wait
_for
_ready
()
187 raise ClusterError("Sending evict to mds.{0}".format(self
.gid
), ret
, outs
)
192 except Exception as e
:
199 class EvictionError(Exception):
203 class CephFSVolumeClientError(Exception):
205 Something went wrong talking to Ceph using CephFSVolumeClient.
210 CEPHFSVOLUMECLIENT_VERSION_HISTORY
= """
212 CephFSVolumeClient Version History:
214 * 1 - Initial version
215 * 2 - Added get_object, put_object, delete_object methods to CephFSVolumeClient
216 * 3 - Allow volumes to be created without RADOS namespace isolation
217 * 4 - Added get_object_and_version, put_object_versioned method to CephFSVolumeClient
221 class CephFSVolumeClient(object):
223 Combine libcephfs and librados interfaces to implement a
224 'Volume' concept implemented as a cephfs directory and
225 client capabilities which restrict mount access to this
228 Additionally, volumes may be in a 'Group'. Conveniently,
229 volumes are a lot like manila shares, and groups are a lot
230 like manila consistency groups.
232 Refer to volumes with VolumePath, which specifies the
233 volume and group IDs (both strings). The group ID may
236 In general, functions in this class are allowed raise rados.Error
237 or cephfs.Error exceptions in unexpected situations.
243 # Where shall we create our volumes?
244 POOL_PREFIX
= "fsvolume_"
245 DEFAULT_VOL_PREFIX
= "/volumes"
246 DEFAULT_NS_PREFIX
= "fsvolumens_"
248 def __init__(self
, auth_id
=None, conf_path
=None, cluster_name
=None,
249 volume_prefix
=None, pool_ns_prefix
=None, rados
=None,
252 Either set all three of ``auth_id``, ``conf_path`` and
253 ``cluster_name`` (rados constructed on connect), or
254 set ``rados`` (existing rados instance).
257 self
.fs_name
= fs_name
258 self
.connected
= False
260 self
.conf_path
= conf_path
261 self
.cluster_name
= cluster_name
262 self
.auth_id
= auth_id
266 # Using an externally owned rados, so we won't tear it down
268 self
.own_rados
= False
270 # self.rados will be constructed in connect
271 self
.own_rados
= True
273 self
.volume_prefix
= volume_prefix
if volume_prefix
else self
.DEFAULT_VOL_PREFIX
274 self
.pool_ns_prefix
= pool_ns_prefix
if pool_ns_prefix
else self
.DEFAULT_NS_PREFIX
275 # For flock'ing in cephfs, I want a unique ID to distinguish me
276 # from any other manila-share services that are loading this module.
277 # We could use pid, but that's unnecessary weak: generate a
279 self
._id
= struct
.unpack(">Q", uuid
.uuid1().bytes
[0:8])[0]
281 # TODO: version the on-disk structures
284 # Scan all auth keys to see if they're dirty: if they are, they have
285 # state that might not have propagated to Ceph or to the related
288 # Important: we *always* acquire locks in the order auth->volume
289 # That means a volume can never be dirty without the auth key
290 # we're updating it with being dirty at the same time.
292 # First list the auth IDs that have potentially dirty on-disk metadata
293 log
.debug("Recovering from partial auth updates (if any)...")
296 dir_handle
= self
.fs
.opendir(self
.volume_prefix
)
297 except cephfs
.ObjectNotFound
:
298 log
.debug("Nothing to recover. No auth meta files.")
301 d
= self
.fs
.readdir(dir_handle
)
305 log
.debug("Nothing to recover. No auth meta files.")
308 # Identify auth IDs from auth meta filenames. The auth meta files
309 # are named as, "$<auth_id><meta filename extension>"
310 regex
= "^\$(.*){0}$".format(re
.escape(META_FILE_EXT
))
311 match
= re
.search(regex
, d
.d_name
.decode(encoding
='utf-8'))
313 auth_ids
.append(match
.group(1))
315 d
= self
.fs
.readdir(dir_handle
)
317 self
.fs
.closedir(dir_handle
)
319 # Key points based on ordering:
320 # * Anything added in VMeta is already added in AMeta
321 # * Anything added in Ceph is already added in VMeta
322 # * Anything removed in VMeta is already removed in Ceph
323 # * Anything removed in AMeta is already removed in VMeta
325 # Deauthorization: because I only update metadata AFTER the
326 # update of the next level down, I have the same ordering of
327 # -> things which exist in the AMeta should also exist
328 # in the VMeta, should also exist in Ceph, and the same
329 # recovery procedure that gets me consistent after crashes
330 # during authorization will also work during deauthorization
332 # Now for each auth ID, check for dirty flag and apply updates
333 # if dirty flag is found
334 for auth_id
in auth_ids
:
335 with self
._auth
_lock
(auth_id
):
336 auth_meta
= self
._auth
_metadata
_get
(auth_id
)
337 if not auth_meta
or not auth_meta
['volumes']:
338 # Clean up auth meta file
339 self
.fs
.unlink(self
._auth
_metadata
_path
(auth_id
))
341 if not auth_meta
['dirty']:
343 self
._recover
_auth
_meta
(auth_id
, auth_meta
)
345 log
.debug("Recovered from partial auth updates (if any).")
347 def _recover_auth_meta(self
, auth_id
, auth_meta
):
349 Call me after locking the auth meta file.
353 for volume
, volume_data
in auth_meta
['volumes'].items():
354 if not volume_data
['dirty']:
357 (group_id
, volume_id
) = volume
.split('/')
358 group_id
= group_id
if group_id
is not 'None' else None
359 volume_path
= VolumePath(group_id
, volume_id
)
360 access_level
= volume_data
['access_level']
362 with self
._volume
_lock
(volume_path
):
363 vol_meta
= self
._volume
_metadata
_get
(volume_path
)
365 # No VMeta update indicates that there was no auth update
366 # in Ceph either. So it's safe to remove corresponding
367 # partial update in AMeta.
368 if not vol_meta
or auth_id
not in vol_meta
['auths']:
369 remove_volumes
.append(volume
)
373 'access_level': access_level
,
376 # VMeta update looks clean. Ceph auth update must have been
378 if vol_meta
['auths'][auth_id
] == want_auth
:
381 readonly
= True if access_level
is 'r' else False
382 self
._authorize
_volume
(volume_path
, auth_id
, readonly
)
384 # Recovered from partial auth updates for the auth ID's access
386 auth_meta
['volumes'][volume
]['dirty'] = False
387 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
389 for volume
in remove_volumes
:
390 del auth_meta
['volumes'][volume
]
392 if not auth_meta
['volumes']:
393 # Clean up auth meta file
394 self
.fs
.unlink(self
._auth
_metadata
_path
(auth_id
))
397 # Recovered from all partial auth updates for the auth ID.
398 auth_meta
['dirty'] = False
399 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
401 def get_mds_map(self
):
402 fs_map
= self
._rados
_command
("fs dump", {})
403 return fs_map
['filesystems'][0]['mdsmap']
405 def evict(self
, auth_id
, timeout
=30, volume_path
=None):
407 Evict all clients based on the authorization ID and optionally based on
408 the volume path mounted. Assumes that the authorization key has been
409 revoked prior to calling this function.
411 This operation can throw an exception if the mon cluster is unresponsive, or
412 any individual MDS daemon is unresponsive for longer than the timeout passed in.
415 client_spec
= ["auth_name={0}".format(auth_id
), ]
417 client_spec
.append("client_metadata.root={0}".
418 format(self
._get
_path
(volume_path
)))
420 log
.info("evict clients with {0}".format(', '.join(client_spec
)))
422 mds_map
= self
.get_mds_map()
424 for name
, gid
in mds_map
['up'].items():
425 # Quirk of the MDSMap JSON dump: keys in the up dict are like "mds_0"
426 assert name
.startswith("mds_")
427 up
[int(name
[4:])] = gid
429 # For all MDS ranks held by a daemon
430 # Do the parallelism in python instead of using "tell mds.*", because
431 # the latter doesn't give us per-mds output
433 for rank
, gid
in up
.items():
434 thread
= RankEvicter(self
, client_spec
, rank
, gid
, mds_map
,
437 threads
.append(thread
)
442 log
.info("evict: joined all")
446 msg
= ("Failed to evict client with {0} from mds {1}/{2}: {3}".
447 format(', '.join(client_spec
), t
.rank
, t
.gid
, t
.exception
)
450 raise EvictionError(msg
)
452 def _get_path(self
, volume_path
):
454 Determine the path within CephFS where this volume will live
455 :return: absolute path (string)
459 volume_path
.group_id
if volume_path
.group_id
is not None else NO_GROUP_NAME
,
460 volume_path
.volume_id
)
462 def _get_group_path(self
, group_id
):
464 raise ValueError("group_id may not be None")
471 def _connect(self
, premount_evict
):
472 log
.debug("Connecting to cephfs...")
473 self
.fs
= cephfs
.LibCephFS(rados_inst
=self
.rados
)
474 log
.debug("CephFS initializing...")
476 if premount_evict
is not None:
477 log
.debug("Premount eviction of {0} starting".format(premount_evict
))
478 self
.evict(premount_evict
)
479 log
.debug("Premount eviction of {0} completes".format(premount_evict
))
480 log
.debug("CephFS mounting...")
481 self
.fs
.mount(filesystem_name
=to_bytes(self
.fs_name
))
482 log
.debug("Connection to cephfs complete")
484 # Recover from partial auth updates due to a previous
488 def connect(self
, premount_evict
= None):
491 :param premount_evict: Optional auth_id to evict before mounting the filesystem: callers
492 may want to use this to specify their own auth ID if they expect
493 to be a unique instance and don't want to wait for caps to time
494 out after failure of another instance of themselves.
497 log
.debug("Configuring to RADOS with config {0}...".format(self
.conf_path
))
498 self
.rados
= rados
.Rados(
499 name
="client.{0}".format(self
.auth_id
),
500 clustername
=self
.cluster_name
,
501 conffile
=self
.conf_path
,
504 if self
.rados
.state
!= "connected":
505 log
.debug("Connecting to RADOS...")
507 log
.debug("Connection to RADOS complete")
508 self
._connect
(premount_evict
)
510 def get_mon_addrs(self
):
511 log
.info("get_mon_addrs")
513 mon_map
= self
._rados
_command
("mon dump")
514 for mon
in mon_map
['mons']:
515 ip_port
= mon
['addr'].split("/")[0]
516 result
.append(ip_port
)
520 def disconnect(self
):
521 log
.info("disconnect")
523 log
.debug("Disconnecting cephfs...")
526 log
.debug("Disconnecting cephfs complete")
528 if self
.rados
and self
.own_rados
:
529 log
.debug("Disconnecting rados...")
530 self
.rados
.shutdown()
532 log
.debug("Disconnecting rados complete")
538 def __exit__(self
, exc_type
, exc_val
, exc_tb
):
544 def _get_pool_id(self
, osd_map
, pool_name
):
545 # Maybe borrow the OSDMap wrapper class from calamari if more helpers
546 # like this are needed.
547 for pool
in osd_map
['pools']:
548 if pool
['pool_name'] == pool_name
:
553 def _create_volume_pool(self
, pool_name
):
555 Idempotently create a pool for use as a CephFS data pool, with the given name
557 :return The ID of the created pool
559 osd_map
= self
._rados
_command
('osd dump', {})
561 existing_id
= self
._get
_pool
_id
(osd_map
, pool_name
)
562 if existing_id
is not None:
563 log
.info("Pool {0} already exists".format(pool_name
))
573 osd_map
= self
._rados
_command
('osd dump', {})
574 pool_id
= self
._get
_pool
_id
(osd_map
, pool_name
)
577 # If the pool isn't there, that's either a ceph bug, or it's some outside influence
578 # removing it right after we created it.
579 log
.error("OSD map doesn't contain expected pool '{0}':\n{1}".format(
580 pool_name
, json
.dumps(osd_map
, indent
=2)
582 raise RuntimeError("Pool '{0}' not present in map after creation".format(pool_name
))
586 def create_group(self
, group_id
, mode
=0o755):
587 # Prevent craftily-named volume groups from colliding with the meta
589 if group_id
.endswith(META_FILE_EXT
):
590 raise ValueError("group ID cannot end with '{0}'.".format(
592 path
= self
._get
_group
_path
(group_id
)
593 self
._mkdir
_p
(path
, mode
)
595 def destroy_group(self
, group_id
):
596 path
= self
._get
_group
_path
(group_id
)
598 self
.fs
.stat(self
.volume_prefix
)
599 except cephfs
.ObjectNotFound
:
604 def _mkdir_p(self
, path
, mode
=0o755):
607 except cephfs
.ObjectNotFound
:
612 parts
= path
.split(os
.path
.sep
)
614 for i
in range(1, len(parts
) + 1):
615 subpath
= os
.path
.join(*parts
[0:i
])
617 self
.fs
.stat(subpath
)
618 except cephfs
.ObjectNotFound
:
619 self
.fs
.mkdir(subpath
, mode
)
621 def create_volume(self
, volume_path
, size
=None, data_isolated
=False, namespace_isolated
=True,
624 Set up metadata, pools and auth for a volume.
626 This function is idempotent. It is safe to call this again
627 for an already-created volume, even if it is in use.
629 :param volume_path: VolumePath instance
630 :param size: In bytes, or None for no size limit
631 :param data_isolated: If true, create a separate OSD pool for this volume
632 :param namespace_isolated: If true, use separate RADOS namespace for this volume
635 path
= self
._get
_path
(volume_path
)
636 log
.info("create_volume: {0}".format(path
))
638 self
._mkdir
_p
(path
, mode
)
641 self
.fs
.setxattr(path
, 'ceph.quota.max_bytes', to_bytes(size
), 0)
643 # data_isolated means create a separate pool for this volume
645 pool_name
= "{0}{1}".format(self
.POOL_PREFIX
, volume_path
.volume_id
)
646 log
.info("create_volume: {0}, create pool {1} as data_isolated =True.".format(volume_path
, pool_name
))
647 pool_id
= self
._create
_volume
_pool
(pool_name
)
648 mds_map
= self
.get_mds_map()
649 if pool_id
not in mds_map
['data_pools']:
650 self
._rados
_command
("fs add_data_pool", {
651 'fs_name': mds_map
['fs_name'],
654 time
.sleep(5) # time for MDSMap to be distributed
655 self
.fs
.setxattr(path
, 'ceph.dir.layout.pool', to_bytes(pool_name
), 0)
657 # enforce security isolation, use separate namespace for this volume
658 if namespace_isolated
:
659 namespace
= "{0}{1}".format(self
.pool_ns_prefix
, volume_path
.volume_id
)
660 log
.info("create_volume: {0}, using rados namespace {1} to isolate data.".format(volume_path
, namespace
))
661 self
.fs
.setxattr(path
, 'ceph.dir.layout.pool_namespace',
662 to_bytes(namespace
), 0)
664 # If volume's namespace layout is not set, then the volume's pool
665 # layout remains unset and will undesirably change with ancestor's
666 # pool layout changes.
667 pool_name
= self
._get
_ancestor
_xattr
(path
, "ceph.dir.layout.pool")
668 self
.fs
.setxattr(path
, 'ceph.dir.layout.pool',
669 to_bytes(pool_name
), 0)
671 # Create a volume meta file, if it does not already exist, to store
672 # data about auth ids having access to the volume
673 fd
= self
.fs
.open(self
._volume
_metadata
_path
(volume_path
),
681 def delete_volume(self
, volume_path
, data_isolated
=False):
683 Make a volume inaccessible to guests. This function is
684 idempotent. This is the fast part of tearing down a volume: you must
685 also later call purge_volume, which is the slow part.
687 :param volume_path: Same identifier used in create_volume
691 path
= self
._get
_path
(volume_path
)
692 log
.info("delete_volume: {0}".format(path
))
694 # Create the trash folder if it doesn't already exist
695 trash
= os
.path
.join(self
.volume_prefix
, "_deleting")
698 # We'll move it to here
699 trashed_volume
= os
.path
.join(trash
, volume_path
.volume_id
)
701 # Move the volume's data to the trash folder
704 except cephfs
.ObjectNotFound
:
705 log
.warning("Trying to delete volume '{0}' but it's already gone".format(
708 self
.fs
.rename(path
, trashed_volume
)
710 # Delete the volume meta file, if it's not already deleted
711 vol_meta_path
= self
._volume
_metadata
_path
(volume_path
)
713 self
.fs
.unlink(vol_meta_path
)
714 except cephfs
.ObjectNotFound
:
717 def purge_volume(self
, volume_path
, data_isolated
=False):
719 Finish clearing up a volume that was previously passed to delete_volume. This
720 function is idempotent.
723 trash
= os
.path
.join(self
.volume_prefix
, "_deleting")
724 trashed_volume
= os
.path
.join(trash
, volume_path
.volume_id
)
727 self
.fs
.stat(trashed_volume
)
728 except cephfs
.ObjectNotFound
:
729 log
.warning("Trying to purge volume '{0}' but it's already been purged".format(
733 def rmtree(root_path
):
734 log
.debug("rmtree {0}".format(root_path
))
735 dir_handle
= self
.fs
.opendir(root_path
)
736 d
= self
.fs
.readdir(dir_handle
)
738 d_name
= d
.d_name
.decode(encoding
='utf-8')
739 if d_name
not in [".", ".."]:
740 # Do not use os.path.join because it is sensitive
741 # to string encoding, we just pass through dnames
743 d_full
= u
"{0}/{1}".format(root_path
, d_name
)
747 self
.fs
.unlink(d_full
)
749 d
= self
.fs
.readdir(dir_handle
)
750 self
.fs
.closedir(dir_handle
)
752 self
.fs
.rmdir(root_path
)
754 rmtree(trashed_volume
)
757 pool_name
= "{0}{1}".format(self
.POOL_PREFIX
, volume_path
.volume_id
)
758 osd_map
= self
._rados
_command
("osd dump", {})
759 pool_id
= self
._get
_pool
_id
(osd_map
, pool_name
)
760 mds_map
= self
.get_mds_map()
761 if pool_id
in mds_map
['data_pools']:
762 self
._rados
_command
("fs rm_data_pool", {
763 'fs_name': mds_map
['fs_name'],
766 self
._rados
_command
("osd pool delete",
770 "yes_i_really_really_mean_it": True
773 def _get_ancestor_xattr(self
, path
, attr
):
775 Helper for reading layout information: if this xattr is missing
776 on the requested path, keep checking parents until we find it.
779 result
= self
.fs
.getxattr(path
, attr
).decode()
781 # Annoying! cephfs gives us empty instead of an error when attr not found
782 raise cephfs
.NoData()
785 except cephfs
.NoData
:
789 return self
._get
_ancestor
_xattr
(os
.path
.split(path
)[0], attr
)
791 def _check_compat_version(self
, compat_version
):
792 if self
.version
< compat_version
:
793 msg
= ("The current version of CephFSVolumeClient, version {0} "
794 "does not support the required feature. Need version {1} "
795 "or greater".format(self
.version
, compat_version
)
798 raise CephFSVolumeClientError(msg
)
800 def _metadata_get(self
, path
):
802 Return a deserialized JSON object, or None
804 fd
= self
.fs
.open(path
, "r")
805 # TODO iterate instead of assuming file < 4MB
806 read_bytes
= self
.fs
.read(fd
, 0, 4096 * 1024)
809 return json
.loads(read_bytes
.decode())
813 def _metadata_set(self
, path
, data
):
814 serialized
= json
.dumps(data
)
815 fd
= self
.fs
.open(path
, "w")
817 self
.fs
.write(fd
, to_bytes(serialized
), 0)
822 def _lock(self
, path
):
826 fd
= self
.fs
.open(path
, os
.O_CREAT
, 0o755)
827 self
.fs
.flock(fd
, fcntl
.LOCK_EX
, self
._id
)
829 # The locked file will be cleaned up sometime. It could be
830 # unlinked e.g., by an another manila share instance, before
831 # lock was applied on it. Perform checks to ensure that this
834 statbuf
= self
.fs
.stat(path
)
835 except cephfs
.ObjectNotFound
:
839 fstatbuf
= self
.fs
.fstat(fd
)
840 if statbuf
.st_ino
== fstatbuf
.st_ino
:
846 self
.fs
.flock(fd
, fcntl
.LOCK_UN
, self
._id
)
851 def _auth_metadata_path(self
, auth_id
):
852 return os
.path
.join(self
.volume_prefix
, "${0}{1}".format(
853 auth_id
, META_FILE_EXT
))
855 def _auth_lock(self
, auth_id
):
856 return self
._lock
(self
._auth
_metadata
_path
(auth_id
))
858 def _auth_metadata_get(self
, auth_id
):
860 Call me with the metadata locked!
862 Check whether a auth metadata structure can be decoded by the current
863 version of CephFSVolumeClient.
865 Return auth metadata that the current version of CephFSVolumeClient
868 auth_metadata
= self
._metadata
_get
(self
._auth
_metadata
_path
(auth_id
))
871 self
._check
_compat
_version
(auth_metadata
['compat_version'])
875 def _auth_metadata_set(self
, auth_id
, data
):
877 Call me with the metadata locked!
879 Fsync the auth metadata.
881 Add two version attributes to the auth metadata,
882 'compat_version', the minimum CephFSVolumeClient version that can
883 decode the metadata, and 'version', the CephFSVolumeClient version
884 that encoded the metadata.
886 data
['compat_version'] = 1
887 data
['version'] = self
.version
888 return self
._metadata
_set
(self
._auth
_metadata
_path
(auth_id
), data
)
890 def _volume_metadata_path(self
, volume_path
):
891 return os
.path
.join(self
.volume_prefix
, "_{0}:{1}{2}".format(
892 volume_path
.group_id
if volume_path
.group_id
else "",
893 volume_path
.volume_id
,
897 def _volume_lock(self
, volume_path
):
899 Return a ContextManager which locks the authorization metadata for
900 a particular volume, and persists a flag to the metadata indicating
901 that it is currently locked, so that we can detect dirty situations
904 This lock isn't just to make access to the metadata safe: it's also
905 designed to be used over the two-step process of checking the
906 metadata and then responding to an authorization request, to
907 ensure that at the point we respond the metadata hasn't changed
908 in the background. It's key to how we avoid security holes
909 resulting from races during that problem ,
911 return self
._lock
(self
._volume
_metadata
_path
(volume_path
))
913 def _volume_metadata_get(self
, volume_path
):
915 Call me with the metadata locked!
917 Check whether a volume metadata structure can be decoded by the current
918 version of CephFSVolumeClient.
920 Return a volume_metadata structure that the current version of
921 CephFSVolumeClient can decode.
923 volume_metadata
= self
._metadata
_get
(self
._volume
_metadata
_path
(volume_path
))
926 self
._check
_compat
_version
(volume_metadata
['compat_version'])
928 return volume_metadata
930 def _volume_metadata_set(self
, volume_path
, data
):
932 Call me with the metadata locked!
934 Add two version attributes to the volume metadata,
935 'compat_version', the minimum CephFSVolumeClient version that can
936 decode the metadata and 'version', the CephFSVolumeClient version
937 that encoded the metadata.
939 data
['compat_version'] = 1
940 data
['version'] = self
.version
941 return self
._metadata
_set
(self
._volume
_metadata
_path
(volume_path
), data
)
943 def authorize(self
, volume_path
, auth_id
, readonly
=False, tenant_id
=None):
945 Get-or-create a Ceph auth identity for `auth_id` and grant them access
950 :param tenant_id: Optionally provide a stringizable object to
951 restrict any created cephx IDs to other callers
952 passing the same tenant ID.
956 with self
._auth
_lock
(auth_id
):
957 # Existing meta, or None, to be updated
958 auth_meta
= self
._auth
_metadata
_get
(auth_id
)
960 # volume data to be inserted
961 volume_path_str
= str(volume_path
)
964 # The access level at which the auth_id is authorized to
966 'access_level': 'r' if readonly
else 'rw',
970 if auth_meta
is None:
971 sys
.stderr
.write("Creating meta for ID {0} with tenant {1}\n".format(
974 log
.debug("Authorize: no existing meta")
977 'tenant_id': tenant_id
.__str
__() if tenant_id
else None,
981 # Note: this is *not* guaranteeing that the key doesn't already
982 # exist in Ceph: we are allowing VolumeClient tenants to
983 # 'claim' existing Ceph keys. In order to prevent VolumeClient
984 # tenants from reading e.g. client.admin keys, you need to
985 # have configured your VolumeClient user (e.g. Manila) to
986 # have mon auth caps that prevent it from accessing those keys
987 # (e.g. limit it to only access keys with a manila.* prefix)
989 # Disallow tenants to share auth IDs
990 if auth_meta
['tenant_id'].__str
__() != tenant_id
.__str
__():
991 msg
= "auth ID: {0} is already in use".format(auth_id
)
993 raise CephFSVolumeClientError(msg
)
995 if auth_meta
['dirty']:
996 self
._recover
_auth
_meta
(auth_id
, auth_meta
)
998 log
.debug("Authorize: existing tenant {tenant}".format(
999 tenant
=auth_meta
['tenant_id']
1001 auth_meta
['dirty'] = True
1002 auth_meta
['volumes'].update(volume
)
1004 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
1006 with self
._volume
_lock
(volume_path
):
1007 key
= self
._authorize
_volume
(volume_path
, auth_id
, readonly
)
1009 auth_meta
['dirty'] = False
1010 auth_meta
['volumes'][volume_path_str
]['dirty'] = False
1011 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
1018 # Caller wasn't multi-tenant aware: be safe and don't give
1024 def _authorize_volume(self
, volume_path
, auth_id
, readonly
):
1025 vol_meta
= self
._volume
_metadata
_get
(volume_path
)
1027 access_level
= 'r' if readonly
else 'rw'
1030 'access_level': access_level
,
1035 if vol_meta
is None:
1040 vol_meta
['auths'].update(auth
)
1041 self
._volume
_metadata
_set
(volume_path
, vol_meta
)
1043 key
= self
._authorize
_ceph
(volume_path
, auth_id
, readonly
)
1045 vol_meta
['auths'][auth_id
]['dirty'] = False
1046 self
._volume
_metadata
_set
(volume_path
, vol_meta
)
1050 def _authorize_ceph(self
, volume_path
, auth_id
, readonly
):
1051 path
= self
._get
_path
(volume_path
)
1052 log
.debug("Authorizing Ceph id '{0}' for path '{1}'".format(
1056 # First I need to work out what the data pool is for this share:
1058 pool_name
= self
._get
_ancestor
_xattr
(path
, "ceph.dir.layout.pool")
1061 namespace
= self
.fs
.getxattr(path
, "ceph.dir.layout.pool_"
1062 "namespace").decode()
1063 except cephfs
.NoData
:
1066 # Now construct auth capabilities that give the guest just enough
1067 # permissions to access the share
1068 client_entity
= "client.{0}".format(auth_id
)
1069 want_access_level
= 'r' if readonly
else 'rw'
1070 want_mds_cap
= 'allow {0} path={1}'.format(want_access_level
, path
)
1072 want_osd_cap
= 'allow {0} pool={1} namespace={2}'.format(
1073 want_access_level
, pool_name
, namespace
)
1075 want_osd_cap
= 'allow {0} pool={1}'.format(want_access_level
,
1079 existing
= self
._rados
_command
(
1082 'entity': client_entity
1085 # FIXME: rados raising Error instead of ObjectNotFound in auth get failure
1087 caps
= self
._rados
_command
(
1088 'auth get-or-create',
1090 'entity': client_entity
,
1092 'mds', want_mds_cap
,
1093 'osd', want_osd_cap
,
1097 # entity exists, update it
1100 # Construct auth caps that if present might conflict with the desired
1102 unwanted_access_level
= 'r' if want_access_level
is 'rw' else 'rw'
1103 unwanted_mds_cap
= 'allow {0} path={1}'.format(unwanted_access_level
, path
)
1105 unwanted_osd_cap
= 'allow {0} pool={1} namespace={2}'.format(
1106 unwanted_access_level
, pool_name
, namespace
)
1108 unwanted_osd_cap
= 'allow {0} pool={1}'.format(
1109 unwanted_access_level
, pool_name
)
1112 orig_mds_caps
, orig_osd_caps
, want_mds_cap
,
1113 want_osd_cap
, unwanted_mds_cap
, unwanted_osd_cap
):
1115 if not orig_mds_caps
:
1116 return want_mds_cap
, want_osd_cap
1118 mds_cap_tokens
= orig_mds_caps
.split(",")
1119 osd_cap_tokens
= orig_osd_caps
.split(",")
1121 if want_mds_cap
in mds_cap_tokens
:
1122 return orig_mds_caps
, orig_osd_caps
1124 if unwanted_mds_cap
in mds_cap_tokens
:
1125 mds_cap_tokens
.remove(unwanted_mds_cap
)
1126 osd_cap_tokens
.remove(unwanted_osd_cap
)
1128 mds_cap_tokens
.append(want_mds_cap
)
1129 osd_cap_tokens
.append(want_osd_cap
)
1131 return ",".join(mds_cap_tokens
), ",".join(osd_cap_tokens
)
1133 orig_mds_caps
= cap
['caps'].get('mds', "")
1134 orig_osd_caps
= cap
['caps'].get('osd', "")
1136 mds_cap_str
, osd_cap_str
= cap_update(
1137 orig_mds_caps
, orig_osd_caps
, want_mds_cap
, want_osd_cap
,
1138 unwanted_mds_cap
, unwanted_osd_cap
)
1140 caps
= self
._rados
_command
(
1143 'entity': client_entity
,
1147 'mon', cap
['caps'].get('mon', 'allow r')]
1149 caps
= self
._rados
_command
(
1152 'entity': client_entity
1156 # Result expected like this:
1159 # "entity": "client.foobar",
1160 # "key": "AQBY0\/pViX\/wBBAAUpPs9swy7rey1qPhzmDVGQ==",
1167 assert len(caps
) == 1
1168 assert caps
[0]['entity'] == client_entity
1169 return caps
[0]['key']
1171 def deauthorize(self
, volume_path
, auth_id
):
1172 with self
._auth
_lock
(auth_id
):
1173 # Existing meta, or None, to be updated
1174 auth_meta
= self
._auth
_metadata
_get
(auth_id
)
1176 volume_path_str
= str(volume_path
)
1177 if (auth_meta
is None) or (not auth_meta
['volumes']):
1178 log
.warn("deauthorized called for already-removed auth"
1179 "ID '{auth_id}' for volume ID '{volume}'".format(
1180 auth_id
=auth_id
, volume
=volume_path
.volume_id
1182 # Clean up the auth meta file of an auth ID
1183 self
.fs
.unlink(self
._auth
_metadata
_path
(auth_id
))
1186 if volume_path_str
not in auth_meta
['volumes']:
1187 log
.warn("deauthorized called for already-removed auth"
1188 "ID '{auth_id}' for volume ID '{volume}'".format(
1189 auth_id
=auth_id
, volume
=volume_path
.volume_id
1193 if auth_meta
['dirty']:
1194 self
._recover
_auth
_meta
(auth_id
, auth_meta
)
1196 auth_meta
['dirty'] = True
1197 auth_meta
['volumes'][volume_path_str
]['dirty'] = True
1198 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
1200 self
._deauthorize
_volume
(volume_path
, auth_id
)
1202 # Filter out the volume we're deauthorizing
1203 del auth_meta
['volumes'][volume_path_str
]
1205 # Clean up auth meta file
1206 if not auth_meta
['volumes']:
1207 self
.fs
.unlink(self
._auth
_metadata
_path
(auth_id
))
1210 auth_meta
['dirty'] = False
1211 self
._auth
_metadata
_set
(auth_id
, auth_meta
)
1213 def _deauthorize_volume(self
, volume_path
, auth_id
):
1214 with self
._volume
_lock
(volume_path
):
1215 vol_meta
= self
._volume
_metadata
_get
(volume_path
)
1217 if (vol_meta
is None) or (auth_id
not in vol_meta
['auths']):
1218 log
.warn("deauthorized called for already-removed auth"
1219 "ID '{auth_id}' for volume ID '{volume}'".format(
1220 auth_id
=auth_id
, volume
=volume_path
.volume_id
1224 vol_meta
['auths'][auth_id
]['dirty'] = True
1225 self
._volume
_metadata
_set
(volume_path
, vol_meta
)
1227 self
._deauthorize
(volume_path
, auth_id
)
1229 # Remove the auth_id from the metadata *after* removing it
1230 # from ceph, so that if we crashed here, we would actually
1231 # recreate the auth ID during recovery (i.e. end up with
1232 # a consistent state).
1234 # Filter out the auth we're removing
1235 del vol_meta
['auths'][auth_id
]
1236 self
._volume
_metadata
_set
(volume_path
, vol_meta
)
1238 def _deauthorize(self
, volume_path
, auth_id
):
1240 The volume must still exist.
1242 client_entity
= "client.{0}".format(auth_id
)
1243 path
= self
._get
_path
(volume_path
)
1244 pool_name
= self
._get
_ancestor
_xattr
(path
, "ceph.dir.layout.pool")
1246 namespace
= self
.fs
.getxattr(path
, "ceph.dir.layout.pool_"
1247 "namespace").decode()
1248 except cephfs
.NoData
:
1251 # The auth_id might have read-only or read-write mount access for the
1253 access_levels
= ('r', 'rw')
1254 want_mds_caps
= ['allow {0} path={1}'.format(access_level
, path
)
1255 for access_level
in access_levels
]
1257 want_osd_caps
= ['allow {0} pool={1} namespace={2}'.format(access_level
, pool_name
, namespace
)
1258 for access_level
in access_levels
]
1260 want_osd_caps
= ['allow {0} pool={1}'.format(access_level
, pool_name
)
1261 for access_level
in access_levels
]
1265 existing
= self
._rados
_command
(
1268 'entity': client_entity
1272 def cap_remove(orig_mds_caps
, orig_osd_caps
, want_mds_caps
, want_osd_caps
):
1273 mds_cap_tokens
= orig_mds_caps
.split(",")
1274 osd_cap_tokens
= orig_osd_caps
.split(",")
1276 for want_mds_cap
, want_osd_cap
in zip(want_mds_caps
, want_osd_caps
):
1277 if want_mds_cap
in mds_cap_tokens
:
1278 mds_cap_tokens
.remove(want_mds_cap
)
1279 osd_cap_tokens
.remove(want_osd_cap
)
1282 return ",".join(mds_cap_tokens
), ",".join(osd_cap_tokens
)
1285 orig_mds_caps
= cap
['caps'].get('mds', "")
1286 orig_osd_caps
= cap
['caps'].get('osd', "")
1287 mds_cap_str
, osd_cap_str
= cap_remove(orig_mds_caps
, orig_osd_caps
,
1288 want_mds_caps
, want_osd_caps
)
1291 self
._rados
_command
('auth del', {'entity': client_entity
}, decode
=False)
1293 self
._rados
_command
(
1296 'entity': client_entity
,
1300 'mon', cap
['caps'].get('mon', 'allow r')]
1303 # FIXME: rados raising Error instead of ObjectNotFound in auth get failure
1305 # Already gone, great.
1308 def get_authorized_ids(self
, volume_path
):
1310 Expose a list of auth IDs that have access to a volume.
1312 return: a list of (auth_id, access_level) tuples, where
1313 the access_level can be 'r' , or 'rw'.
1314 None if no auth ID is given access to the volume.
1316 with self
._volume
_lock
(volume_path
):
1317 meta
= self
._volume
_metadata
_get
(volume_path
)
1319 if not meta
or not meta
['auths']:
1322 for auth
, auth_data
in meta
['auths'].items():
1323 # Skip partial auth updates.
1324 if not auth_data
['dirty']:
1325 auths
.append((auth
, auth_data
['access_level']))
1329 def _rados_command(self
, prefix
, args
=None, decode
=True):
1331 Safer wrapper for ceph_argparse.json_command, which raises
1332 Error exception instead of relying on caller to check return
1335 Error exception can result from:
1337 * Actual legitimate errors
1338 * Malformed JSON output
1340 return: Decoded object from ceph, or None if empty string returned.
1341 If decode is False, return a string (the data returned by
1347 argdict
= args
.copy()
1348 argdict
['format'] = 'json'
1350 ret
, outbuf
, outs
= json_command(self
.rados
,
1353 timeout
=RADOS_TIMEOUT
)
1355 raise rados
.Error(outs
)
1360 return json
.loads(outbuf
.decode())
1361 except (ValueError, TypeError):
1362 raise RadosError("Invalid JSON output for command {0}".format(argdict
))
1368 def get_used_bytes(self
, volume_path
):
1369 return int(self
.fs
.getxattr(self
._get
_path
(volume_path
), "ceph.dir."
1372 def set_max_bytes(self
, volume_path
, max_bytes
):
1373 self
.fs
.setxattr(self
._get
_path
(volume_path
), 'ceph.quota.max_bytes',
1374 to_bytes(max_bytes
if max_bytes
else 0), 0)
1376 def _snapshot_path(self
, dir_path
, snapshot_name
):
1377 return os
.path
.join(
1378 dir_path
, self
.rados
.conf_get('client_snapdir'), snapshot_name
1381 def _snapshot_create(self
, dir_path
, snapshot_name
, mode
=0o755):
1382 # TODO: raise intelligible exception for clusters where snaps are disabled
1383 self
.fs
.mkdir(self
._snapshot
_path
(dir_path
, snapshot_name
), mode
)
1385 def _snapshot_destroy(self
, dir_path
, snapshot_name
):
1387 Remove a snapshot, or do nothing if it already doesn't exist.
1390 self
.fs
.rmdir(self
._snapshot
_path
(dir_path
, snapshot_name
))
1391 except cephfs
.ObjectNotFound
:
1392 log
.warn("Snapshot was already gone: {0}".format(snapshot_name
))
1394 def create_snapshot_volume(self
, volume_path
, snapshot_name
, mode
=0o755):
1395 self
._snapshot
_create
(self
._get
_path
(volume_path
), snapshot_name
, mode
)
1397 def destroy_snapshot_volume(self
, volume_path
, snapshot_name
):
1398 self
._snapshot
_destroy
(self
._get
_path
(volume_path
), snapshot_name
)
1400 def create_snapshot_group(self
, group_id
, snapshot_name
, mode
=0o755):
1401 if group_id
is None:
1402 raise RuntimeError("Group ID may not be None")
1404 return self
._snapshot
_create
(self
._get
_group
_path
(group_id
), snapshot_name
,
1407 def destroy_snapshot_group(self
, group_id
, snapshot_name
):
1408 if group_id
is None:
1409 raise RuntimeError("Group ID may not be None")
1410 if snapshot_name
is None:
1411 raise RuntimeError("Snapshot name may not be None")
1413 return self
._snapshot
_destroy
(self
._get
_group
_path
(group_id
), snapshot_name
)
1415 def _cp_r(self
, src
, dst
):
1417 raise NotImplementedError()
1419 def clone_volume_to_existing(self
, dest_volume_path
, src_volume_path
, src_snapshot_name
):
1420 dest_fs_path
= self
._get
_path
(dest_volume_path
)
1421 src_snapshot_path
= self
._snapshot
_path
(self
._get
_path
(src_volume_path
), src_snapshot_name
)
1423 self
._cp
_r
(src_snapshot_path
, dest_fs_path
)
1425 def put_object(self
, pool_name
, object_name
, data
):
1427 Synchronously write data to an object.
1429 :param pool_name: name of the pool
1430 :type pool_name: str
1431 :param object_name: name of the object
1432 :type object_name: str
1433 :param data: data to write
1436 return self
.put_object_versioned(pool_name
, object_name
, data
)
1438 def put_object_versioned(self
, pool_name
, object_name
, data
, version
=None):
1440 Synchronously write data to an object only if version of the object
1441 version matches the expected version.
1443 :param pool_name: name of the pool
1444 :type pool_name: str
1445 :param object_name: name of the object
1446 :type object_name: str
1447 :param data: data to write
1449 :param version: expected version of the object to write
1452 ioctx
= self
.rados
.open_ioctx(pool_name
)
1454 max_size
= int(self
.rados
.conf_get('osd_max_write_size')) * 1024 * 1024
1455 if len(data
) > max_size
:
1456 msg
= ("Data to be written to object '{0}' exceeds "
1457 "{1} bytes".format(object_name
, max_size
))
1459 raise CephFSVolumeClientError(msg
)
1462 with rados
.WriteOpCtx() as wop
:
1463 if version
is not None:
1464 wop
.assert_version(version
)
1465 wop
.write_full(data
)
1466 ioctx
.operate_write_op(wop
, object_name
)
1467 except rados
.OSError as e
:
1473 def get_object(self
, pool_name
, object_name
):
1475 Synchronously read data from object.
1477 :param pool_name: name of the pool
1478 :type pool_name: str
1479 :param object_name: name of the object
1480 :type object_name: str
1482 :returns: bytes - data read from object
1484 return self
.get_object_and_version(pool_name
, object_name
)[0]
1486 def get_object_and_version(self
, pool_name
, object_name
):
1488 Synchronously read data from object and get its version.
1490 :param pool_name: name of the pool
1491 :type pool_name: str
1492 :param object_name: name of the object
1493 :type object_name: str
1495 :returns: tuple of object data and version
1497 ioctx
= self
.rados
.open_ioctx(pool_name
)
1498 max_size
= int(self
.rados
.conf_get('osd_max_write_size')) * 1024 * 1024
1500 bytes_read
= ioctx
.read(object_name
, max_size
)
1501 if ((len(bytes_read
) == max_size
) and
1502 (ioctx
.read(object_name
, 1, offset
=max_size
))):
1503 log
.warning("Size of object {0} exceeds '{1}' bytes "
1504 "read".format(object_name
, max_size
))
1505 obj_version
= ioctx
.get_last_version()
1508 return (bytes_read
, obj_version
)
1510 def delete_object(self
, pool_name
, object_name
):
1511 ioctx
= self
.rados
.open_ioctx(pool_name
)
1513 ioctx
.remove_object(object_name
)
1514 except rados
.ObjectNotFound
:
1515 log
.warn("Object '{0}' was already removed".format(object_name
))