ceph/qa/tasks/cephfs/filesystem.py

   1
   2 from StringIO import StringIO
   3 import json
   4 import logging
   5 from gevent import Greenlet
   6 import os
   7 import time
   8 import datetime
   9 import re
  10 import errno
  11
  12 from teuthology.exceptions import CommandFailedError
  13 from teuthology import misc
  14 from teuthology.nuke import clear_firewall
  15 from teuthology.parallel import parallel
  16 from tasks.ceph_manager import write_conf
  17 from tasks import ceph_manager
  18
  19
  20 log = logging.getLogger(__name__)
  21
  22
  23 DAEMON_WAIT_TIMEOUT = 120
  24 ROOT_INO = 1
  25
  26
  27 class ObjectNotFound(Exception):
  28     def __init__(self, object_name):
  29         self._object_name = object_name
  30
  31     def __str__(self):
  32         return "Object not found: '{0}'".format(self._object_name)
  33
  34 class FSStatus(object):
  35     """
  36     Operations on a snapshot of the FSMap.
  37     """
  38     def __init__(self, mon_manager):
  39         self.mon = mon_manager
  40         self.map = json.loads(self.mon.raw_cluster_cmd("fs", "dump", "--format=json"))
  41
  42     def __str__(self):
  43         return json.dumps(self.map, indent = 2, sort_keys = True)
  44
  45     # Expose the fsmap for manual inspection.
  46     def __getitem__(self, key):
  47         """
  48         Get a field from the fsmap.
  49         """
  50         return self.map[key]
  51
  52     def get_filesystems(self):
  53         """
  54         Iterator for all filesystems.
  55         """
  56         for fs in self.map['filesystems']:
  57             yield fs
  58
  59     def get_all(self):
  60         """
  61         Iterator for all the mds_info components in the FSMap.
  62         """
  63         for info in self.get_standbys():
  64             yield info
  65         for fs in self.map['filesystems']:
  66             for info in fs['mdsmap']['info'].values():
  67                 yield info
  68
  69     def get_standbys(self):
  70         """
  71         Iterator for all standbys.
  72         """
  73         for info in self.map['standbys']:
  74             yield info
  75
  76     def get_fsmap(self, fscid):
  77         """
  78         Get the fsmap for the given FSCID.
  79         """
  80         for fs in self.map['filesystems']:
  81             if fscid is None or fs['id'] == fscid:
  82                 return fs
  83         raise RuntimeError("FSCID {0} not in map".format(fscid))
  84
  85     def get_fsmap_byname(self, name):
  86         """
  87         Get the fsmap for the given file system name.
  88         """
  89         for fs in self.map['filesystems']:
  90             if name is None or fs['mdsmap']['fs_name'] == name:
  91                 return fs
  92         raise RuntimeError("FS {0} not in map".format(name))
  93
  94     def get_replays(self, fscid):
  95         """
  96         Get the standby:replay MDS for the given FSCID.
  97         """
  98         fs = self.get_fsmap(fscid)
  99         for info in fs['mdsmap']['info'].values():
 100             if info['state'] == 'up:standby-replay':
 101                 yield info
 102
 103     def get_ranks(self, fscid):
 104         """
 105         Get the ranks for the given FSCID.
 106         """
 107         fs = self.get_fsmap(fscid)
 108         for info in fs['mdsmap']['info'].values():
 109             if info['rank'] >= 0:
 110                 yield info
 111
 112     def get_rank(self, fscid, rank):
 113         """
 114         Get the rank for the given FSCID.
 115         """
 116         for info in self.get_ranks(fscid):
 117             if info['rank'] == rank:
 118                 return info
 119         raise RuntimeError("FSCID {0} has no rank {1}".format(fscid, rank))
 120
 121     def get_mds(self, name):
 122         """
 123         Get the info for the given MDS name.
 124         """
 125         for info in self.get_all():
 126             if info['name'] == name:
 127                 return info
 128         return None
 129
 130     def get_mds_addr(self, name):
 131         """
 132         Return the instance addr as a string, like "10.214.133.138:6807\/10825"
 133         """
 134         info = self.get_mds(name)
 135         if info:
 136             return info['addr']
 137         else:
 138             log.warn(json.dumps(list(self.get_all()), indent=2))  # dump for debugging
 139             raise RuntimeError("MDS id '{0}' not found in map".format(name))
 140
 141 class CephCluster(object):
 142     @property
 143     def admin_remote(self):
 144         first_mon = misc.get_first_mon(self._ctx, None)
 145         (result,) = self._ctx.cluster.only(first_mon).remotes.iterkeys()
 146         return result
 147
 148     def __init__(self, ctx):
 149         self._ctx = ctx
 150         self.mon_manager = ceph_manager.CephManager(self.admin_remote, ctx=ctx, logger=log.getChild('ceph_manager'))
 151
 152     def get_config(self, key, service_type=None):
 153         """
 154         Get config from mon by default, or a specific service if caller asks for it
 155         """
 156         if service_type is None:
 157             service_type = 'mon'
 158
 159         service_id = sorted(misc.all_roles_of_type(self._ctx.cluster, service_type))[0]
 160         return self.json_asok(['config', 'get', key], service_type, service_id)[key]
 161
 162     def set_ceph_conf(self, subsys, key, value):
 163         if subsys not in self._ctx.ceph['ceph'].conf:
 164             self._ctx.ceph['ceph'].conf[subsys] = {}
 165         self._ctx.ceph['ceph'].conf[subsys][key] = value
 166         write_conf(self._ctx)  # XXX because we don't have the ceph task's config object, if they
 167                                # used a different config path this won't work.
 168
 169     def clear_ceph_conf(self, subsys, key):
 170         del self._ctx.ceph['ceph'].conf[subsys][key]
 171         write_conf(self._ctx)
 172
 173     def json_asok(self, command, service_type, service_id):
 174         proc = self.mon_manager.admin_socket(service_type, service_id, command)
 175         response_data = proc.stdout.getvalue()
 176         log.info("_json_asok output: {0}".format(response_data))
 177         if response_data.strip():
 178             return json.loads(response_data)
 179         else:
 180             return None
 181
 182
 183 class MDSCluster(CephCluster):
 184     """
 185     Collective operations on all the MDS daemons in the Ceph cluster.  These
 186     daemons may be in use by various Filesystems.
 187
 188     For the benefit of pre-multi-filesystem tests, this class is also
 189     a parent of Filesystem.  The correct way to use MDSCluster going forward is
 190     as a separate instance outside of your (multiple) Filesystem instances.
 191     """
 192     def __init__(self, ctx):
 193         super(MDSCluster, self).__init__(ctx)
 194
 195         self.mds_ids = list(misc.all_roles_of_type(ctx.cluster, 'mds'))
 196
 197         if len(self.mds_ids) == 0:
 198             raise RuntimeError("This task requires at least one MDS")
 199
 200         if hasattr(self._ctx, "daemons"):
 201             # Presence of 'daemons' attribute implies ceph task rather than ceph_deploy task
 202             self.mds_daemons = dict([(mds_id, self._ctx.daemons.get_daemon('mds', mds_id)) for mds_id in self.mds_ids])
 203
 204     def _one_or_all(self, mds_id, cb, in_parallel=True):
 205         """
 206         Call a callback for a single named MDS, or for all.
 207
 208         Note that the parallelism here isn't for performance, it's to avoid being overly kind
 209         to the cluster by waiting a graceful ssh-latency of time between doing things, and to
 210         avoid being overly kind by executing them in a particular order.  However, some actions
 211         don't cope with being done in parallel, so it's optional (`in_parallel`)
 212
 213         :param mds_id: MDS daemon name, or None
 214         :param cb: Callback taking single argument of MDS daemon name
 215         :param in_parallel: whether to invoke callbacks concurrently (else one after the other)
 216         """
 217         if mds_id is None:
 218             if in_parallel:
 219                 with parallel() as p:
 220                     for mds_id in self.mds_ids:
 221                         p.spawn(cb, mds_id)
 222             else:
 223                 for mds_id in self.mds_ids:
 224                     cb(mds_id)
 225         else:
 226             cb(mds_id)
 227
 228     def mds_stop(self, mds_id=None):
 229         """
 230         Stop the MDS daemon process(se).  If it held a rank, that rank
 231         will eventually go laggy.
 232         """
 233         self._one_or_all(mds_id, lambda id_: self.mds_daemons[id_].stop())
 234
 235     def mds_fail(self, mds_id=None):
 236         """
 237         Inform MDSMonitor of the death of the daemon process(es).  If it held
 238         a rank, that rank will be relinquished.
 239         """
 240         self._one_or_all(mds_id, lambda id_: self.mon_manager.raw_cluster_cmd("mds", "fail", id_))
 241
 242     def mds_restart(self, mds_id=None):
 243         self._one_or_all(mds_id, lambda id_: self.mds_daemons[id_].restart())
 244
 245     def mds_fail_restart(self, mds_id=None):
 246         """
 247         Variation on restart that includes marking MDSs as failed, so that doing this
 248         operation followed by waiting for healthy daemon states guarantees that they
 249         have gone down and come up, rather than potentially seeing the healthy states
 250         that existed before the restart.
 251         """
 252         def _fail_restart(id_):
 253             self.mds_daemons[id_].stop()
 254             self.mon_manager.raw_cluster_cmd("mds", "fail", id_)
 255             self.mds_daemons[id_].restart()
 256
 257         self._one_or_all(mds_id, _fail_restart)
 258
 259     def newfs(self, name):
 260         return Filesystem(self._ctx, create=name)
 261
 262     def status(self):
 263         return FSStatus(self.mon_manager)
 264
 265     def delete_all_filesystems(self):
 266         """
 267         Remove all filesystems that exist, and any pools in use by them.
 268         """
 269         pools = json.loads(self.mon_manager.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools']
 270         pool_id_name = {}
 271         for pool in pools:
 272             pool_id_name[pool['pool']] = pool['pool_name']
 273
 274         # mark cluster down for each fs to prevent churn during deletion
 275         status = self.status()
 276         for fs in status.get_filesystems():
 277             self.mon_manager.raw_cluster_cmd("fs", "set", fs['mdsmap']['fs_name'], "cluster_down", "true")
 278
 279         # get a new copy as actives may have since changed
 280         status = self.status()
 281         for fs in status.get_filesystems():
 282             mdsmap = fs['mdsmap']
 283             metadata_pool = pool_id_name[mdsmap['metadata_pool']]
 284
 285             for gid in mdsmap['up'].values():
 286                 self.mon_manager.raw_cluster_cmd('mds', 'fail', gid.__str__())
 287
 288             self.mon_manager.raw_cluster_cmd('fs', 'rm', mdsmap['fs_name'], '--yes-i-really-mean-it')
 289             self.mon_manager.raw_cluster_cmd('osd', 'pool', 'delete',
 290                                              metadata_pool, metadata_pool,
 291                                              '--yes-i-really-really-mean-it')
 292             for data_pool in mdsmap['data_pools']:
 293                 data_pool = pool_id_name[data_pool]
 294                 try:
 295                     self.mon_manager.raw_cluster_cmd('osd', 'pool', 'delete',
 296                                                      data_pool, data_pool,
 297                                                      '--yes-i-really-really-mean-it')
 298                 except CommandFailedError as e:
 299                     if e.exitstatus == 16: # EBUSY, this data pool is used
 300                         pass               # by two metadata pools, let the 2nd
 301                     else:                  # pass delete it
 302                         raise
 303
 304     def get_standby_daemons(self):
 305         return set([s['name'] for s in self.status().get_standbys()])
 306
 307     def get_mds_hostnames(self):
 308         result = set()
 309         for mds_id in self.mds_ids:
 310             mds_remote = self.mon_manager.find_remote('mds', mds_id)
 311             result.add(mds_remote.hostname)
 312
 313         return list(result)
 314
 315     def set_clients_block(self, blocked, mds_id=None):
 316         """
 317         Block (using iptables) client communications to this MDS.  Be careful: if
 318         other services are running on this MDS, or other MDSs try to talk to this
 319         MDS, their communications may also be blocked as collatoral damage.
 320
 321         :param mds_id: Optional ID of MDS to block, default to all
 322         :return:
 323         """
 324         da_flag = "-A" if blocked else "-D"
 325
 326         def set_block(_mds_id):
 327             remote = self.mon_manager.find_remote('mds', _mds_id)
 328             status = self.status()
 329
 330             addr = status.get_mds_addr(_mds_id)
 331             ip_str, port_str, inst_str = re.match("(.+):(.+)/(.+)", addr).groups()
 332
 333             remote.run(
 334                 args=["sudo", "iptables", da_flag, "OUTPUT", "-p", "tcp", "--sport", port_str, "-j", "REJECT", "-m",
 335                       "comment", "--comment", "teuthology"])
 336             remote.run(
 337                 args=["sudo", "iptables", da_flag, "INPUT", "-p", "tcp", "--dport", port_str, "-j", "REJECT", "-m",
 338                       "comment", "--comment", "teuthology"])
 339
 340         self._one_or_all(mds_id, set_block, in_parallel=False)
 341
 342     def clear_firewall(self):
 343         clear_firewall(self._ctx)
 344
 345     def get_mds_info(self, mds_id):
 346         return FSStatus(self.mon_manager).get_mds(mds_id)
 347
 348     def is_full(self):
 349         flags = json.loads(self.mon_manager.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['flags']
 350         return 'full' in flags
 351
 352     def is_pool_full(self, pool_name):
 353         pools = json.loads(self.mon_manager.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools']
 354         for pool in pools:
 355             if pool['pool_name'] == pool_name:
 356                 return 'full' in pool['flags_names'].split(",")
 357
 358         raise RuntimeError("Pool not found '{0}'".format(pool_name))
 359
 360 class Filesystem(MDSCluster):
 361     """
 362     This object is for driving a CephFS filesystem.  The MDS daemons driven by
 363     MDSCluster may be shared with other Filesystems.
 364     """
 365     def __init__(self, ctx, fscid=None, create=None):
 366         super(Filesystem, self).__init__(ctx)
 367
 368         self.id = None
 369         self.name = None
 370         self.metadata_pool_name = None
 371         self.data_pools = None
 372
 373         client_list = list(misc.all_roles_of_type(self._ctx.cluster, 'client'))
 374         self.client_id = client_list[0]
 375         self.client_remote = list(misc.get_clients(ctx=ctx, roles=["client.{0}".format(self.client_id)]))[0][1]
 376
 377         if create is not None:
 378             if fscid is not None:
 379                 raise RuntimeError("cannot specify fscid when creating fs")
 380             if create is True:
 381                 self.name = 'cephfs'
 382             else:
 383                 self.name = create
 384             if not self.legacy_configured():
 385                 self.create()
 386         elif fscid is not None:
 387             self.id = fscid
 388         self.getinfo(refresh = True)
 389
 390         # Stash a reference to the first created filesystem on ctx, so
 391         # that if someone drops to the interactive shell they can easily
 392         # poke our methods.
 393         if not hasattr(self._ctx, "filesystem"):
 394             self._ctx.filesystem = self
 395
 396     def getinfo(self, refresh = False):
 397         status = self.status()
 398         if self.id is not None:
 399             fsmap = status.get_fsmap(self.id)
 400         elif self.name is not None:
 401             fsmap = status.get_fsmap_byname(self.name)
 402         else:
 403             fss = [fs for fs in status.get_filesystems()]
 404             if len(fss) == 1:
 405                 fsmap = fss[0]
 406             elif len(fss) == 0:
 407                 raise RuntimeError("no file system available")
 408             else:
 409                 raise RuntimeError("more than one file system available")
 410         self.id = fsmap['id']
 411         self.name = fsmap['mdsmap']['fs_name']
 412         self.get_pool_names(status = status, refresh = refresh)
 413         return status
 414
 415     def deactivate(self, rank):
 416         if rank < 0:
 417             raise RuntimeError("invalid rank")
 418         elif rank == 0:
 419             raise RuntimeError("cannot deactivate rank 0")
 420         self.mon_manager.raw_cluster_cmd("mds", "deactivate", "%d:%d" % (self.id, rank))
 421
 422     def set_max_mds(self, max_mds):
 423         self.mon_manager.raw_cluster_cmd("fs", "set", self.name, "max_mds", "%d" % max_mds)
 424
 425     def set_allow_dirfrags(self, yes):
 426         self.mon_manager.raw_cluster_cmd("fs", "set", self.name, "allow_dirfrags", str(yes).lower(), '--yes-i-really-mean-it')
 427
 428     def get_pgs_per_fs_pool(self):
 429         """
 430         Calculate how many PGs to use when creating a pool, in order to avoid raising any
 431         health warnings about mon_pg_warn_min_per_osd
 432
 433         :return: an integer number of PGs
 434         """
 435         pg_warn_min_per_osd = int(self.get_config('mon_pg_warn_min_per_osd'))
 436         osd_count = len(list(misc.all_roles_of_type(self._ctx.cluster, 'osd')))
 437         return pg_warn_min_per_osd * osd_count
 438
 439     def create(self):
 440         if self.name is None:
 441             self.name = "cephfs"
 442         if self.metadata_pool_name is None:
 443             self.metadata_pool_name = "{0}_metadata".format(self.name)
 444         data_pool_name = "{0}_data".format(self.name)
 445
 446         log.info("Creating filesystem '{0}'".format(self.name))
 447
 448         pgs_per_fs_pool = self.get_pgs_per_fs_pool()
 449
 450         self.mon_manager.raw_cluster_cmd('osd', 'pool', 'create',
 451                                          self.metadata_pool_name, pgs_per_fs_pool.__str__())
 452         self.mon_manager.raw_cluster_cmd('osd', 'pool', 'create',
 453                                          data_pool_name, pgs_per_fs_pool.__str__())
 454         self.mon_manager.raw_cluster_cmd('fs', 'new',
 455                                          self.name, self.metadata_pool_name, data_pool_name)
 456         # Turn off spurious standby count warnings from modifying max_mds in tests.
 457         try:
 458             self.mon_manager.raw_cluster_cmd('fs', 'set', self.name, 'standby_count_wanted', '0')
 459         except CommandFailedError as e:
 460             if e.exitstatus == 22:
 461                 # standby_count_wanted not available prior to luminous (upgrade tests would fail otherwise)
 462                 pass
 463             else:
 464                 raise
 465
 466         self.getinfo(refresh = True)
 467
 468     def __del__(self):
 469         if getattr(self._ctx, "filesystem", None) == self:
 470             delattr(self._ctx, "filesystem")
 471
 472     def exists(self):
 473         """
 474         Whether a filesystem exists in the mon's filesystem list
 475         """
 476         fs_list = json.loads(self.mon_manager.raw_cluster_cmd('fs', 'ls', '--format=json-pretty'))
 477         return self.name in [fs['name'] for fs in fs_list]
 478
 479     def legacy_configured(self):
 480         """
 481         Check if a legacy (i.e. pre "fs new") filesystem configuration is present.  If this is
 482         the case, the caller should avoid using Filesystem.create
 483         """
 484         try:
 485             out_text = self.mon_manager.raw_cluster_cmd('--format=json-pretty', 'osd', 'lspools')
 486             pools = json.loads(out_text)
 487             metadata_pool_exists = 'metadata' in [p['poolname'] for p in pools]
 488             if metadata_pool_exists:
 489                 self.metadata_pool_name = 'metadata'
 490         except CommandFailedError as e:
 491             # For use in upgrade tests, Ceph cuttlefish and earlier don't support
 492             # structured output (--format) from the CLI.
 493             if e.exitstatus == 22:
 494                 metadata_pool_exists = True
 495             else:
 496                 raise
 497
 498         return metadata_pool_exists
 499
 500     def _df(self):
 501         return json.loads(self.mon_manager.raw_cluster_cmd("df", "--format=json-pretty"))
 502
 503     def get_mds_map(self):
 504         return self.status().get_fsmap(self.id)['mdsmap']
 505
 506     def add_data_pool(self, name):
 507         self.mon_manager.raw_cluster_cmd('osd', 'pool', 'create', name, self.get_pgs_per_fs_pool().__str__())
 508         self.mon_manager.raw_cluster_cmd('fs', 'add_data_pool', self.name, name)
 509         self.get_pool_names(refresh = True)
 510         for poolid, fs_name in self.data_pools.items():
 511             if name == fs_name:
 512                 return poolid
 513         raise RuntimeError("could not get just created pool '{0}'".format(name))
 514
 515     def get_pool_names(self, refresh = False, status = None):
 516         if refresh or self.metadata_pool_name is None or self.data_pools is None:
 517             if status is None:
 518                 status = self.status()
 519             fsmap = status.get_fsmap(self.id)
 520
 521             osd_map = self.mon_manager.get_osd_dump_json()
 522             id_to_name = {}
 523             for p in osd_map['pools']:
 524                 id_to_name[p['pool']] = p['pool_name']
 525
 526             self.metadata_pool_name = id_to_name[fsmap['mdsmap']['metadata_pool']]
 527             self.data_pools = {}
 528             for data_pool in fsmap['mdsmap']['data_pools']:
 529                 self.data_pools[data_pool] = id_to_name[data_pool]
 530
 531     def get_data_pool_name(self, refresh = False):
 532         if refresh or self.data_pools is None:
 533             self.get_pool_names(refresh = True)
 534         assert(len(self.data_pools) == 1)
 535         return self.data_pools.values()[0]
 536
 537     def get_data_pool_id(self, refresh = False):
 538         """
 539         Don't call this if you have multiple data pools
 540         :return: integer
 541         """
 542         if refresh or self.data_pools is None:
 543             self.get_pool_names(refresh = True)
 544         assert(len(self.data_pools) == 1)
 545         return self.data_pools.keys()[0]
 546
 547     def get_data_pool_names(self, refresh = False):
 548         if refresh or self.data_pools is None:
 549             self.get_pool_names(refresh = True)
 550         return self.data_pools.values()
 551
 552     def get_metadata_pool_name(self):
 553         return self.metadata_pool_name
 554
 555     def get_namespace_id(self):
 556         return self.id
 557
 558     def get_pool_df(self, pool_name):
 559         """
 560         Return a dict like:
 561         {u'bytes_used': 0, u'max_avail': 83848701, u'objects': 0, u'kb_used': 0}
 562         """
 563         for pool_df in self._df()['pools']:
 564             if pool_df['name'] == pool_name:
 565                 return pool_df['stats']
 566
 567         raise RuntimeError("Pool name '{0}' not found".format(pool_name))
 568
 569     def get_usage(self):
 570         return self._df()['stats']['total_used_bytes']
 571
 572     def are_daemons_healthy(self):
 573         """
 574         Return true if all daemons are in one of active, standby, standby-replay, and
 575         at least max_mds daemons are in 'active'.
 576
 577         Unlike most of Filesystem, this function is tolerant of new-style `fs`
 578         commands being missing, because we are part of the ceph installation
 579         process during upgrade suites, so must fall back to old style commands
 580         when we get an EINVAL on a new style command.
 581
 582         :return:
 583         """
 584
 585         active_count = 0
 586         try:
 587             mds_map = self.get_mds_map()
 588         except CommandFailedError as cfe:
 589             # Old version, fall back to non-multi-fs commands
 590             if cfe.exitstatus == errno.EINVAL:
 591                 mds_map = json.loads(
 592                         self.mon_manager.raw_cluster_cmd('mds', 'dump', '--format=json'))
 593             else:
 594                 raise
 595
 596         log.info("are_daemons_healthy: mds map: {0}".format(mds_map))
 597
 598         for mds_id, mds_status in mds_map['info'].items():
 599             if mds_status['state'] not in ["up:active", "up:standby", "up:standby-replay"]:
 600                 log.warning("Unhealthy mds state {0}:{1}".format(mds_id, mds_status['state']))
 601                 return False
 602             elif mds_status['state'] == 'up:active':
 603                 active_count += 1
 604
 605         log.info("are_daemons_healthy: {0}/{1}".format(
 606             active_count, mds_map['max_mds']
 607         ))
 608
 609         if active_count >= mds_map['max_mds']:
 610             # The MDSMap says these guys are active, but let's check they really are
 611             for mds_id, mds_status in mds_map['info'].items():
 612                 if mds_status['state'] == 'up:active':
 613                     try:
 614                         daemon_status = self.mds_asok(["status"], mds_id=mds_status['name'])
 615                     except CommandFailedError as cfe:
 616                         if cfe.exitstatus == errno.EINVAL:
 617                             # Old version, can't do this check
 618                             continue
 619                         else:
 620                             # MDS not even running
 621                             return False
 622
 623                     if daemon_status['state'] != 'up:active':
 624                         # MDS hasn't taken the latest map yet
 625                         return False
 626
 627             return True
 628         else:
 629             return False
 630
 631     def get_daemon_names(self, state=None):
 632         """
 633         Return MDS daemon names of those daemons in the given state
 634         :param state:
 635         :return:
 636         """
 637         status = self.get_mds_map()
 638         result = []
 639         for mds_status in sorted(status['info'].values(), lambda a, b: cmp(a['rank'], b['rank'])):
 640             if mds_status['state'] == state or state is None:
 641                 result.append(mds_status['name'])
 642
 643         return result
 644
 645     def get_active_names(self):
 646         """
 647         Return MDS daemon names of those daemons holding ranks
 648         in state up:active
 649
 650         :return: list of strings like ['a', 'b'], sorted by rank
 651         """
 652         return self.get_daemon_names("up:active")
 653
 654     def get_all_mds_rank(self):
 655         status = self.get_mds_map()
 656         result = []
 657         for mds_status in sorted(status['info'].values(), lambda a, b: cmp(a['rank'], b['rank'])):
 658             if mds_status['rank'] != -1 and mds_status['state'] != 'up:standby-replay':
 659                 result.append(mds_status['rank'])
 660
 661         return result
 662
 663     def get_rank_names(self):
 664         """
 665         Return MDS daemon names of those daemons holding a rank,
 666         sorted by rank.  This includes e.g. up:replay/reconnect
 667         as well as active, but does not include standby or
 668         standby-replay.
 669         """
 670         status = self.get_mds_map()
 671         result = []
 672         for mds_status in sorted(status['info'].values(), lambda a, b: cmp(a['rank'], b['rank'])):
 673             if mds_status['rank'] != -1 and mds_status['state'] != 'up:standby-replay':
 674                 result.append(mds_status['name'])
 675
 676         return result
 677
 678     def wait_for_daemons(self, timeout=None):
 679         """
 680         Wait until all daemons are healthy
 681         :return:
 682         """
 683
 684         if timeout is None:
 685             timeout = DAEMON_WAIT_TIMEOUT
 686
 687         elapsed = 0
 688         while True:
 689             if self.are_daemons_healthy():
 690                 return
 691             else:
 692                 time.sleep(1)
 693                 elapsed += 1
 694
 695             if elapsed > timeout:
 696                 raise RuntimeError("Timed out waiting for MDS daemons to become healthy")
 697
 698     def get_lone_mds_id(self):
 699         """
 700         Get a single MDS ID: the only one if there is only one
 701         configured, else the only one currently holding a rank,
 702         else raise an error.
 703         """
 704         if len(self.mds_ids) != 1:
 705             alive = self.get_rank_names()
 706             if len(alive) == 1:
 707                 return alive[0]
 708             else:
 709                 raise ValueError("Explicit MDS argument required when multiple MDSs in use")
 710         else:
 711             return self.mds_ids[0]
 712
 713     def recreate(self):
 714         log.info("Creating new filesystem")
 715         self.delete_all_filesystems()
 716         self.id = None
 717         self.create()
 718
 719     def put_metadata_object_raw(self, object_id, infile):
 720         """
 721         Save an object to the metadata pool
 722         """
 723         temp_bin_path = infile
 724         self.client_remote.run(args=[
 725             'sudo', os.path.join(self._prefix, 'rados'), '-p', self.metadata_pool_name, 'put', object_id, temp_bin_path
 726         ])
 727
 728     def get_metadata_object_raw(self, object_id):
 729         """
 730         Retrieve an object from the metadata pool and store it in a file.
 731         """
 732         temp_bin_path = '/tmp/' + object_id + '.bin'
 733
 734         self.client_remote.run(args=[
 735             'sudo', os.path.join(self._prefix, 'rados'), '-p', self.metadata_pool_name, 'get', object_id, temp_bin_path
 736         ])
 737
 738         return temp_bin_path
 739
 740     def get_metadata_object(self, object_type, object_id):
 741         """
 742         Retrieve an object from the metadata pool, pass it through
 743         ceph-dencoder to dump it to JSON, and return the decoded object.
 744         """
 745         temp_bin_path = '/tmp/out.bin'
 746
 747         self.client_remote.run(args=[
 748             'sudo', os.path.join(self._prefix, 'rados'), '-p', self.metadata_pool_name, 'get', object_id, temp_bin_path
 749         ])
 750
 751         stdout = StringIO()
 752         self.client_remote.run(args=[
 753             'sudo', os.path.join(self._prefix, 'ceph-dencoder'), 'type', object_type, 'import', temp_bin_path, 'decode', 'dump_json'
 754         ], stdout=stdout)
 755         dump_json = stdout.getvalue().strip()
 756         try:
 757             dump = json.loads(dump_json)
 758         except (TypeError, ValueError):
 759             log.error("Failed to decode JSON: '{0}'".format(dump_json))
 760             raise
 761
 762         return dump
 763
 764     def get_journal_version(self):
 765         """
 766         Read the JournalPointer and Journal::Header objects to learn the version of
 767         encoding in use.
 768         """
 769         journal_pointer_object = '400.00000000'
 770         journal_pointer_dump = self.get_metadata_object("JournalPointer", journal_pointer_object)
 771         journal_ino = journal_pointer_dump['journal_pointer']['front']
 772
 773         journal_header_object = "{0:x}.00000000".format(journal_ino)
 774         journal_header_dump = self.get_metadata_object('Journaler::Header', journal_header_object)
 775
 776         version = journal_header_dump['journal_header']['stream_format']
 777         log.info("Read journal version {0}".format(version))
 778
 779         return version
 780
 781     def mds_asok(self, command, mds_id=None):
 782         if mds_id is None:
 783             mds_id = self.get_lone_mds_id()
 784
 785         return self.json_asok(command, 'mds', mds_id)
 786
 787     def read_cache(self, path, depth=None):
 788         cmd = ["dump", "tree", path]
 789         if depth is not None:
 790             cmd.append(depth.__str__())
 791         result = self.mds_asok(cmd)
 792         if len(result) == 0:
 793             raise RuntimeError("Path not found in cache: {0}".format(path))
 794
 795         return result
 796
 797     def wait_for_state(self, goal_state, reject=None, timeout=None, mds_id=None, rank=None):
 798         """
 799         Block until the MDS reaches a particular state, or a failure condition
 800         is met.
 801
 802         When there are multiple MDSs, succeed when exaclty one MDS is in the
 803         goal state, or fail when any MDS is in the reject state.
 804
 805         :param goal_state: Return once the MDS is in this state
 806         :param reject: Fail if the MDS enters this state before the goal state
 807         :param timeout: Fail if this many seconds pass before reaching goal
 808         :return: number of seconds waited, rounded down to integer
 809         """
 810
 811         started_at = time.time()
 812         while True:
 813             status = self.status()
 814             if rank is not None:
 815                 mds_info = status.get_rank(self.id, rank)
 816                 current_state = mds_info['state'] if mds_info else None
 817                 log.info("Looked up MDS state for mds.{0}: {1}".format(rank, current_state))
 818             elif mds_id is not None:
 819                 # mds_info is None if no daemon with this ID exists in the map
 820                 mds_info = status.get_mds(mds_id)
 821                 current_state = mds_info['state'] if mds_info else None
 822                 log.info("Looked up MDS state for {0}: {1}".format(mds_id, current_state))
 823             else:
 824                 # In general, look for a single MDS
 825                 states = [m['state'] for m in status.get_ranks(self.id)]
 826                 if [s for s in states if s == goal_state] == [goal_state]:
 827                     current_state = goal_state
 828                 elif reject in states:
 829                     current_state = reject
 830                 else:
 831                     current_state = None
 832                 log.info("mapped states {0} to {1}".format(states, current_state))
 833
 834             elapsed = time.time() - started_at
 835             if current_state == goal_state:
 836                 log.info("reached state '{0}' in {1}s".format(current_state, elapsed))
 837                 return elapsed
 838             elif reject is not None and current_state == reject:
 839                 raise RuntimeError("MDS in reject state {0}".format(current_state))
 840             elif timeout is not None and elapsed > timeout:
 841                 log.error("MDS status at timeout: {0}".format(status.get_fsmap(self.id)))
 842                 raise RuntimeError(
 843                     "Reached timeout after {0} seconds waiting for state {1}, while in state {2}".format(
 844                         elapsed, goal_state, current_state
 845                     ))
 846             else:
 847                 time.sleep(1)
 848
 849     def _read_data_xattr(self, ino_no, xattr_name, type, pool):
 850         mds_id = self.mds_ids[0]
 851         remote = self.mds_daemons[mds_id].remote
 852         if pool is None:
 853             pool = self.get_data_pool_name()
 854
 855         obj_name = "{0:x}.00000000".format(ino_no)
 856
 857         args = [
 858             os.path.join(self._prefix, "rados"), "-p", pool, "getxattr", obj_name, xattr_name
 859         ]
 860         try:
 861             proc = remote.run(
 862                 args=args,
 863                 stdout=StringIO())
 864         except CommandFailedError as e:
 865             log.error(e.__str__())
 866             raise ObjectNotFound(obj_name)
 867
 868         data = proc.stdout.getvalue()
 869
 870         p = remote.run(
 871             args=[os.path.join(self._prefix, "ceph-dencoder"), "type", type, "import", "-", "decode", "dump_json"],
 872             stdout=StringIO(),
 873             stdin=data
 874         )
 875
 876         return json.loads(p.stdout.getvalue().strip())
 877
 878     def _write_data_xattr(self, ino_no, xattr_name, data, pool=None):
 879         """
 880         Write to an xattr of the 0th data object of an inode.  Will
 881         succeed whether the object and/or xattr already exist or not.
 882
 883         :param ino_no: integer inode number
 884         :param xattr_name: string name of the xattr
 885         :param data: byte array data to write to the xattr
 886         :param pool: name of data pool or None to use primary data pool
 887         :return: None
 888         """
 889         remote = self.mds_daemons[self.mds_ids[0]].remote
 890         if pool is None:
 891             pool = self.get_data_pool_name()
 892
 893         obj_name = "{0:x}.00000000".format(ino_no)
 894         args = [
 895             os.path.join(self._prefix, "rados"), "-p", pool, "setxattr",
 896             obj_name, xattr_name, data
 897         ]
 898         remote.run(
 899             args=args,
 900             stdout=StringIO())
 901
 902     def read_backtrace(self, ino_no, pool=None):
 903         """
 904         Read the backtrace from the data pool, return a dict in the format
 905         given by inode_backtrace_t::dump, which is something like:
 906
 907         ::
 908
 909             rados -p cephfs_data getxattr 10000000002.00000000 parent > out.bin
 910             ceph-dencoder type inode_backtrace_t import out.bin decode dump_json
 911
 912             { "ino": 1099511627778,
 913               "ancestors": [
 914                     { "dirino": 1,
 915                       "dname": "blah",
 916                       "version": 11}],
 917               "pool": 1,
 918               "old_pools": []}
 919
 920         :param pool: name of pool to read backtrace from.  If omitted, FS must have only
 921                      one data pool and that will be used.
 922         """
 923         return self._read_data_xattr(ino_no, "parent", "inode_backtrace_t", pool)
 924
 925     def read_layout(self, ino_no, pool=None):
 926         """
 927         Read 'layout' xattr of an inode and parse the result, returning a dict like:
 928         ::
 929             {
 930                 "stripe_unit": 4194304,
 931                 "stripe_count": 1,
 932                 "object_size": 4194304,
 933                 "pool_id": 1,
 934                 "pool_ns": "",
 935             }
 936
 937         :param pool: name of pool to read backtrace from.  If omitted, FS must have only
 938                      one data pool and that will be used.
 939         """
 940         return self._read_data_xattr(ino_no, "layout", "file_layout_t", pool)
 941
 942     def _enumerate_data_objects(self, ino, size):
 943         """
 944         Get the list of expected data objects for a range, and the list of objects
 945         that really exist.
 946
 947         :return a tuple of two lists of strings (expected, actual)
 948         """
 949         stripe_size = 1024 * 1024 * 4
 950
 951         size = max(stripe_size, size)
 952
 953         want_objects = [
 954             "{0:x}.{1:08x}".format(ino, n)
 955             for n in range(0, ((size - 1) / stripe_size) + 1)
 956         ]
 957
 958         exist_objects = self.rados(["ls"], pool=self.get_data_pool_name()).split("\n")
 959
 960         return want_objects, exist_objects
 961
 962     def data_objects_present(self, ino, size):
 963         """
 964         Check that *all* the expected data objects for an inode are present in the data pool
 965         """
 966
 967         want_objects, exist_objects = self._enumerate_data_objects(ino, size)
 968         missing = set(want_objects) - set(exist_objects)
 969
 970         if missing:
 971             log.info("Objects missing (ino {0}, size {1}): {2}".format(
 972                 ino, size, missing
 973             ))
 974             return False
 975         else:
 976             log.info("All objects for ino {0} size {1} found".format(ino, size))
 977             return True
 978
 979     def data_objects_absent(self, ino, size):
 980         want_objects, exist_objects = self._enumerate_data_objects(ino, size)
 981         present = set(want_objects) & set(exist_objects)
 982
 983         if present:
 984             log.info("Objects not absent (ino {0}, size {1}): {2}".format(
 985                 ino, size, present
 986             ))
 987             return False
 988         else:
 989             log.info("All objects for ino {0} size {1} are absent".format(ino, size))
 990             return True
 991
 992     def dirfrag_exists(self, ino, frag):
 993         try:
 994             self.rados(["stat", "{0:x}.{1:08x}".format(ino, frag)])
 995         except CommandFailedError as e:
 996             return False
 997         else:
 998             return True
 999
1000     def rados(self, args, pool=None, namespace=None, stdin_data=None):
1001         """
1002         Call into the `rados` CLI from an MDS
1003         """
1004
1005         if pool is None:
1006             pool = self.get_metadata_pool_name()
1007
1008         # Doesn't matter which MDS we use to run rados commands, they all
1009         # have access to the pools
1010         mds_id = self.mds_ids[0]
1011         remote = self.mds_daemons[mds_id].remote
1012
1013         # NB we could alternatively use librados pybindings for this, but it's a one-liner
1014         # using the `rados` CLI
1015         args = ([os.path.join(self._prefix, "rados"), "-p", pool] +
1016                 (["--namespace", namespace] if namespace else []) +
1017                 args)
1018         p = remote.run(
1019             args=args,
1020             stdin=stdin_data,
1021             stdout=StringIO())
1022         return p.stdout.getvalue().strip()
1023
1024     def list_dirfrag(self, dir_ino):
1025         """
1026         Read the named object and return the list of omap keys
1027
1028         :return a list of 0 or more strings
1029         """
1030
1031         dirfrag_obj_name = "{0:x}.00000000".format(dir_ino)
1032
1033         try:
1034             key_list_str = self.rados(["listomapkeys", dirfrag_obj_name])
1035         except CommandFailedError as e:
1036             log.error(e.__str__())
1037             raise ObjectNotFound(dirfrag_obj_name)
1038
1039         return key_list_str.split("\n") if key_list_str else []
1040
1041     def erase_metadata_objects(self, prefix):
1042         """
1043         For all objects in the metadata pool matching the prefix,
1044         erase them.
1045
1046         This O(N) with the number of objects in the pool, so only suitable
1047         for use on toy test filesystems.
1048         """
1049         all_objects = self.rados(["ls"]).split("\n")
1050         matching_objects = [o for o in all_objects if o.startswith(prefix)]
1051         for o in matching_objects:
1052             self.rados(["rm", o])
1053
1054     def erase_mds_objects(self, rank):
1055         """
1056         Erase all the per-MDS objects for a particular rank.  This includes
1057         inotable, sessiontable, journal
1058         """
1059
1060         def obj_prefix(multiplier):
1061             """
1062             MDS object naming conventions like rank 1's
1063             journal is at 201.***
1064             """
1065             return "%x." % (multiplier * 0x100 + rank)
1066
1067         # MDS_INO_LOG_OFFSET
1068         self.erase_metadata_objects(obj_prefix(2))
1069         # MDS_INO_LOG_BACKUP_OFFSET
1070         self.erase_metadata_objects(obj_prefix(3))
1071         # MDS_INO_LOG_POINTER_OFFSET
1072         self.erase_metadata_objects(obj_prefix(4))
1073         # MDSTables & SessionMap
1074         self.erase_metadata_objects("mds{rank:d}_".format(rank=rank))
1075
1076     @property
1077     def _prefix(self):
1078         """
1079         Override this to set a different
1080         """
1081         return ""
1082
1083     def _run_tool(self, tool, args, rank=None, quiet=False):
1084         # Tests frequently have [client] configuration that jacks up
1085         # the objecter log level (unlikely to be interesting here)
1086         # and does not set the mds log level (very interesting here)
1087         if quiet:
1088             base_args = [os.path.join(self._prefix, tool), '--debug-mds=1', '--debug-objecter=1']
1089         else:
1090             base_args = [os.path.join(self._prefix, tool), '--debug-mds=4', '--debug-objecter=1']
1091
1092         if rank is not None:
1093             base_args.extend(["--rank", "%d" % rank])
1094
1095         t1 = datetime.datetime.now()
1096         r = self.tool_remote.run(
1097             args=base_args + args,
1098             stdout=StringIO()).stdout.getvalue().strip()
1099         duration = datetime.datetime.now() - t1
1100         log.info("Ran {0} in time {1}, result:\n{2}".format(
1101             base_args + args, duration, r
1102         ))
1103         return r
1104
1105     @property
1106     def tool_remote(self):
1107         """
1108         An arbitrary remote to use when invoking recovery tools.  Use an MDS host because
1109         it'll definitely have keys with perms to access cephfs metadata pool.  This is public
1110         so that tests can use this remote to go get locally written output files from the tools.
1111         """
1112         mds_id = self.mds_ids[0]
1113         return self.mds_daemons[mds_id].remote
1114
1115     def journal_tool(self, args, rank=None, quiet=False):
1116         """
1117         Invoke cephfs-journal-tool with the passed arguments, and return its stdout
1118         """
1119         return self._run_tool("cephfs-journal-tool", args, rank, quiet)
1120
1121     def table_tool(self, args, quiet=False):
1122         """
1123         Invoke cephfs-table-tool with the passed arguments, and return its stdout
1124         """
1125         return self._run_tool("cephfs-table-tool", args, None, quiet)
1126
1127     def data_scan(self, args, quiet=False, worker_count=1):
1128         """
1129         Invoke cephfs-data-scan with the passed arguments, and return its stdout
1130
1131         :param worker_count: if greater than 1, multiple workers will be run
1132                              in parallel and the return value will be None
1133         """
1134
1135         workers = []
1136
1137         for n in range(0, worker_count):
1138             if worker_count > 1:
1139                 # data-scan args first token is a command, followed by args to it.
1140                 # insert worker arguments after the command.
1141                 cmd = args[0]
1142                 worker_args = [cmd] + ["--worker_n", n.__str__(), "--worker_m", worker_count.__str__()] + args[1:]
1143             else:
1144                 worker_args = args
1145
1146             workers.append(Greenlet.spawn(lambda wargs=worker_args:
1147                                           self._run_tool("cephfs-data-scan", wargs, None, quiet)))
1148
1149         for w in workers:
1150             w.get()
1151
1152         if worker_count == 1:
1153             return workers[0].value
1154         else:
1155             return None