ceph/src/pybind/mgr/dashboard/controllers/cephfs.py

   1 # -*- coding: utf-8 -*-
   2 import os
   3 from collections import defaultdict
   4
   5 import cephfs
   6 import cherrypy
   7
   8 from .. import mgr
   9 from ..exceptions import DashboardException
  10 from ..security import Scope
  11 from ..services.ceph_service import CephService
  12 from ..services.cephfs import CephFS as CephFS_
  13 from ..services.exception import handle_cephfs_error
  14 from ..tools import ViewCache
  15 from . import APIDoc, APIRouter, EndpointDoc, RESTController, UIRouter, allow_empty_body
  16
  17 GET_QUOTAS_SCHEMA = {
  18     'max_bytes': (int, ''),
  19     'max_files': (int, '')
  20 }
  21
  22
  23 @APIRouter('/cephfs', Scope.CEPHFS)
  24 @APIDoc("Cephfs Management API", "Cephfs")
  25 class CephFS(RESTController):
  26     def __init__(self):  # pragma: no cover
  27         super().__init__()
  28
  29         # Stateful instances of CephFSClients, hold cached results.  Key to
  30         # dict is FSCID
  31         self.cephfs_clients = {}
  32
  33     def list(self):
  34         fsmap = mgr.get("fs_map")
  35         return fsmap['filesystems']
  36
  37     def get(self, fs_id):
  38         fs_id = self.fs_id_to_int(fs_id)
  39         return self.fs_status(fs_id)
  40
  41     @RESTController.Resource('GET')
  42     def clients(self, fs_id):
  43         fs_id = self.fs_id_to_int(fs_id)
  44
  45         return self._clients(fs_id)
  46
  47     @RESTController.Resource('DELETE', path='/client/{client_id}')
  48     def evict(self, fs_id, client_id):
  49         fs_id = self.fs_id_to_int(fs_id)
  50         client_id = self.client_id_to_int(client_id)
  51
  52         return self._evict(fs_id, client_id)
  53
  54     @RESTController.Resource('GET')
  55     def mds_counters(self, fs_id, counters=None):
  56         fs_id = self.fs_id_to_int(fs_id)
  57         return self._mds_counters(fs_id, counters)
  58
  59     def _mds_counters(self, fs_id, counters=None):
  60         """
  61         Result format: map of daemon name to map of counter to list of datapoints
  62         rtype: dict[str, dict[str, list]]
  63         """
  64
  65         if counters is None:
  66             # Opinionated list of interesting performance counters for the GUI
  67             counters = [
  68                 "mds_server.handle_client_request",
  69                 "mds_log.ev",
  70                 "mds_cache.num_strays",
  71                 "mds.exported",
  72                 "mds.exported_inodes",
  73                 "mds.imported",
  74                 "mds.imported_inodes",
  75                 "mds.inodes",
  76                 "mds.caps",
  77                 "mds.subtrees",
  78                 "mds_mem.ino"
  79             ]
  80
  81         result: dict = {}
  82         mds_names = self._get_mds_names(fs_id)
  83
  84         for mds_name in mds_names:
  85             result[mds_name] = {}
  86             for counter in counters:
  87                 data = mgr.get_counter("mds", mds_name, counter)
  88                 if data is not None:
  89                     result[mds_name][counter] = data[counter]
  90                 else:
  91                     result[mds_name][counter] = []
  92
  93         return dict(result)
  94
  95     @staticmethod
  96     def fs_id_to_int(fs_id):
  97         try:
  98             return int(fs_id)
  99         except ValueError:
 100             raise DashboardException(code='invalid_cephfs_id',
 101                                      msg="Invalid cephfs ID {}".format(fs_id),
 102                                      component='cephfs')
 103
 104     @staticmethod
 105     def client_id_to_int(client_id):
 106         try:
 107             return int(client_id)
 108         except ValueError:
 109             raise DashboardException(code='invalid_cephfs_client_id',
 110                                      msg="Invalid cephfs client ID {}".format(client_id),
 111                                      component='cephfs')
 112
 113     def _get_mds_names(self, filesystem_id=None):
 114         names = []
 115
 116         fsmap = mgr.get("fs_map")
 117         for fs in fsmap['filesystems']:
 118             if filesystem_id is not None and fs['id'] != filesystem_id:
 119                 continue
 120             names.extend([info['name']
 121                           for _, info in fs['mdsmap']['info'].items()])
 122
 123         if filesystem_id is None:
 124             names.extend(info['name'] for info in fsmap['standbys'])
 125
 126         return names
 127
 128     def _append_mds_metadata(self, mds_versions, metadata_key):
 129         metadata = mgr.get_metadata('mds', metadata_key)
 130         if metadata is None:
 131             return
 132         mds_versions[metadata.get('ceph_version', 'unknown')].append(metadata_key)
 133
 134     # pylint: disable=too-many-statements,too-many-branches
 135     def fs_status(self, fs_id):
 136         mds_versions: dict = defaultdict(list)
 137
 138         fsmap = mgr.get("fs_map")
 139         filesystem = None
 140         for fs in fsmap['filesystems']:
 141             if fs['id'] == fs_id:
 142                 filesystem = fs
 143                 break
 144
 145         if filesystem is None:
 146             raise cherrypy.HTTPError(404,
 147                                      "CephFS id {0} not found".format(fs_id))
 148
 149         rank_table = []
 150
 151         mdsmap = filesystem['mdsmap']
 152
 153         client_count = 0
 154
 155         for rank in mdsmap["in"]:
 156             up = "mds_{0}".format(rank) in mdsmap["up"]
 157             if up:
 158                 gid = mdsmap['up']["mds_{0}".format(rank)]
 159                 info = mdsmap['info']['gid_{0}'.format(gid)]
 160                 dns = mgr.get_latest("mds", info['name'], "mds_mem.dn")
 161                 inos = mgr.get_latest("mds", info['name'], "mds_mem.ino")
 162                 dirs = mgr.get_latest("mds", info['name'], "mds_mem.dir")
 163                 caps = mgr.get_latest("mds", info['name'], "mds_mem.cap")
 164
 165                 if rank == 0:
 166                     client_count = mgr.get_latest("mds", info['name'],
 167                                                   "mds_sessions.session_count")
 168                 elif client_count == 0:
 169                     # In case rank 0 was down, look at another rank's
 170                     # sessionmap to get an indication of clients.
 171                     client_count = mgr.get_latest("mds", info['name'],
 172                                                   "mds_sessions.session_count")
 173
 174                 laggy = "laggy_since" in info
 175
 176                 state = info['state'].split(":")[1]
 177                 if laggy:
 178                     state += "(laggy)"
 179
 180                 # Populate based on context of state, e.g. client
 181                 # ops for an active daemon, replay progress, reconnect
 182                 # progress
 183                 if state == "active":
 184                     activity = CephService.get_rate("mds",
 185                                                     info['name'],
 186                                                     "mds_server.handle_client_request")
 187                 else:
 188                     activity = 0.0  # pragma: no cover
 189
 190                 self._append_mds_metadata(mds_versions, info['name'])
 191                 rank_table.append(
 192                     {
 193                         "rank": rank,
 194                         "state": state,
 195                         "mds": info['name'],
 196                         "activity": activity,
 197                         "dns": dns,
 198                         "inos": inos,
 199                         "dirs": dirs,
 200                         "caps": caps
 201                     }
 202                 )
 203
 204             else:
 205                 rank_table.append(
 206                     {
 207                         "rank": rank,
 208                         "state": "failed",
 209                         "mds": "",
 210                         "activity": 0.0,
 211                         "dns": 0,
 212                         "inos": 0,
 213                         "dirs": 0,
 214                         "caps": 0
 215                     }
 216                 )
 217
 218         # Find the standby replays
 219         # pylint: disable=unused-variable
 220         for gid_str, daemon_info in mdsmap['info'].items():
 221             if daemon_info['state'] != "up:standby-replay":
 222                 continue
 223
 224             inos = mgr.get_latest("mds", daemon_info['name'], "mds_mem.ino")
 225             dns = mgr.get_latest("mds", daemon_info['name'], "mds_mem.dn")
 226             dirs = mgr.get_latest("mds", daemon_info['name'], "mds_mem.dir")
 227             caps = mgr.get_latest("mds", daemon_info['name'], "mds_mem.cap")
 228
 229             activity = CephService.get_rate(
 230                 "mds", daemon_info['name'], "mds_log.replay")
 231
 232             rank_table.append(
 233                 {
 234                     "rank": "{0}-s".format(daemon_info['rank']),
 235                     "state": "standby-replay",
 236                     "mds": daemon_info['name'],
 237                     "activity": activity,
 238                     "dns": dns,
 239                     "inos": inos,
 240                     "dirs": dirs,
 241                     "caps": caps
 242                 }
 243             )
 244
 245         df = mgr.get("df")
 246         pool_stats = {p['id']: p['stats'] for p in df['pools']}
 247         osdmap = mgr.get("osd_map")
 248         pools = {p['pool']: p for p in osdmap['pools']}
 249         metadata_pool_id = mdsmap['metadata_pool']
 250         data_pool_ids = mdsmap['data_pools']
 251
 252         pools_table = []
 253         for pool_id in [metadata_pool_id] + data_pool_ids:
 254             pool_type = "metadata" if pool_id == metadata_pool_id else "data"
 255             stats = pool_stats[pool_id]
 256             pools_table.append({
 257                 "pool": pools[pool_id]['pool_name'],
 258                 "type": pool_type,
 259                 "used": stats['stored'],
 260                 "avail": stats['max_avail']
 261             })
 262
 263         standby_table = []
 264         for standby in fsmap['standbys']:
 265             self._append_mds_metadata(mds_versions, standby['name'])
 266             standby_table.append({
 267                 'name': standby['name']
 268             })
 269
 270         return {
 271             "cephfs": {
 272                 "id": fs_id,
 273                 "name": mdsmap['fs_name'],
 274                 "client_count": client_count,
 275                 "ranks": rank_table,
 276                 "pools": pools_table
 277             },
 278             "standbys": standby_table,
 279             "versions": mds_versions
 280         }
 281
 282     def _clients(self, fs_id):
 283         cephfs_clients = self.cephfs_clients.get(fs_id, None)
 284         if cephfs_clients is None:
 285             cephfs_clients = CephFSClients(mgr, fs_id)
 286             self.cephfs_clients[fs_id] = cephfs_clients
 287
 288         try:
 289             status, clients = cephfs_clients.get()
 290         except AttributeError:
 291             raise cherrypy.HTTPError(404,
 292                                      "No cephfs with id {0}".format(fs_id))
 293
 294         if clients is None:
 295             raise cherrypy.HTTPError(404,
 296                                      "No cephfs with id {0}".format(fs_id))
 297
 298         # Decorate the metadata with some fields that will be
 299         # indepdendent of whether it's a kernel or userspace
 300         # client, so that the javascript doesn't have to grok that.
 301         for client in clients:
 302             if "ceph_version" in client['client_metadata']:  # pragma: no cover - no complexity
 303                 client['type'] = "userspace"
 304                 client['version'] = client['client_metadata']['ceph_version']
 305                 client['hostname'] = client['client_metadata']['hostname']
 306                 client['root'] = client['client_metadata']['root']
 307             elif "kernel_version" in client['client_metadata']:  # pragma: no cover - no complexity
 308                 client['type'] = "kernel"
 309                 client['version'] = client['client_metadata']['kernel_version']
 310                 client['hostname'] = client['client_metadata']['hostname']
 311                 client['root'] = client['client_metadata']['root']
 312             else:  # pragma: no cover - no complexity there
 313                 client['type'] = "unknown"
 314                 client['version'] = ""
 315                 client['hostname'] = ""
 316
 317         return {
 318             'status': status,
 319             'data': clients
 320         }
 321
 322     def _evict(self, fs_id, client_id):
 323         clients = self._clients(fs_id)
 324         if not [c for c in clients['data'] if c['id'] == client_id]:
 325             raise cherrypy.HTTPError(404,
 326                                      "Client {0} does not exist in cephfs {1}".format(client_id,
 327                                                                                       fs_id))
 328         CephService.send_command('mds', 'client evict',
 329                                  srv_spec='{0}:0'.format(fs_id), id=client_id)
 330
 331     @staticmethod
 332     def _cephfs_instance(fs_id):
 333         """
 334         :param fs_id: The filesystem identifier.
 335         :type fs_id: int | str
 336         :return: A instance of the CephFS class.
 337         """
 338         fs_name = CephFS_.fs_name_from_id(fs_id)
 339         if fs_name is None:
 340             raise cherrypy.HTTPError(404, "CephFS id {} not found".format(fs_id))
 341         return CephFS_(fs_name)
 342
 343     @RESTController.Resource('GET')
 344     def get_root_directory(self, fs_id):
 345         """
 346         The root directory that can't be fetched using ls_dir (api).
 347         :param fs_id: The filesystem identifier.
 348         :return: The root directory
 349         :rtype: dict
 350         """
 351         try:
 352             return self._get_root_directory(self._cephfs_instance(fs_id))
 353         except (cephfs.PermissionError, cephfs.ObjectNotFound):  # pragma: no cover
 354             return None
 355
 356     def _get_root_directory(self, cfs):
 357         """
 358         The root directory that can't be fetched using ls_dir (api).
 359         It's used in ls_dir (ui-api) and in get_root_directory (api).
 360         :param cfs: CephFS service instance
 361         :type cfs: CephFS
 362         :return: The root directory
 363         :rtype: dict
 364         """
 365         return cfs.get_directory(os.sep.encode())
 366
 367     @handle_cephfs_error()
 368     @RESTController.Resource('GET')
 369     def ls_dir(self, fs_id, path=None, depth=1):
 370         """
 371         List directories of specified path.
 372         :param fs_id: The filesystem identifier.
 373         :param path: The path where to start listing the directory content.
 374         Defaults to '/' if not set.
 375         :type path: str | bytes
 376         :param depth: The number of steps to go down the directory tree.
 377         :type depth: int | str
 378         :return: The names of the directories below the specified path.
 379         :rtype: list
 380         """
 381         path = self._set_ls_dir_path(path)
 382         try:
 383             cfs = self._cephfs_instance(fs_id)
 384             paths = cfs.ls_dir(path, depth)
 385         except (cephfs.PermissionError, cephfs.ObjectNotFound):  # pragma: no cover
 386             paths = []
 387         return paths
 388
 389     def _set_ls_dir_path(self, path):
 390         """
 391         Transforms input path parameter of ls_dir methods (api and ui-api).
 392         :param path: The path where to start listing the directory content.
 393         Defaults to '/' if not set.
 394         :type path: str | bytes
 395         :return: Normalized path or root path
 396         :return: str
 397         """
 398         if path is None:
 399             path = os.sep
 400         else:
 401             path = os.path.normpath(path)
 402         return path
 403
 404     @RESTController.Resource('POST', path='/tree')
 405     @allow_empty_body
 406     def mk_tree(self, fs_id, path):
 407         """
 408         Create a directory.
 409         :param fs_id: The filesystem identifier.
 410         :param path: The path of the directory.
 411         """
 412         cfs = self._cephfs_instance(fs_id)
 413         cfs.mk_dirs(path)
 414
 415     @RESTController.Resource('DELETE', path='/tree')
 416     def rm_tree(self, fs_id, path):
 417         """
 418         Remove a directory.
 419         :param fs_id: The filesystem identifier.
 420         :param path: The path of the directory.
 421         """
 422         cfs = self._cephfs_instance(fs_id)
 423         cfs.rm_dir(path)
 424
 425     @RESTController.Resource('PUT', path='/quota')
 426     @allow_empty_body
 427     def quota(self, fs_id, path, max_bytes=None, max_files=None):
 428         """
 429         Set the quotas of the specified path.
 430         :param fs_id: The filesystem identifier.
 431         :param path: The path of the directory/file.
 432         :param max_bytes: The byte limit.
 433         :param max_files: The file limit.
 434         """
 435         cfs = self._cephfs_instance(fs_id)
 436         return cfs.set_quotas(path, max_bytes, max_files)
 437
 438     @RESTController.Resource('GET', path='/quota')
 439     @EndpointDoc("Get Cephfs Quotas of the specified path",
 440                  parameters={
 441                      'fs_id': (str, 'File System Identifier'),
 442                      'path': (str, 'File System Path'),
 443                  },
 444                  responses={200: GET_QUOTAS_SCHEMA})
 445     def get_quota(self, fs_id, path):
 446         """
 447         Get the quotas of the specified path.
 448         :param fs_id: The filesystem identifier.
 449         :param path: The path of the directory/file.
 450         :return: Returns a dictionary containing 'max_bytes'
 451         and 'max_files'.
 452         :rtype: dict
 453         """
 454         cfs = self._cephfs_instance(fs_id)
 455         return cfs.get_quotas(path)
 456
 457     @RESTController.Resource('POST', path='/snapshot')
 458     @allow_empty_body
 459     def snapshot(self, fs_id, path, name=None):
 460         """
 461         Create a snapshot.
 462         :param fs_id: The filesystem identifier.
 463         :param path: The path of the directory.
 464         :param name: The name of the snapshot. If not specified, a name using the
 465         current time in RFC3339 UTC format will be generated.
 466         :return: The name of the snapshot.
 467         :rtype: str
 468         """
 469         cfs = self._cephfs_instance(fs_id)
 470         return cfs.mk_snapshot(path, name)
 471
 472     @RESTController.Resource('DELETE', path='/snapshot')
 473     def rm_snapshot(self, fs_id, path, name):
 474         """
 475         Remove a snapshot.
 476         :param fs_id: The filesystem identifier.
 477         :param path: The path of the directory.
 478         :param name: The name of the snapshot.
 479         """
 480         cfs = self._cephfs_instance(fs_id)
 481         cfs.rm_snapshot(path, name)
 482
 483
 484 class CephFSClients(object):
 485     def __init__(self, module_inst, fscid):
 486         self._module = module_inst
 487         self.fscid = fscid
 488
 489     @ViewCache()
 490     def get(self):
 491         return CephService.send_command('mds', 'session ls', srv_spec='{0}:0'.format(self.fscid))
 492
 493
 494 @UIRouter('/cephfs', Scope.CEPHFS)
 495 @APIDoc("Dashboard UI helper function; not part of the public API", "CephFSUi")
 496 class CephFsUi(CephFS):
 497     RESOURCE_ID = 'fs_id'
 498
 499     @RESTController.Resource('GET')
 500     def tabs(self, fs_id):
 501         data = {}
 502         fs_id = self.fs_id_to_int(fs_id)
 503
 504         # Needed for detail tab
 505         fs_status = self.fs_status(fs_id)
 506         for pool in fs_status['cephfs']['pools']:
 507             pool['size'] = pool['used'] + pool['avail']
 508         data['pools'] = fs_status['cephfs']['pools']
 509         data['ranks'] = fs_status['cephfs']['ranks']
 510         data['name'] = fs_status['cephfs']['name']
 511         data['standbys'] = ', '.join([x['name'] for x in fs_status['standbys']])
 512         counters = self._mds_counters(fs_id)
 513         for k, v in counters.items():
 514             v['name'] = k
 515         data['mds_counters'] = counters
 516
 517         # Needed for client tab
 518         data['clients'] = self._clients(fs_id)
 519
 520         return data
 521
 522     @handle_cephfs_error()
 523     @RESTController.Resource('GET')
 524     def ls_dir(self, fs_id, path=None, depth=1):
 525         """
 526         The difference to the API version is that the root directory will be send when listing
 527         the root directory.
 528         To only do one request this endpoint was created.
 529         :param fs_id: The filesystem identifier.
 530         :type fs_id: int | str
 531         :param path: The path where to start listing the directory content.
 532         Defaults to '/' if not set.
 533         :type path: str | bytes
 534         :param depth: The number of steps to go down the directory tree.
 535         :type depth: int | str
 536         :return: The names of the directories below the specified path.
 537         :rtype: list
 538         """
 539         path = self._set_ls_dir_path(path)
 540         try:
 541             cfs = self._cephfs_instance(fs_id)
 542             paths = cfs.ls_dir(path, depth)
 543             if path == os.sep:
 544                 paths = [self._get_root_directory(cfs)] + paths
 545         except (cephfs.PermissionError, cephfs.ObjectNotFound):  # pragma: no cover
 546             paths = []
 547         return paths