Add stage base class, refactor discovery, etc
diff --git a/wally/ceph.py b/wally/ceph.py
new file mode 100644
index 0000000..e23343e
--- /dev/null
+++ b/wally/ceph.py
@@ -0,0 +1,112 @@
+""" Collect data about ceph nodes"""
+import json
+import logging
+from typing import Set, Dict, cast
+
+
+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
+
+
+logger = logging.getLogger("wally.discover")
+
+
+def get_osds_ips(node: IRPCNode, conf: str, key: str) -> Set[IP]:
+ """Get set of osd's ip"""
+
+ data = node.run("ceph -c {} -k {} --format json osd dump".format(conf, key))
+ jdata = json.loads(data)
+ ips = set() # type: Set[IP]
+ first_error = True
+ for osd_data in jdata["osds"]:
+ if "public_addr" not in osd_data:
+ if first_error:
+ osd_id = osd_data.get("osd", "<OSD_ID_MISSED>")
+ logger.warning("No 'public_addr' field in 'ceph osd dump' output for osd %s" +
+ "(all subsequent errors omitted)", osd_id)
+ first_error = False
+ else:
+ ip_port = osd_data["public_addr"]
+ if '/' in ip_port:
+ ip_port = ip_port.split("/", 1)[0]
+ ips.add(IP(ip_port.split(":")[0]))
+ return ips
+
+
+def get_mons_ips(node: IRPCNode, conf: str, key: str) -> Set[IP]:
+ """Return mon ip set"""
+
+ data = node.run("ceph -c {} -k {} --format json mon_status".format(conf, key))
+ jdata = json.loads(data)
+ ips = set() # type: Set[IP]
+ first_error = True
+ for mon_data in jdata["monmap"]["mons"]:
+ if "addr" not in mon_data:
+ if first_error:
+ mon_name = mon_data.get("name", "<MON_NAME_MISSED>")
+ logger.warning("No 'addr' field in 'ceph mon_status' output for mon %s" +
+ "(all subsequent errors omitted)", mon_name)
+ first_error = False
+ else:
+ ip_port = mon_data["addr"]
+ if '/' in ip_port:
+ ip_port = ip_port.split("/", 1)[0]
+ ips.add(IP(ip_port.split(":")[0]))
+
+ return ips
+
+
+class DiscoverCephStage(Stage):
+ config_block = 'ceph'
+ priority = StepOrder.DISCOVER
+
+ def run(self, ctx: TestRun) -> None:
+ """Return list of ceph's nodes NodeInfo"""
+
+ if 'ceph_nodes' in ctx.storage:
+ ctx.nodes_info.extend(ctx.storage.load_list(NodeInfo, 'ceph_nodes'))
+ else:
+ 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")
+ info = NodeInfo(parse_ssh_uri(root_node_uri), set())
+ ceph_nodes = {} # type: Dict[IP, NodeInfo]
+
+ if conf is None:
+ conf = "/etc/ceph/{}.conf".format(cluster)
+
+ if key is None:
+ key = "/etc/ceph/{}.client.admin.keyring".format(cluster)
+
+ with setup_rpc(connect(info), ctx.rpc_code) as node:
+
+ # new_nodes.extend(ceph.discover_ceph_nodes(ceph_root_conn, cluster=cluster, conf=conf, key=key))
+ ssh_key = node.get_file_content("~/.ssh/id_rsa")
+
+ try:
+ for ip in get_osds_ips(node, conf, key):
+ if ip in ceph_nodes:
+ ceph_nodes[ip].roles.add("ceph-osd")
+ else:
+ ceph_nodes[ip] = NodeInfo(ConnCreds(cast(str, ip), user="root", key=ssh_key), {"ceph-osd"})
+ except Exception as exc:
+ logger.error("OSD discovery failed: %s", exc)
+
+ try:
+ for ip in get_mons_ips(node, conf, key):
+ if ip in ceph_nodes:
+ ceph_nodes[ip].roles.add("ceph-mon")
+ else:
+ ceph_nodes[ip] = NodeInfo(ConnCreds(cast(str, ip), user="root", key=ssh_key), {"ceph-mon"})
+ except Exception as exc:
+ logger.error("MON discovery failed: %s", exc)
+
+ ctx.nodes_info.extend(ceph_nodes.values())
+ ctx.storage['ceph-nodes'] = list(ceph_nodes.values())