| """ Collect data about ceph nodes""" |
| import json |
| import logging |
| from typing import Dict, cast, List, Set, Optional |
| |
| |
| from .node_interfaces import NodeInfo, IRPCNode |
| from .ssh_utils import ConnCreds |
| from .common_types import IP |
| from .stage import Stage, StepOrder |
| from .test_run_class import TestRun |
| from .ssh_utils import parse_ssh_uri |
| from .node import connect, setup_rpc |
| from .utils import StopTestError, to_ip |
| |
| |
| from cephlib import discover |
| from cephlib.discover import OSDInfo |
| |
| |
| logger = logging.getLogger("wally") |
| |
| |
| def get_osds_info(node: IRPCNode, ceph_extra_args: str = "") -> Dict[IP, List[OSDInfo]]: |
| """Get set of osd's ip""" |
| res = {} # type: Dict[IP, List[OSDInfo]] |
| return {IP(ip): osd_info_list |
| for ip, osd_info_list in discover.get_osds_nodes(node.run, ceph_extra_args)} |
| |
| |
| def get_mons_ips(node: IRPCNode, ceph_extra_args: str = "") -> Set[IP]: |
| """Return mon ip set""" |
| return {IP(ip) for ip in discover.get_mons_nodes(node.run, ceph_extra_args).values()} |
| |
| |
| class DiscoverCephStage(Stage): |
| config_block = 'ceph' |
| priority = StepOrder.DISCOVER |
| |
| def run(self, ctx: TestRun) -> None: |
| """Return list of ceph's nodes NodeInfo""" |
| |
| discovery = ctx.config.get("discovery") |
| if discovery == 'disable' or discovery == 'metadata': |
| logger.info("Skip ceph discovery due to config setting") |
| return |
| |
| if 'all_nodes' in ctx.storage: |
| logger.debug("Skip ceph discovery, use previously discovered nodes") |
| return |
| |
| ceph = ctx.config.ceph |
| root_node_uri = cast(str, ceph.root_node) |
| cluster = ceph.get("cluster", "ceph") |
| |
| conf = ceph.get("conf") |
| key = ceph.get("key") |
| |
| if conf is None: |
| conf = "/etc/ceph/{}.conf".format(cluster) |
| |
| if key is None: |
| key = "/etc/ceph/{}.client.admin.keyring".format(cluster) |
| |
| ceph_extra_args = "" |
| |
| if conf: |
| ceph_extra_args += " -c '{}'".format(conf) |
| |
| if key: |
| ceph_extra_args += " -k '{}'".format(key) |
| |
| logger.debug("Start discovering ceph nodes from root %s", root_node_uri) |
| logger.debug("cluster=%s key=%s conf=%s", cluster, conf, key) |
| |
| info = NodeInfo(parse_ssh_uri(root_node_uri), set()) |
| |
| ceph_params = {"cluster": cluster, "conf": conf, "key": key} |
| |
| with setup_rpc(connect(info), ctx.rpc_code, ctx.default_rpc_plugins, |
| log_level=ctx.config.rpc_log_level) as node: |
| |
| ssh_key = node.get_file_content("~/.ssh/id_rsa") |
| |
| try: |
| ips = set() |
| for ip, osds_info in get_osds_info(node, ceph_extra_args).items(): |
| ips.add(ip) |
| creds = ConnCreds(to_ip(cast(str, ip)), user="root", key=ssh_key) |
| info = ctx.merge_node(creds, {'ceph-osd'}) |
| info.params.setdefault('ceph-osds', []).extend(osds_info) |
| assert 'ceph' not in info.params or info.params['ceph'] == ceph_params |
| info.params['ceph'] = ceph_params |
| |
| logger.debug("Found %s nodes with ceph-osd role", len(ips)) |
| except Exception as exc: |
| if discovery != 'ignore_errors': |
| logger.exception("OSD discovery failed") |
| raise StopTestError() |
| else: |
| logger.warning("OSD discovery failed %s", exc) |
| |
| try: |
| counter = 0 |
| for counter, ip in enumerate(get_mons_ips(node, ceph_extra_args)): |
| creds = ConnCreds(to_ip(cast(str, ip)), user="root", key=ssh_key) |
| info = ctx.merge_node(creds, {'ceph-mon'}) |
| assert 'ceph' not in info.params or info.params['ceph'] == ceph_params |
| info.params['ceph'] = ceph_params |
| logger.debug("Found %s nodes with ceph-mon role", counter + 1) |
| except Exception as exc: |
| if discovery != 'ignore_errors': |
| logger.exception("MON discovery failed") |
| raise StopTestError() |
| else: |
| logger.warning("MON discovery failed %s", exc) |