ceph/src/pybind/mgr/volumes/fs/volume.py

   1 import json
   2 import errno
   3 import logging
   4 from typing import TYPE_CHECKING
   5
   6 import cephfs
   7
   8 from mgr_util import CephfsClient
   9
  10 from .fs_util import listdir
  11
  12 from .operations.volume import create_volume, \
  13     delete_volume, rename_volume, list_volumes, open_volume, get_pool_names
  14 from .operations.group import open_group, create_group, remove_group, open_group_unique
  15 from .operations.subvolume import open_subvol, create_subvol, remove_subvol, \
  16     create_clone
  17 from .operations.trash import Trash
  18
  19 from .vol_spec import VolSpec
  20 from .exception import VolumeException, ClusterError, ClusterTimeout, EvictionError
  21 from .async_cloner import Cloner
  22 from .purge_queue import ThreadPoolPurgeQueueMixin
  23 from .operations.template import SubvolumeOpType
  24
  25 if TYPE_CHECKING:
  26     from volumes import Module
  27
  28 log = logging.getLogger(__name__)
  29
  30 ALLOWED_ACCESS_LEVELS = ('r', 'rw')
  31
  32
  33 def octal_str_to_decimal_int(mode):
  34     try:
  35         return int(mode, 8)
  36     except ValueError:
  37         raise VolumeException(-errno.EINVAL, "Invalid mode '{0}'".format(mode))
  38
  39
  40 def name_to_json(names):
  41     """
  42     convert the list of names to json
  43     """
  44     namedict = []
  45     for i in range(len(names)):
  46         namedict.append({'name': names[i].decode('utf-8')})
  47     return json.dumps(namedict, indent=4, sort_keys=True)
  48
  49
  50 class VolumeClient(CephfsClient["Module"]):
  51     def __init__(self, mgr):
  52         super().__init__(mgr)
  53         # volume specification
  54         self.volspec = VolSpec(mgr.rados.conf_get('client_snapdir'))
  55         self.cloner = Cloner(self, self.mgr.max_concurrent_clones, self.mgr.snapshot_clone_delay)
  56         self.purge_queue = ThreadPoolPurgeQueueMixin(self, 4)
  57         # on startup, queue purge job for available volumes to kickstart
  58         # purge for leftover subvolume entries in trash. note that, if the
  59         # trash directory does not exist or if there are no purge entries
  60         # available for a volume, the volume is removed from the purge
  61         # job list.
  62         fs_map = self.mgr.get('fs_map')
  63         for fs in fs_map['filesystems']:
  64             self.cloner.queue_job(fs['mdsmap']['fs_name'])
  65             self.purge_queue.queue_job(fs['mdsmap']['fs_name'])
  66
  67     def shutdown(self):
  68         # Overrides CephfsClient.shutdown()
  69         log.info("shutting down")
  70         # first, note that we're shutting down
  71         self.stopping.set()
  72         # stop clones
  73         self.cloner.shutdown()
  74         # stop purge threads
  75         self.purge_queue.shutdown()
  76         # last, delete all libcephfs handles from connection pool
  77         self.connection_pool.del_all_connections()
  78
  79     def cluster_log(self, msg, lvl=None):
  80         """
  81         log to cluster log with default log level as WARN.
  82         """
  83         if not lvl:
  84             lvl = self.mgr.ClusterLogPrio.WARN
  85         self.mgr.cluster_log("cluster", lvl, msg)
  86
  87     def volume_exception_to_retval(self, ve):
  88         """
  89         return a tuple representation from a volume exception
  90         """
  91         return ve.to_tuple()
  92
  93     ### volume operations -- create, rm, ls
  94
  95     def create_fs_volume(self, volname, placement):
  96         if self.is_stopping():
  97             return -errno.ESHUTDOWN, "", "shutdown in progress"
  98         return create_volume(self.mgr, volname, placement)
  99
 100     def delete_fs_volume(self, volname, confirm):
 101         if self.is_stopping():
 102             return -errno.ESHUTDOWN, "", "shutdown in progress"
 103
 104         if confirm != "--yes-i-really-mean-it":
 105             return -errno.EPERM, "", "WARNING: this will *PERMANENTLY DESTROY* all data " \
 106                 "stored in the filesystem '{0}'. If you are *ABSOLUTELY CERTAIN* " \
 107                 "that is what you want, re-issue the command followed by " \
 108                 "--yes-i-really-mean-it.".format(volname)
 109
 110         ret, out, err = self.mgr.check_mon_command({
 111             'prefix': 'config get',
 112             'key': 'mon_allow_pool_delete',
 113             'who': 'mon',
 114             'format': 'json',
 115         })
 116         mon_allow_pool_delete = json.loads(out)
 117         if not mon_allow_pool_delete:
 118             return -errno.EPERM, "", "pool deletion is disabled; you must first " \
 119                 "set the mon_allow_pool_delete config option to true before volumes " \
 120                 "can be deleted"
 121
 122         metadata_pool, data_pools = get_pool_names(self.mgr, volname)
 123         if not metadata_pool:
 124             return -errno.ENOENT, "", "volume {0} doesn't exist".format(volname)
 125         self.purge_queue.cancel_jobs(volname)
 126         self.connection_pool.del_connections(volname, wait=True)
 127         return delete_volume(self.mgr, volname, metadata_pool, data_pools)
 128
 129     def list_fs_volumes(self):
 130         if self.stopping.is_set():
 131             return -errno.ESHUTDOWN, "", "shutdown in progress"
 132         volumes = list_volumes(self.mgr)
 133         return 0, json.dumps(volumes, indent=4, sort_keys=True), ""
 134
 135     def rename_fs_volume(self, volname, newvolname, sure):
 136         if self.is_stopping():
 137             return -errno.ESHUTDOWN, "", "shutdown in progress"
 138
 139         if not sure:
 140             return (
 141                 -errno.EPERM, "",
 142                 "WARNING: This will rename the filesystem and possibly its "
 143                 "pools. It is a potentially disruptive operation, clients' "
 144                 "cephx credentials need reauthorized to access the file system "
 145                 "and its pools with the new name. Add --yes-i-really-mean-it "
 146                 "if you are sure you wish to continue.")
 147
 148         return rename_volume(self.mgr, volname, newvolname)
 149
 150     ### subvolume operations
 151
 152     def _create_subvolume(self, fs_handle, volname, group, subvolname, **kwargs):
 153         size       = kwargs['size']
 154         pool       = kwargs['pool_layout']
 155         uid        = kwargs['uid']
 156         gid        = kwargs['gid']
 157         mode       = kwargs['mode']
 158         isolate_nspace = kwargs['namespace_isolated']
 159
 160         oct_mode = octal_str_to_decimal_int(mode)
 161         try:
 162             create_subvol(
 163                 self.mgr, fs_handle, self.volspec, group, subvolname, size, isolate_nspace, pool, oct_mode, uid, gid)
 164         except VolumeException as ve:
 165             # kick the purge threads for async removal -- note that this
 166             # assumes that the subvolume is moved to trashcan for cleanup on error.
 167             self.purge_queue.queue_job(volname)
 168             raise ve
 169
 170     def create_subvolume(self, **kwargs):
 171         ret        = 0, "", ""
 172         volname    = kwargs['vol_name']
 173         subvolname = kwargs['sub_name']
 174         groupname  = kwargs['group_name']
 175         size       = kwargs['size']
 176         pool       = kwargs['pool_layout']
 177         uid        = kwargs['uid']
 178         gid        = kwargs['gid']
 179         mode       = kwargs['mode']
 180         isolate_nspace = kwargs['namespace_isolated']
 181
 182         try:
 183             with open_volume(self, volname) as fs_handle:
 184                 with open_group(fs_handle, self.volspec, groupname) as group:
 185                     try:
 186                         with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.CREATE) as subvolume:
 187                             # idempotent creation -- valid. Attributes set is supported.
 188                             attrs = {
 189                                 'uid': uid if uid else subvolume.uid,
 190                                 'gid': gid if gid else subvolume.gid,
 191                                 'mode': octal_str_to_decimal_int(mode),
 192                                 'data_pool': pool,
 193                                 'pool_namespace': subvolume.namespace if isolate_nspace else None,
 194                                 'quota': size
 195                             }
 196                             subvolume.set_attrs(subvolume.path, attrs)
 197                     except VolumeException as ve:
 198                         if ve.errno == -errno.ENOENT:
 199                             self._create_subvolume(fs_handle, volname, group, subvolname, **kwargs)
 200                         else:
 201                             raise
 202         except VolumeException as ve:
 203             # volume/group does not exist or subvolume creation failed
 204             ret = self.volume_exception_to_retval(ve)
 205         return ret
 206
 207     def remove_subvolume(self, **kwargs):
 208         ret         = 0, "", ""
 209         volname     = kwargs['vol_name']
 210         subvolname  = kwargs['sub_name']
 211         groupname   = kwargs['group_name']
 212         force       = kwargs['force']
 213         retainsnaps = kwargs['retain_snapshots']
 214
 215         try:
 216             with open_volume(self, volname) as fs_handle:
 217                 with open_group(fs_handle, self.volspec, groupname) as group:
 218                     remove_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, force, retainsnaps)
 219                     # kick the purge threads for async removal -- note that this
 220                     # assumes that the subvolume is moved to trash can.
 221                     # TODO: make purge queue as singleton so that trash can kicks
 222                     # the purge threads on dump.
 223                     self.purge_queue.queue_job(volname)
 224         except VolumeException as ve:
 225             if ve.errno == -errno.EAGAIN and not force:
 226                 ve = VolumeException(ve.errno, ve.error_str + " (use --force to override)")
 227                 ret = self.volume_exception_to_retval(ve)
 228             elif not (ve.errno == -errno.ENOENT and force):
 229                 ret = self.volume_exception_to_retval(ve)
 230         return ret
 231
 232     def authorize_subvolume(self, **kwargs):
 233         ret = 0, "", ""
 234         volname     = kwargs['vol_name']
 235         subvolname  = kwargs['sub_name']
 236         authid      = kwargs['auth_id']
 237         groupname   = kwargs['group_name']
 238         accesslevel = kwargs['access_level']
 239         tenant_id   = kwargs['tenant_id']
 240         allow_existing_id = kwargs['allow_existing_id']
 241
 242         try:
 243             with open_volume(self, volname) as fs_handle:
 244                 with open_group(fs_handle, self.volspec, groupname) as group:
 245                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.ALLOW_ACCESS) as subvolume:
 246                         key = subvolume.authorize(authid, accesslevel, tenant_id, allow_existing_id)
 247                         ret = 0, key, ""
 248         except VolumeException as ve:
 249             ret = self.volume_exception_to_retval(ve)
 250         return ret
 251
 252     def deauthorize_subvolume(self, **kwargs):
 253         ret = 0, "", ""
 254         volname     = kwargs['vol_name']
 255         subvolname  = kwargs['sub_name']
 256         authid      = kwargs['auth_id']
 257         groupname   = kwargs['group_name']
 258
 259         try:
 260             with open_volume(self, volname) as fs_handle:
 261                 with open_group(fs_handle, self.volspec, groupname) as group:
 262                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.DENY_ACCESS) as subvolume:
 263                         subvolume.deauthorize(authid)
 264         except VolumeException as ve:
 265             ret = self.volume_exception_to_retval(ve)
 266         return ret
 267
 268     def authorized_list(self, **kwargs):
 269         ret = 0, "", ""
 270         volname     = kwargs['vol_name']
 271         subvolname  = kwargs['sub_name']
 272         groupname   = kwargs['group_name']
 273
 274         try:
 275             with open_volume(self, volname) as fs_handle:
 276                 with open_group(fs_handle, self.volspec, groupname) as group:
 277                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.AUTH_LIST) as subvolume:
 278                         auths = subvolume.authorized_list()
 279                         ret = 0, json.dumps(auths, indent=4, sort_keys=True), ""
 280         except VolumeException as ve:
 281             ret = self.volume_exception_to_retval(ve)
 282         return ret
 283
 284     def evict(self, **kwargs):
 285         ret = 0, "", ""
 286         volname     = kwargs['vol_name']
 287         subvolname  = kwargs['sub_name']
 288         authid      = kwargs['auth_id']
 289         groupname   = kwargs['group_name']
 290
 291         try:
 292             with open_volume(self, volname) as fs_handle:
 293                 with open_group(fs_handle, self.volspec, groupname) as group:
 294                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.EVICT) as subvolume:
 295                         key = subvolume.evict(volname, authid)
 296                         ret = 0, "", ""
 297         except (VolumeException, ClusterTimeout, ClusterError, EvictionError) as e:
 298             if isinstance(e, VolumeException):
 299                 ret = self.volume_exception_to_retval(e)
 300             elif isinstance(e, ClusterTimeout):
 301                 ret = -errno.ETIMEDOUT , "", "Timedout trying to talk to ceph cluster"
 302             elif isinstance(e, ClusterError):
 303                 ret = e._result_code , "", e._result_str
 304             elif isinstance(e, EvictionError):
 305                 ret = -errno.EINVAL, "", str(e)
 306         return ret
 307
 308     def resize_subvolume(self, **kwargs):
 309         ret        = 0, "", ""
 310         volname    = kwargs['vol_name']
 311         subvolname = kwargs['sub_name']
 312         newsize    = kwargs['new_size']
 313         noshrink   = kwargs['no_shrink']
 314         groupname  = kwargs['group_name']
 315
 316         try:
 317             with open_volume(self, volname) as fs_handle:
 318                 with open_group(fs_handle, self.volspec, groupname) as group:
 319                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.RESIZE) as subvolume:
 320                         nsize, usedbytes = subvolume.resize(newsize, noshrink)
 321                         ret = 0, json.dumps(
 322                             [{'bytes_used': usedbytes},{'bytes_quota': nsize},
 323                              {'bytes_pcent': "undefined" if nsize == 0 else '{0:.2f}'.format((float(usedbytes) / nsize) * 100.0)}],
 324                             indent=4, sort_keys=True), ""
 325         except VolumeException as ve:
 326             ret = self.volume_exception_to_retval(ve)
 327         return ret
 328
 329     def subvolume_pin(self, **kwargs):
 330         ret         = 0, "", ""
 331         volname     = kwargs['vol_name']
 332         subvolname  = kwargs['sub_name']
 333         pin_type    = kwargs['pin_type']
 334         pin_setting = kwargs['pin_setting']
 335         groupname   = kwargs['group_name']
 336
 337         try:
 338             with open_volume(self, volname) as fs_handle:
 339                 with open_group(fs_handle, self.volspec, groupname) as group:
 340                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.PIN) as subvolume:
 341                         subvolume.pin(pin_type, pin_setting)
 342                         ret = 0, json.dumps({}), ""
 343         except VolumeException as ve:
 344             ret = self.volume_exception_to_retval(ve)
 345         return ret
 346
 347     def subvolume_getpath(self, **kwargs):
 348         ret        = None
 349         volname    = kwargs['vol_name']
 350         subvolname = kwargs['sub_name']
 351         groupname  = kwargs['group_name']
 352
 353         try:
 354             with open_volume(self, volname) as fs_handle:
 355                 with open_group(fs_handle, self.volspec, groupname) as group:
 356                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.GETPATH) as subvolume:
 357                         subvolpath = subvolume.path
 358                         ret = 0, subvolpath.decode("utf-8"), ""
 359         except VolumeException as ve:
 360             ret = self.volume_exception_to_retval(ve)
 361         return ret
 362
 363     def subvolume_info(self, **kwargs):
 364         ret        = None
 365         volname    = kwargs['vol_name']
 366         subvolname = kwargs['sub_name']
 367         groupname  = kwargs['group_name']
 368
 369         try:
 370             with open_volume(self, volname) as fs_handle:
 371                 with open_group(fs_handle, self.volspec, groupname) as group:
 372                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.INFO) as subvolume:
 373                         mon_addr_lst = []
 374                         mon_map_mons = self.mgr.get('mon_map')['mons']
 375                         for mon in mon_map_mons:
 376                             ip_port = mon['addr'].split("/")[0]
 377                             mon_addr_lst.append(ip_port)
 378
 379                         subvol_info_dict = subvolume.info()
 380                         subvol_info_dict["mon_addrs"] = mon_addr_lst
 381                         ret = 0, json.dumps(subvol_info_dict, indent=4, sort_keys=True), ""
 382         except VolumeException as ve:
 383             ret = self.volume_exception_to_retval(ve)
 384         return ret
 385
 386     def set_user_metadata(self, **kwargs):
 387         ret        = 0, "", ""
 388         volname    = kwargs['vol_name']
 389         subvolname = kwargs['sub_name']
 390         groupname  = kwargs['group_name']
 391         keyname    = kwargs['key_name']
 392         value      = kwargs['value']
 393
 394         try:
 395             with open_volume(self, volname) as fs_handle:
 396                 with open_group(fs_handle, self.volspec, groupname) as group:
 397                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_SET) as subvolume:
 398                         subvolume.set_user_metadata(keyname, value)
 399         except VolumeException as ve:
 400             ret = self.volume_exception_to_retval(ve)
 401         return ret
 402
 403     def get_user_metadata(self, **kwargs):
 404         ret        = 0, "", ""
 405         volname    = kwargs['vol_name']
 406         subvolname = kwargs['sub_name']
 407         groupname  = kwargs['group_name']
 408         keyname    = kwargs['key_name']
 409
 410         try:
 411             with open_volume(self, volname) as fs_handle:
 412                 with open_group(fs_handle, self.volspec, groupname) as group:
 413                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_GET) as subvolume:
 414                         value = subvolume.get_user_metadata(keyname)
 415                         ret = 0, value, ""
 416         except VolumeException as ve:
 417             ret = self.volume_exception_to_retval(ve)
 418         return ret
 419
 420     def list_user_metadata(self, **kwargs):
 421         ret        = 0, "", ""
 422         volname    = kwargs['vol_name']
 423         subvolname = kwargs['sub_name']
 424         groupname  = kwargs['group_name']
 425
 426         try:
 427             with open_volume(self, volname) as fs_handle:
 428                 with open_group(fs_handle, self.volspec, groupname) as group:
 429                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_LIST) as subvolume:
 430                         subvol_metadata_dict = subvolume.list_user_metadata()
 431                         ret = 0, json.dumps(subvol_metadata_dict, indent=4, sort_keys=True), ""
 432         except VolumeException as ve:
 433             ret = self.volume_exception_to_retval(ve)
 434         return ret
 435
 436     def remove_user_metadata(self, **kwargs):
 437         ret        = 0, "", ""
 438         volname    = kwargs['vol_name']
 439         subvolname = kwargs['sub_name']
 440         groupname  = kwargs['group_name']
 441         keyname    = kwargs['key_name']
 442         force      = kwargs['force']
 443
 444         try:
 445             with open_volume(self, volname) as fs_handle:
 446                 with open_group(fs_handle, self.volspec, groupname) as group:
 447                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.USER_METADATA_REMOVE) as subvolume:
 448                         subvolume.remove_user_metadata(keyname)
 449         except VolumeException as ve:
 450             if not (ve.errno == -errno.ENOENT and force):
 451                 ret = self.volume_exception_to_retval(ve)
 452         return ret
 453
 454     def list_subvolumes(self, **kwargs):
 455         ret        = 0, "", ""
 456         volname    = kwargs['vol_name']
 457         groupname  = kwargs['group_name']
 458
 459         try:
 460             with open_volume(self, volname) as fs_handle:
 461                 with open_group(fs_handle, self.volspec, groupname) as group:
 462                     subvolumes = group.list_subvolumes()
 463                     ret = 0, name_to_json(subvolumes), ""
 464         except VolumeException as ve:
 465             ret = self.volume_exception_to_retval(ve)
 466         return ret
 467
 468     ### subvolume snapshot
 469
 470     def create_subvolume_snapshot(self, **kwargs):
 471         ret        = 0, "", ""
 472         volname    = kwargs['vol_name']
 473         subvolname = kwargs['sub_name']
 474         snapname   = kwargs['snap_name']
 475         groupname  = kwargs['group_name']
 476
 477         try:
 478             with open_volume(self, volname) as fs_handle:
 479                 with open_group(fs_handle, self.volspec, groupname) as group:
 480                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_CREATE) as subvolume:
 481                         subvolume.create_snapshot(snapname)
 482         except VolumeException as ve:
 483             ret = self.volume_exception_to_retval(ve)
 484         return ret
 485
 486     def remove_subvolume_snapshot(self, **kwargs):
 487         ret        = 0, "", ""
 488         volname    = kwargs['vol_name']
 489         subvolname = kwargs['sub_name']
 490         snapname   = kwargs['snap_name']
 491         groupname  = kwargs['group_name']
 492         force      = kwargs['force']
 493
 494         try:
 495             with open_volume(self, volname) as fs_handle:
 496                 with open_group(fs_handle, self.volspec, groupname) as group:
 497                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_REMOVE) as subvolume:
 498                         subvolume.remove_snapshot(snapname)
 499         except VolumeException as ve:
 500             # ESTALE serves as an error to state that subvolume is currently stale due to internal removal and,
 501             # we should tickle the purge jobs to purge the same
 502             if ve.errno == -errno.ESTALE:
 503                 self.purge_queue.queue_job(volname)
 504             elif not (ve.errno == -errno.ENOENT and force):
 505                 ret = self.volume_exception_to_retval(ve)
 506         return ret
 507
 508     def subvolume_snapshot_info(self, **kwargs):
 509         ret        = 0, "", ""
 510         volname    = kwargs['vol_name']
 511         subvolname = kwargs['sub_name']
 512         snapname   = kwargs['snap_name']
 513         groupname  = kwargs['group_name']
 514
 515         try:
 516             with open_volume(self, volname) as fs_handle:
 517                 with open_group(fs_handle, self.volspec, groupname) as group:
 518                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_INFO) as subvolume:
 519                         snap_info_dict = subvolume.snapshot_info(snapname)
 520                         ret = 0, json.dumps(snap_info_dict, indent=4, sort_keys=True), ""
 521         except VolumeException as ve:
 522                 ret = self.volume_exception_to_retval(ve)
 523         return ret
 524
 525     def set_subvolume_snapshot_metadata(self, **kwargs):
 526         ret        = 0, "", ""
 527         volname    = kwargs['vol_name']
 528         subvolname = kwargs['sub_name']
 529         snapname   = kwargs['snap_name']
 530         groupname  = kwargs['group_name']
 531         keyname    = kwargs['key_name']
 532         value      = kwargs['value']
 533
 534         try:
 535             with open_volume(self, volname) as fs_handle:
 536                 with open_group(fs_handle, self.volspec, groupname) as group:
 537                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_METADATA_SET) as subvolume:
 538                         if not snapname.encode('utf-8') in subvolume.list_snapshots():
 539                             raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
 540                         subvolume.set_snapshot_metadata(snapname, keyname, value)
 541         except VolumeException as ve:
 542             ret = self.volume_exception_to_retval(ve)
 543         return ret
 544
 545     def get_subvolume_snapshot_metadata(self, **kwargs):
 546         ret        = 0, "", ""
 547         volname    = kwargs['vol_name']
 548         subvolname = kwargs['sub_name']
 549         snapname   = kwargs['snap_name']
 550         groupname  = kwargs['group_name']
 551         keyname    = kwargs['key_name']
 552
 553         try:
 554             with open_volume(self, volname) as fs_handle:
 555                 with open_group(fs_handle, self.volspec, groupname) as group:
 556                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_METADATA_GET) as subvolume:
 557                         if not snapname.encode('utf-8') in subvolume.list_snapshots():
 558                             raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
 559                         value = subvolume.get_snapshot_metadata(snapname, keyname)
 560                         ret = 0, value, ""
 561         except VolumeException as ve:
 562             ret = self.volume_exception_to_retval(ve)
 563         return ret
 564
 565     def list_subvolume_snapshot_metadata(self, **kwargs):
 566         ret        = 0, "", ""
 567         volname    = kwargs['vol_name']
 568         subvolname = kwargs['sub_name']
 569         snapname   = kwargs['snap_name']
 570         groupname  = kwargs['group_name']
 571
 572         try:
 573             with open_volume(self, volname) as fs_handle:
 574                 with open_group(fs_handle, self.volspec, groupname) as group:
 575                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_METADATA_LIST) as subvolume:
 576                         if not snapname.encode('utf-8') in subvolume.list_snapshots():
 577                             raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
 578                         snap_metadata_dict = subvolume.list_snapshot_metadata(snapname)
 579                         ret = 0, json.dumps(snap_metadata_dict, indent=4, sort_keys=True), ""
 580         except VolumeException as ve:
 581             ret = self.volume_exception_to_retval(ve)
 582         return ret
 583
 584     def remove_subvolume_snapshot_metadata(self, **kwargs):
 585         ret        = 0, "", ""
 586         volname    = kwargs['vol_name']
 587         subvolname = kwargs['sub_name']
 588         snapname   = kwargs['snap_name']
 589         groupname  = kwargs['group_name']
 590         keyname    = kwargs['key_name']
 591         force      = kwargs['force']
 592
 593         try:
 594             with open_volume(self, volname) as fs_handle:
 595                 with open_group(fs_handle, self.volspec, groupname) as group:
 596                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_METADATA_REMOVE) as subvolume:
 597                         if not snapname.encode('utf-8') in subvolume.list_snapshots():
 598                             raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(snapname))
 599                         subvolume.remove_snapshot_metadata(snapname, keyname)
 600         except VolumeException as ve:
 601             if not (ve.errno == -errno.ENOENT and force):
 602                 ret = self.volume_exception_to_retval(ve)
 603         return ret
 604
 605     def list_subvolume_snapshots(self, **kwargs):
 606         ret        = 0, "", ""
 607         volname    = kwargs['vol_name']
 608         subvolname = kwargs['sub_name']
 609         groupname  = kwargs['group_name']
 610
 611         try:
 612             with open_volume(self, volname) as fs_handle:
 613                 with open_group(fs_handle, self.volspec, groupname) as group:
 614                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_LIST) as subvolume:
 615                         snapshots = subvolume.list_snapshots()
 616                         ret = 0, name_to_json(snapshots), ""
 617         except VolumeException as ve:
 618             ret = self.volume_exception_to_retval(ve)
 619         return ret
 620
 621     def protect_subvolume_snapshot(self, **kwargs):
 622         ret        = 0, "", "Deprecation warning: 'snapshot protect' call is deprecated and will be removed in a future release"
 623         volname    = kwargs['vol_name']
 624         subvolname = kwargs['sub_name']
 625         groupname  = kwargs['group_name']
 626
 627         try:
 628             with open_volume(self, volname) as fs_handle:
 629                 with open_group(fs_handle, self.volspec, groupname) as group:
 630                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_PROTECT) as subvolume:
 631                         log.warning("snapshot protect call is deprecated and will be removed in a future release")
 632         except VolumeException as ve:
 633             ret = self.volume_exception_to_retval(ve)
 634         return ret
 635
 636     def unprotect_subvolume_snapshot(self, **kwargs):
 637         ret        = 0, "", "Deprecation warning: 'snapshot unprotect' call is deprecated and will be removed in a future release"
 638         volname    = kwargs['vol_name']
 639         subvolname = kwargs['sub_name']
 640         groupname  = kwargs['group_name']
 641
 642         try:
 643             with open_volume(self, volname) as fs_handle:
 644                 with open_group(fs_handle, self.volspec, groupname) as group:
 645                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_UNPROTECT) as subvolume:
 646                         log.warning("snapshot unprotect call is deprecated and will be removed in a future release")
 647         except VolumeException as ve:
 648             ret = self.volume_exception_to_retval(ve)
 649         return ret
 650
 651     def _prepare_clone_subvolume(self, fs_handle, volname, s_subvolume, s_snapname, t_group, t_subvolname, **kwargs):
 652         t_pool              = kwargs['pool_layout']
 653         s_subvolname        = kwargs['sub_name']
 654         s_groupname         = kwargs['group_name']
 655         t_groupname         = kwargs['target_group_name']
 656
 657         create_clone(self.mgr, fs_handle, self.volspec, t_group, t_subvolname, t_pool, volname, s_subvolume, s_snapname)
 658         with open_subvol(self.mgr, fs_handle, self.volspec, t_group, t_subvolname, SubvolumeOpType.CLONE_INTERNAL) as t_subvolume:
 659             try:
 660                 if t_groupname == s_groupname and t_subvolname == s_subvolname:
 661                     t_subvolume.attach_snapshot(s_snapname, t_subvolume)
 662                 else:
 663                     s_subvolume.attach_snapshot(s_snapname, t_subvolume)
 664                 self.cloner.queue_job(volname)
 665             except VolumeException as ve:
 666                 try:
 667                     t_subvolume.remove()
 668                     self.purge_queue.queue_job(volname)
 669                 except Exception as e:
 670                     log.warning("failed to cleanup clone subvolume '{0}' ({1})".format(t_subvolname, e))
 671                 raise ve
 672
 673     def _clone_subvolume_snapshot(self, fs_handle, volname, s_group, s_subvolume, **kwargs):
 674         s_snapname          = kwargs['snap_name']
 675         target_subvolname   = kwargs['target_sub_name']
 676         target_groupname    = kwargs['target_group_name']
 677         s_groupname         = kwargs['group_name']
 678
 679         if not s_snapname.encode('utf-8') in s_subvolume.list_snapshots():
 680             raise VolumeException(-errno.ENOENT, "snapshot '{0}' does not exist".format(s_snapname))
 681
 682         with open_group_unique(fs_handle, self.volspec, target_groupname, s_group, s_groupname) as target_group:
 683             try:
 684                 with open_subvol(self.mgr, fs_handle, self.volspec, target_group, target_subvolname, SubvolumeOpType.CLONE_CREATE):
 685                     raise VolumeException(-errno.EEXIST, "subvolume '{0}' exists".format(target_subvolname))
 686             except VolumeException as ve:
 687                 if ve.errno == -errno.ENOENT:
 688                     self._prepare_clone_subvolume(fs_handle, volname, s_subvolume, s_snapname,
 689                                                   target_group, target_subvolname, **kwargs)
 690                 else:
 691                     raise
 692
 693     def clone_subvolume_snapshot(self, **kwargs):
 694         ret        = 0, "", ""
 695         volname    = kwargs['vol_name']
 696         s_subvolname = kwargs['sub_name']
 697         s_groupname  = kwargs['group_name']
 698
 699         try:
 700             with open_volume(self, volname) as fs_handle:
 701                 with open_group(fs_handle, self.volspec, s_groupname) as s_group:
 702                     with open_subvol(self.mgr, fs_handle, self.volspec, s_group, s_subvolname, SubvolumeOpType.CLONE_SOURCE) as s_subvolume:
 703                         self._clone_subvolume_snapshot(fs_handle, volname, s_group, s_subvolume, **kwargs)
 704         except VolumeException as ve:
 705             ret = self.volume_exception_to_retval(ve)
 706         return ret
 707
 708     def clone_status(self, **kwargs):
 709         ret       = 0, "", ""
 710         volname   = kwargs['vol_name']
 711         clonename = kwargs['clone_name']
 712         groupname = kwargs['group_name']
 713
 714         try:
 715             with open_volume(self, volname) as fs_handle:
 716                 with open_group(fs_handle, self.volspec, groupname) as group:
 717                     with open_subvol(self.mgr, fs_handle, self.volspec, group, clonename, SubvolumeOpType.CLONE_STATUS) as subvolume:
 718                         ret = 0, json.dumps({'status' : subvolume.status}, indent=2), ""
 719         except VolumeException as ve:
 720             ret = self.volume_exception_to_retval(ve)
 721         return ret
 722
 723     def clone_cancel(self, **kwargs):
 724         ret       = 0, "", ""
 725         volname   = kwargs['vol_name']
 726         clonename = kwargs['clone_name']
 727         groupname = kwargs['group_name']
 728
 729         try:
 730             self.cloner.cancel_job(volname, (clonename, groupname))
 731         except VolumeException as ve:
 732             ret = self.volume_exception_to_retval(ve)
 733         return ret
 734
 735     ### group operations
 736
 737     def create_subvolume_group(self, **kwargs):
 738         ret       = 0, "", ""
 739         volname    = kwargs['vol_name']
 740         groupname = kwargs['group_name']
 741         pool      = kwargs['pool_layout']
 742         uid       = kwargs['uid']
 743         gid       = kwargs['gid']
 744         mode      = kwargs['mode']
 745
 746         try:
 747             with open_volume(self, volname) as fs_handle:
 748                 try:
 749                     with open_group(fs_handle, self.volspec, groupname):
 750                         # idempotent creation -- valid.
 751                         pass
 752                 except VolumeException as ve:
 753                     if ve.errno == -errno.ENOENT:
 754                         oct_mode = octal_str_to_decimal_int(mode)
 755                         create_group(fs_handle, self.volspec, groupname, pool, oct_mode, uid, gid)
 756                     else:
 757                         raise
 758         except VolumeException as ve:
 759             # volume does not exist or subvolume group creation failed
 760             ret = self.volume_exception_to_retval(ve)
 761         return ret
 762
 763     def remove_subvolume_group(self, **kwargs):
 764         ret       = 0, "", ""
 765         volname    = kwargs['vol_name']
 766         groupname = kwargs['group_name']
 767         force     = kwargs['force']
 768
 769         try:
 770             with open_volume(self, volname) as fs_handle:
 771                 remove_group(fs_handle, self.volspec, groupname)
 772         except VolumeException as ve:
 773             if not (ve.errno == -errno.ENOENT and force):
 774                 ret = self.volume_exception_to_retval(ve)
 775         return ret
 776
 777     def getpath_subvolume_group(self, **kwargs):
 778         volname    = kwargs['vol_name']
 779         groupname  = kwargs['group_name']
 780
 781         try:
 782             with open_volume(self, volname) as fs_handle:
 783                 with open_group(fs_handle, self.volspec, groupname) as group:
 784                     return 0, group.path.decode('utf-8'), ""
 785         except VolumeException as ve:
 786             return self.volume_exception_to_retval(ve)
 787
 788     def list_subvolume_groups(self, **kwargs):
 789         volname = kwargs['vol_name']
 790         ret     = 0, '[]', ""
 791         volume_exists = False
 792         try:
 793             with open_volume(self, volname) as fs_handle:
 794                 volume_exists = True
 795                 groups = listdir(fs_handle, self.volspec.base_dir, filter_entries=[Trash.GROUP_NAME.encode('utf-8')])
 796                 ret = 0, name_to_json(groups), ""
 797         except VolumeException as ve:
 798             if not ve.errno == -errno.ENOENT or not volume_exists:
 799                 ret = self.volume_exception_to_retval(ve)
 800         return ret
 801
 802     def pin_subvolume_group(self, **kwargs):
 803         ret           = 0, "", ""
 804         volname       = kwargs['vol_name']
 805         groupname     = kwargs['group_name']
 806         pin_type      = kwargs['pin_type']
 807         pin_setting   = kwargs['pin_setting']
 808
 809         try:
 810             with open_volume(self, volname) as fs_handle:
 811                 with open_group(fs_handle, self.volspec, groupname) as group:
 812                     group.pin(pin_type, pin_setting)
 813                     ret = 0, json.dumps({}), ""
 814         except VolumeException as ve:
 815             ret = self.volume_exception_to_retval(ve)
 816         return ret
 817
 818     ### group snapshot
 819
 820     def create_subvolume_group_snapshot(self, **kwargs):
 821         ret       = -errno.ENOSYS, "", "subvolume group snapshots are not supported"
 822         volname   = kwargs['vol_name']
 823         groupname = kwargs['group_name']
 824         # snapname  = kwargs['snap_name']
 825
 826         try:
 827             with open_volume(self, volname) as fs_handle:
 828                 with open_group(fs_handle, self.volspec, groupname) as group:
 829                     # as subvolumes are marked with the vxattr ceph.dir.subvolume deny snapshots
 830                     # at the subvolume group (see: https://tracker.ceph.com/issues/46074)
 831                     # group.create_snapshot(snapname)
 832                     pass
 833         except VolumeException as ve:
 834             ret = self.volume_exception_to_retval(ve)
 835         return ret
 836
 837     def remove_subvolume_group_snapshot(self, **kwargs):
 838         ret       = 0, "", ""
 839         volname   = kwargs['vol_name']
 840         groupname = kwargs['group_name']
 841         snapname  = kwargs['snap_name']
 842         force     = kwargs['force']
 843
 844         try:
 845             with open_volume(self, volname) as fs_handle:
 846                 with open_group(fs_handle, self.volspec, groupname) as group:
 847                     group.remove_snapshot(snapname)
 848         except VolumeException as ve:
 849             if not (ve.errno == -errno.ENOENT and force):
 850                 ret = self.volume_exception_to_retval(ve)
 851         return ret
 852
 853     def list_subvolume_group_snapshots(self, **kwargs):
 854         ret       = 0, "", ""
 855         volname   = kwargs['vol_name']
 856         groupname = kwargs['group_name']
 857
 858         try:
 859             with open_volume(self, volname) as fs_handle:
 860                 with open_group(fs_handle, self.volspec, groupname) as group:
 861                     snapshots = group.list_snapshots()
 862                     ret = 0, name_to_json(snapshots), ""
 863         except VolumeException as ve:
 864             ret = self.volume_exception_to_retval(ve)
 865         return ret