]>
git.proxmox.com Git - ceph.git/blob - ceph/qa/tasks/cephfs/filesystem.py
2 from StringIO
import StringIO
5 from gevent
import Greenlet
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
20 log
= logging
.getLogger(__name__
)
23 DAEMON_WAIT_TIMEOUT
= 120
27 class ObjectNotFound(Exception):
28 def __init__(self
, object_name
):
29 self
._object
_name
= object_name
32 return "Object not found: '{0}'".format(self
._object
_name
)
34 class FSStatus(object):
36 Operations on a snapshot of the FSMap.
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"))
43 return json
.dumps(self
.map, indent
= 2, sort_keys
= True)
45 # Expose the fsmap for manual inspection.
46 def __getitem__(self
, key
):
48 Get a field from the fsmap.
52 def get_filesystems(self
):
54 Iterator for all filesystems.
56 for fs
in self
.map['filesystems']:
61 Iterator for all the mds_info components in the FSMap.
63 for info
in self
.get_standbys():
65 for fs
in self
.map['filesystems']:
66 for info
in fs
['mdsmap']['info'].values():
69 def get_standbys(self
):
71 Iterator for all standbys.
73 for info
in self
.map['standbys']:
76 def get_fsmap(self
, fscid
):
78 Get the fsmap for the given FSCID.
80 for fs
in self
.map['filesystems']:
81 if fscid
is None or fs
['id'] == fscid
:
83 raise RuntimeError("FSCID {0} not in map".format(fscid
))
85 def get_fsmap_byname(self
, name
):
87 Get the fsmap for the given file system name.
89 for fs
in self
.map['filesystems']:
90 if name
is None or fs
['mdsmap']['fs_name'] == name
:
92 raise RuntimeError("FS {0} not in map".format(name
))
94 def get_replays(self
, fscid
):
96 Get the standby:replay MDS for the given FSCID.
98 fs
= self
.get_fsmap(fscid
)
99 for info
in fs
['mdsmap']['info'].values():
100 if info
['state'] == 'up:standby-replay':
103 def get_ranks(self
, fscid
):
105 Get the ranks for the given FSCID.
107 fs
= self
.get_fsmap(fscid
)
108 for info
in fs
['mdsmap']['info'].values():
109 if info
['rank'] >= 0:
112 def get_rank(self
, fscid
, rank
):
114 Get the rank for the given FSCID.
116 for info
in self
.get_ranks(fscid
):
117 if info
['rank'] == rank
:
119 raise RuntimeError("FSCID {0} has no rank {1}".format(fscid
, rank
))
121 def get_mds(self
, name
):
123 Get the info for the given MDS name.
125 for info
in self
.get_all():
126 if info
['name'] == name
:
130 def get_mds_addr(self
, name
):
132 Return the instance addr as a string, like "10.214.133.138:6807\/10825"
134 info
= self
.get_mds(name
)
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
))
141 class CephCluster(object):
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()
148 def __init__(self
, ctx
):
150 self
.mon_manager
= ceph_manager
.CephManager(self
.admin_remote
, ctx
=ctx
, logger
=log
.getChild('ceph_manager'))
152 def get_config(self
, key
, service_type
=None):
154 Get config from mon by default, or a specific service if caller asks for it
156 if service_type
is None:
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
]
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.
169 def clear_ceph_conf(self
, subsys
, key
):
170 del self
._ctx
.ceph
['ceph'].conf
[subsys
][key
]
171 write_conf(self
._ctx
)
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
)
183 class MDSCluster(CephCluster
):
185 Collective operations on all the MDS daemons in the Ceph cluster. These
186 daemons may be in use by various Filesystems.
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.
192 def __init__(self
, ctx
):
193 super(MDSCluster
, self
).__init
__(ctx
)
195 self
.mds_ids
= list(misc
.all_roles_of_type(ctx
.cluster
, 'mds'))
197 if len(self
.mds_ids
) == 0:
198 raise RuntimeError("This task requires at least one MDS")
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
])
204 def _one_or_all(self
, mds_id
, cb
, in_parallel
=True):
206 Call a callback for a single named MDS, or for all.
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`)
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)
219 with
parallel() as p
:
220 for mds_id
in self
.mds_ids
:
223 for mds_id
in self
.mds_ids
:
228 def mds_stop(self
, mds_id
=None):
230 Stop the MDS daemon process(se). If it held a rank, that rank
231 will eventually go laggy.
233 self
._one
_or
_all
(mds_id
, lambda id_
: self
.mds_daemons
[id_
].stop())
235 def mds_fail(self
, mds_id
=None):
237 Inform MDSMonitor of the death of the daemon process(es). If it held
238 a rank, that rank will be relinquished.
240 self
._one
_or
_all
(mds_id
, lambda id_
: self
.mon_manager
.raw_cluster_cmd("mds", "fail", id_
))
242 def mds_restart(self
, mds_id
=None):
243 self
._one
_or
_all
(mds_id
, lambda id_
: self
.mds_daemons
[id_
].restart())
245 def mds_fail_restart(self
, mds_id
=None):
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.
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()
257 self
._one
_or
_all
(mds_id
, _fail_restart
)
259 def newfs(self
, name
):
260 return Filesystem(self
._ctx
, create
=name
)
263 return FSStatus(self
.mon_manager
)
265 def delete_all_filesystems(self
):
267 Remove all filesystems that exist, and any pools in use by them.
269 pools
= json
.loads(self
.mon_manager
.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools']
272 pool_id_name
[pool
['pool']] = pool
['pool_name']
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")
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']]
285 for gid
in mdsmap
['up'].values():
286 self
.mon_manager
.raw_cluster_cmd('mds', 'fail', gid
.__str
__())
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
]
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
304 def get_standby_daemons(self
):
305 return set([s
['name'] for s
in self
.status().get_standbys()])
307 def get_mds_hostnames(self
):
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
)
315 def set_clients_block(self
, blocked
, mds_id
=None):
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.
321 :param mds_id: Optional ID of MDS to block, default to all
324 da_flag
= "-A" if blocked
else "-D"
326 def set_block(_mds_id
):
327 remote
= self
.mon_manager
.find_remote('mds', _mds_id
)
328 status
= self
.status()
330 addr
= status
.get_mds_addr(_mds_id
)
331 ip_str
, port_str
, inst_str
= re
.match("(.+):(.+)/(.+)", addr
).groups()
334 args
=["sudo", "iptables", da_flag
, "OUTPUT", "-p", "tcp", "--sport", port_str
, "-j", "REJECT", "-m",
335 "comment", "--comment", "teuthology"])
337 args
=["sudo", "iptables", da_flag
, "INPUT", "-p", "tcp", "--dport", port_str
, "-j", "REJECT", "-m",
338 "comment", "--comment", "teuthology"])
340 self
._one
_or
_all
(mds_id
, set_block
, in_parallel
=False)
342 def clear_firewall(self
):
343 clear_firewall(self
._ctx
)
345 def get_mds_info(self
, mds_id
):
346 return FSStatus(self
.mon_manager
).get_mds(mds_id
)
349 flags
= json
.loads(self
.mon_manager
.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['flags']
350 return 'full' in flags
352 def is_pool_full(self
, pool_name
):
353 pools
= json
.loads(self
.mon_manager
.raw_cluster_cmd("osd", "dump", "--format=json-pretty"))['pools']
355 if pool
['pool_name'] == pool_name
:
356 return 'full' in pool
['flags_names'].split(",")
358 raise RuntimeError("Pool not found '{0}'".format(pool_name
))
360 class Filesystem(MDSCluster
):
362 This object is for driving a CephFS filesystem. The MDS daemons driven by
363 MDSCluster may be shared with other Filesystems.
365 def __init__(self
, ctx
, fscid
=None, create
=None):
366 super(Filesystem
, self
).__init
__(ctx
)
370 self
.metadata_pool_name
= None
371 self
.data_pools
= None
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]
377 if create
is not None:
378 if fscid
is not None:
379 raise RuntimeError("cannot specify fscid when creating fs")
384 if not self
.legacy_configured():
386 elif fscid
is not None:
388 self
.getinfo(refresh
= True)
390 # Stash a reference to the first created filesystem on ctx, so
391 # that if someone drops to the interactive shell they can easily
393 if not hasattr(self
._ctx
, "filesystem"):
394 self
._ctx
.filesystem
= self
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
)
403 fss
= [fs
for fs
in status
.get_filesystems()]
407 raise RuntimeError("no file system available")
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
)
415 def deactivate(self
, rank
):
417 raise RuntimeError("invalid rank")
419 raise RuntimeError("cannot deactivate rank 0")
420 self
.mon_manager
.raw_cluster_cmd("mds", "deactivate", "%d:%d" % (self
.id, rank
))
422 def set_max_mds(self
, max_mds
):
423 self
.mon_manager
.raw_cluster_cmd("fs", "set", self
.name
, "max_mds", "%d" % max_mds
)
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')
428 def get_pgs_per_fs_pool(self
):
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
433 :return: an integer number of PGs
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
440 if self
.name
is None:
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
)
446 log
.info("Creating filesystem '{0}'".format(self
.name
))
448 pgs_per_fs_pool
= self
.get_pgs_per_fs_pool()
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.
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)
466 self
.getinfo(refresh
= True)
469 if getattr(self
._ctx
, "filesystem", None) == self
:
470 delattr(self
._ctx
, "filesystem")
474 Whether a filesystem exists in the mon's filesystem list
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
]
479 def legacy_configured(self
):
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
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
498 return metadata_pool_exists
501 return json
.loads(self
.mon_manager
.raw_cluster_cmd("df", "--format=json-pretty"))
503 def get_mds_map(self
):
504 return self
.status().get_fsmap(self
.id)['mdsmap']
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():
513 raise RuntimeError("could not get just created pool '{0}'".format(name
))
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:
518 status
= self
.status()
519 fsmap
= status
.get_fsmap(self
.id)
521 osd_map
= self
.mon_manager
.get_osd_dump_json()
523 for p
in osd_map
['pools']:
524 id_to_name
[p
['pool']] = p
['pool_name']
526 self
.metadata_pool_name
= id_to_name
[fsmap
['mdsmap']['metadata_pool']]
528 for data_pool
in fsmap
['mdsmap']['data_pools']:
529 self
.data_pools
[data_pool
] = id_to_name
[data_pool
]
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]
537 def get_data_pool_id(self
, refresh
= False):
539 Don't call this if you have multiple data pools
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]
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()
552 def get_metadata_pool_name(self
):
553 return self
.metadata_pool_name
555 def get_namespace_id(self
):
558 def get_pool_df(self
, pool_name
):
561 {u'bytes_used': 0, u'max_avail': 83848701, u'objects': 0, u'kb_used': 0}
563 for pool_df
in self
._df
()['pools']:
564 if pool_df
['name'] == pool_name
:
565 return pool_df
['stats']
567 raise RuntimeError("Pool name '{0}' not found".format(pool_name
))
570 return self
._df
()['stats']['total_used_bytes']
572 def are_daemons_healthy(self
):
574 Return true if all daemons are in one of active, standby, standby-replay, and
575 at least max_mds daemons are in 'active'.
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.
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'))
596 log
.info("are_daemons_healthy: mds map: {0}".format(mds_map
))
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']))
602 elif mds_status
['state'] == 'up:active':
605 log
.info("are_daemons_healthy: {0}/{1}".format(
606 active_count
, mds_map
['max_mds']
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':
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
620 # MDS not even running
623 if daemon_status
['state'] != 'up:active':
624 # MDS hasn't taken the latest map yet
631 def get_daemon_names(self
, state
=None):
633 Return MDS daemon names of those daemons in the given state
637 status
= self
.get_mds_map()
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'])
645 def get_active_names(self
):
647 Return MDS daemon names of those daemons holding ranks
650 :return: list of strings like ['a', 'b'], sorted by rank
652 return self
.get_daemon_names("up:active")
654 def get_all_mds_rank(self
):
655 status
= self
.get_mds_map()
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'])
663 def get_rank_names(self
):
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
670 status
= self
.get_mds_map()
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'])
678 def wait_for_daemons(self
, timeout
=None):
680 Wait until all daemons are healthy
685 timeout
= DAEMON_WAIT_TIMEOUT
689 if self
.are_daemons_healthy():
695 if elapsed
> timeout
:
696 raise RuntimeError("Timed out waiting for MDS daemons to become healthy")
698 def get_lone_mds_id(self
):
700 Get a single MDS ID: the only one if there is only one
701 configured, else the only one currently holding a rank,
704 if len(self
.mds_ids
) != 1:
705 alive
= self
.get_rank_names()
709 raise ValueError("Explicit MDS argument required when multiple MDSs in use")
711 return self
.mds_ids
[0]
714 log
.info("Creating new filesystem")
715 self
.delete_all_filesystems()
719 def put_metadata_object_raw(self
, object_id
, infile
):
721 Save an object to the metadata pool
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
728 def get_metadata_object_raw(self
, object_id
):
730 Retrieve an object from the metadata pool and store it in a file.
732 temp_bin_path
= '/tmp/' + object_id
+ '.bin'
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
740 def get_metadata_object(self
, object_type
, object_id
):
742 Retrieve an object from the metadata pool, pass it through
743 ceph-dencoder to dump it to JSON, and return the decoded object.
745 temp_bin_path
= '/tmp/out.bin'
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
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'
755 dump_json
= stdout
.getvalue().strip()
757 dump
= json
.loads(dump_json
)
758 except (TypeError, ValueError):
759 log
.error("Failed to decode JSON: '{0}'".format(dump_json
))
764 def get_journal_version(self
):
766 Read the JournalPointer and Journal::Header objects to learn the version of
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']
773 journal_header_object
= "{0:x}.00000000".format(journal_ino
)
774 journal_header_dump
= self
.get_metadata_object('Journaler::Header', journal_header_object
)
776 version
= journal_header_dump
['journal_header']['stream_format']
777 log
.info("Read journal version {0}".format(version
))
781 def mds_asok(self
, command
, mds_id
=None):
783 mds_id
= self
.get_lone_mds_id()
785 return self
.json_asok(command
, 'mds', mds_id
)
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
)
793 raise RuntimeError("Path not found in cache: {0}".format(path
))
797 def wait_for_state(self
, goal_state
, reject
=None, timeout
=None, mds_id
=None, rank
=None):
799 Block until the MDS reaches a particular state, or a failure condition
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.
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
811 started_at
= time
.time()
813 status
= self
.status()
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
))
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
832 log
.info("mapped states {0} to {1}".format(states
, current_state
))
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
))
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)))
843 "Reached timeout after {0} seconds waiting for state {1}, while in state {2}".format(
844 elapsed
, goal_state
, current_state
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
853 pool
= self
.get_data_pool_name()
855 obj_name
= "{0:x}.00000000".format(ino_no
)
858 os
.path
.join(self
._prefix
, "rados"), "-p", pool
, "getxattr", obj_name
, xattr_name
864 except CommandFailedError
as e
:
865 log
.error(e
.__str
__())
866 raise ObjectNotFound(obj_name
)
868 data
= proc
.stdout
.getvalue()
871 args
=[os
.path
.join(self
._prefix
, "ceph-dencoder"), "type", type, "import", "-", "decode", "dump_json"],
876 return json
.loads(p
.stdout
.getvalue().strip())
878 def _write_data_xattr(self
, ino_no
, xattr_name
, data
, pool
=None):
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.
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
889 remote
= self
.mds_daemons
[self
.mds_ids
[0]].remote
891 pool
= self
.get_data_pool_name()
893 obj_name
= "{0:x}.00000000".format(ino_no
)
895 os
.path
.join(self
._prefix
, "rados"), "-p", pool
, "setxattr",
896 obj_name
, xattr_name
, data
902 def read_backtrace(self
, ino_no
, pool
=None):
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:
909 rados -p cephfs_data getxattr 10000000002.00000000 parent > out.bin
910 ceph-dencoder type inode_backtrace_t import out.bin decode dump_json
912 { "ino": 1099511627778,
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.
923 return self
._read
_data
_xattr
(ino_no
, "parent", "inode_backtrace_t", pool
)
925 def read_layout(self
, ino_no
, pool
=None):
927 Read 'layout' xattr of an inode and parse the result, returning a dict like:
930 "stripe_unit": 4194304,
932 "object_size": 4194304,
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.
940 return self
._read
_data
_xattr
(ino_no
, "layout", "file_layout_t", pool
)
942 def _enumerate_data_objects(self
, ino
, size
):
944 Get the list of expected data objects for a range, and the list of objects
947 :return a tuple of two lists of strings (expected, actual)
949 stripe_size
= 1024 * 1024 * 4
951 size
= max(stripe_size
, size
)
954 "{0:x}.{1:08x}".format(ino
, n
)
955 for n
in range(0, ((size
- 1) / stripe_size
) + 1)
958 exist_objects
= self
.rados(["ls"], pool
=self
.get_data_pool_name()).split("\n")
960 return want_objects
, exist_objects
962 def data_objects_present(self
, ino
, size
):
964 Check that *all* the expected data objects for an inode are present in the data pool
967 want_objects
, exist_objects
= self
._enumerate
_data
_objects
(ino
, size
)
968 missing
= set(want_objects
) - set(exist_objects
)
971 log
.info("Objects missing (ino {0}, size {1}): {2}".format(
976 log
.info("All objects for ino {0} size {1} found".format(ino
, size
))
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
)
984 log
.info("Objects not absent (ino {0}, size {1}): {2}".format(
989 log
.info("All objects for ino {0} size {1} are absent".format(ino
, size
))
992 def dirfrag_exists(self
, ino
, frag
):
994 self
.rados(["stat", "{0:x}.{1:08x}".format(ino
, frag
)])
995 except CommandFailedError
as e
:
1000 def rados(self
, args
, pool
=None, namespace
=None, stdin_data
=None):
1002 Call into the `rados` CLI from an MDS
1006 pool
= self
.get_metadata_pool_name()
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
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 []) +
1022 return p
.stdout
.getvalue().strip()
1024 def list_dirfrag(self
, dir_ino
):
1026 Read the named object and return the list of omap keys
1028 :return a list of 0 or more strings
1031 dirfrag_obj_name
= "{0:x}.00000000".format(dir_ino
)
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
)
1039 return key_list_str
.split("\n") if key_list_str
else []
1041 def erase_metadata_objects(self
, prefix
):
1043 For all objects in the metadata pool matching the prefix,
1046 This O(N) with the number of objects in the pool, so only suitable
1047 for use on toy test filesystems.
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
])
1054 def erase_mds_objects(self
, rank
):
1056 Erase all the per-MDS objects for a particular rank. This includes
1057 inotable, sessiontable, journal
1060 def obj_prefix(multiplier
):
1062 MDS object naming conventions like rank 1's
1063 journal is at 201.***
1065 return "%x." % (multiplier
* 0x100 + rank
)
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
))
1079 Override this to set a different
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)
1088 base_args
= [os
.path
.join(self
._prefix
, tool
), '--debug-mds=1', '--debug-objecter=1']
1090 base_args
= [os
.path
.join(self
._prefix
, tool
), '--debug-mds=4', '--debug-objecter=1']
1092 if rank
is not None:
1093 base_args
.extend(["--rank", "%d" % rank
])
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
1106 def tool_remote(self
):
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.
1112 mds_id
= self
.mds_ids
[0]
1113 return self
.mds_daemons
[mds_id
].remote
1115 def journal_tool(self
, args
, rank
=None, quiet
=False):
1117 Invoke cephfs-journal-tool with the passed arguments, and return its stdout
1119 return self
._run
_tool
("cephfs-journal-tool", args
, rank
, quiet
)
1121 def table_tool(self
, args
, quiet
=False):
1123 Invoke cephfs-table-tool with the passed arguments, and return its stdout
1125 return self
._run
_tool
("cephfs-table-tool", args
, None, quiet
)
1127 def data_scan(self
, args
, quiet
=False, worker_count
=1):
1129 Invoke cephfs-data-scan with the passed arguments, and return its stdout
1131 :param worker_count: if greater than 1, multiple workers will be run
1132 in parallel and the return value will be None
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.
1142 worker_args
= [cmd
] + ["--worker_n", n
.__str
__(), "--worker_m", worker_count
.__str
__()] + args
[1:]
1146 workers
.append(Greenlet
.spawn(lambda wargs
=worker_args
:
1147 self
._run
_tool
("cephfs-data-scan", wargs
, None, quiet
)))
1152 if worker_count
== 1:
1153 return workers
[0].value