Explorer + cli

Dockerfile

@@ -10,7 +10,8 @@ RUN pip install --no-cache-dir -r requirements.txt
 
 COPY . .
 
-# Runtime state lives in the working directory: state.json, storage/, logs/
+RUN chmod +x colonyctl.py && ln -s /app/colonyctl.py /usr/local/bin/colonyctl
+
 VOLUME ["/app/storage", "/app/logs"]
 
 EXPOSE 65334/tcp 28149/udp

colony/cli.py

@@ -0,0 +1,193 @@
+import argparse
+import json
+import sys
+import time
+
+from colony.core.network.requests import CommandRequest
+from colony.core.roles.unit.role import Unit
+from colony.settings import configuration
+from colony.transport.whisperer import Whisperer
+
+DEFAULT_TIMEOUT = 5
+DEFAULT_INTERVAL = 0.5
+
+HOME = "\033[H"
+CLEAR_SCREEN = "\033[2J"
+CLEAR_LINE = "\033[K"
+CLEAR_BELOW = "\033[J"
+
+SCORE_COLUMNS = (
+    ("cpu", "CPU"),
+    ("memory", "MEM GB"),
+    ("gpu", "GPU GB"),
+    ("disk", "DISK GB"),
+    ("longevity", "UP H"),
+)
+
+
+def fetch_cluster(host, port, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
+    payload = CommandRequest.form(Unit.COMMANDS.SHOW_CLUSTER).collapse()
+    resp = Whisperer(host, port, cert_file, keyfile).send(payload, timeout=timeout)
+
+    if not isinstance(resp, dict) or resp.get("status") != "ok" or not isinstance(resp.get("result"), dict):
+        reason = resp.get("error") if isinstance(resp, dict) else resp
+        raise RuntimeError(f"the node answered with an error: {reason}")
+    return resp["result"]
+
+
+def format_age(seconds):
+    seconds = int(seconds)
+    if seconds < 60:
+        return f"{seconds}s"
+    if seconds < 3600:
+        return f"{seconds // 60}m{seconds % 60:02d}s"
+    return f"{seconds // 3600}h{seconds % 3600 // 60:02d}m"
+
+
+def table(headers, rows):
+    widths = [max(len(str(cell)) for cell in column) for column in zip(headers, *rows)]
+    lines = ["  ".join(str(cell).ljust(width) for cell, width in zip(row, widths)).rstrip() for row in [headers, *rows]]
+    return "\n".join(lines)
+
+
+def format_score(value):
+    return "-" if value is None else f"{value:.1f}"
+
+
+def address(member):
+    return f"{member['ip']}:{member['port']}"
+
+
+def role_states(payload):
+    alive = [member for member in payload["members"] if member["alive"]]
+    states = []
+    for role in payload["roles"]:
+        holders = sorted(address(member) for member in alive if role["name"] in member["roles"])
+        required = role["required"]
+        if required is None:
+            state = "general"
+        elif len(holders) < required:
+            state = "unmet"
+        elif len(holders) > required:
+            state = "duplicate"
+        else:
+            state = "held"
+        states.append({**role, "holders": holders, "state": state})
+    return states
+
+
+def render_status(payload):
+    members = payload["members"]
+
+    hostnames = [member["hostname"] for member in members]
+    labels = {
+        address(member): member["hostname"] if hostnames.count(member["hostname"]) == 1 else address(member)
+        for member in members
+    }
+
+    alive = [member for member in members if member["alive"]]
+    down = len(members) - len(alive)
+    self_member = next(member for member in members if member["self"])
+    title = f"Cluster as seen by {labels[address(self_member)]}: {len(alive)} node(s) alive"
+    if down:
+        title += f", {down} down"
+
+    show_declined = any(member["declined"] for member in members)
+    node_headers = ["NODE", "ADDRESS", "STATE", "SEEN", "ROLES"] + (["DECLINED"] if show_declined else [])
+    node_rows = []
+    for member in members:
+        label = labels[address(member)] + (" (this node)" if member["self"] else "")
+        row = [
+            label,
+            address(member),
+            "alive" if member["alive"] else "DOWN",
+            "-" if member["self"] else format_age(member["seen_ago"]),
+            " ".join(member["roles"]) or "-",
+        ]
+        if show_declined:
+            row.append(" ".join(member["declined"]) or "-")
+        node_rows.append(row)
+
+    alive_addresses = {address(member) for member in alive}
+    role_rows = []
+    for role in role_states(payload):
+        holders = role["holders"]
+        if role["state"] == "general":
+            state = "ok"
+        else:
+            state = {
+                "held": "ok",
+                "unmet": f"UNMET ({len(holders)}/{role['required']})",
+                "duplicate": f"DUPLICATE ({len(holders)}/{role['required']})",
+            }[role["state"]]
+        if role["state"] == "general" and set(holders) == alive_addresses:
+            names = "all nodes"
+        else:
+            names = ", ".join(labels[holder] for holder in holders) or "-"
+        role_rows.append([role["name"], role["priority"], state, names])
+
+    score_rows = [
+        [labels[address(member)]] + [format_score(member["scores"].get(key)) for key, _ in SCORE_COLUMNS]
+        for member in alive
+    ]
+
+    sections = [
+        title,
+        table(node_headers, node_rows),
+        table(["ROLE", "PRIORITY", "STATE", "HOLDERS"], role_rows),
+        table(["NODE"] + [header for _, header in SCORE_COLUMNS], score_rows),
+    ]
+    return "\n\n".join(sections)
+
+
+def build_parser():
+    parser = argparse.ArgumentParser(prog="colonyctl", description="Inspect a running colony node.")
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    status = commands.add_parser("status", help="show members, roles and hardware scores (live in a terminal)")
+    status.add_argument("--host", default="127.0.0.1", help="node to ask (default: this machine)")
+    status.add_argument("--port", type=int, default=None, help="node port (default: listener.port from config)")
+    status.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
+    status.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="seconds between refreshes")
+    status.add_argument("--raw", action="store_true", help="print once and exit instead of refreshing")
+    status.add_argument("--json", action="store_true", help="print the raw data as JSON once and exit")
+    return parser
+
+
+def draw(out, text):
+    out.write(HOME + text.replace("\n", CLEAR_LINE + "\n") + CLEAR_LINE + CLEAR_BELOW)
+    out.flush()
+
+
+def watch(host, port, args, out):
+    out.write(CLEAR_SCREEN)
+    try:
+        while True:
+            try:
+                text = render_status(fetch_cluster(host, port, timeout=args.timeout))
+            except (OSError, ValueError, RuntimeError) as e:
+                text = f"colonyctl: cannot get status from {host}:{port}: {e}\nretrying..."
+            footer = f"Every {args.interval:g}s. Updated {time.strftime('%H:%M:%S')}. Ctrl-C to quit."
+            draw(out, f"{text}\n\n{footer}")
+            time.sleep(args.interval)
+    except KeyboardInterrupt:
+        out.write("\n")
+        return 0
+
+
+def main(argv=None, out=None):
+    args = build_parser().parse_args(argv)
+    out = sys.stdout if out is None else out
+    port = args.port if args.port is not None else configuration.listener.port
+
+    if not args.raw and not args.json and out.isatty():
+        return watch(args.host, port, args, out)
+
+    try:
+        payload = fetch_cluster(args.host, port, timeout=args.timeout)
+    except (OSError, ValueError, RuntimeError) as e:
+        print(f"colonyctl: cannot get status from {args.host}:{port}: {e}", file=sys.stderr)
+        return 1
+
+    print(json.dumps(payload, indent=2) if args.json else render_status(payload), file=out)
+    return 0

colony/core/network/membership.py

@@ -203,6 +203,16 @@ class Membership:
                       if now - advanced <= self.member_timeout]
             return sorted(alive, key=lambda record: record.key)
 
+    def overview(self):
+        with self._lock:
+            now = self.clock()
+            self._purge(now)
+            rows = [{**self._current_self().to_dict(), "alive": True, "seen_ago": 0.0, "self": True}]
+            for record, advanced in self._entries.values():
+                seen_ago = now - advanced
+                rows.append({**record.to_dict(), "alive": seen_ago <= self.member_timeout, "seen_ago": seen_ago, "self": False})
+            return sorted(rows, key=lambda row: (row["ip"], row["port"]))
+
     def known_addresses(self):
         with self._lock:
             self._purge(self.clock())

colony/core/roles/explorer/role.py

@@ -2,7 +2,9 @@ import datetime
 from concurrent.futures import ThreadPoolExecutor
 from enum import Enum
 
+from colony.core.network.membership import MemberRecord
 from colony.core.storage import DEFAULT_STORAGE_FILES
+from colony.core.roles.gossip.role import Gossip
 from colony.core.roles.role import Role
 from colony.core.roles.unit.role import Unit
 from colony.core.roles.vaultwarden.role import VaultWarden
@@ -66,6 +68,7 @@ class Explorer(Role):
             return
 
         self.save_snapshot(snapshot)
+        self.introduce_unknown_peers(discovered_peers)
 
     def load_snapshot(self):
         try:
@@ -137,3 +140,81 @@ class Explorer(Role):
             self.logger.warning(f"Peer {ip}:{port} answered SHOW_STATS with an error")
             return None
         return resp['result']
+
+    def fetch_members_from_peer(self, ip, port):
+        try:
+            resp = self.context.execute_to_unregistered_node(ip, port, Unit.COMMANDS.SHOW_CLUSTER)
+        except Exception as e:
+            self.logger.warning(f"Could not fetch cluster members from {ip}:{port}: {e}")
+            return []
+
+        result = resp.get('result') if isinstance(resp, dict) and resp.get('status') == 'ok' else None
+        rows = result.get('members') if isinstance(result, dict) else None
+        if not isinstance(rows, list):
+            self.logger.warning(f"Peer {ip}:{port} answered SHOW_CLUSTER with an error")
+            return []
+
+        members = []
+        for row in rows[:setting("membership.max_members")]:
+            try:
+                record = MemberRecord.from_dict(row)
+            except ValueError as e:
+                self.logger.debug(f"Ignoring bad member from {ip}:{port}: {e}")
+                continue
+            if row.get('alive') is True:
+                members.append(record.key)
+        return members
+
+    def _ping(self, address):
+        ip, _ = address
+        try:
+            port = Discover.request(ip, timeout=self.SCAN_TIMEOUT, discover_port=setting("discovery.discover_port"))
+        except Exception:
+            return None
+        return (ip, port) if port is not None else None
+
+    def introduce_unknown_peers(self, discovered_peers):
+        membership = self.context.membership()
+        known = {record.key for record in membership.alive_records()} | set(discovered_peers.items())
+
+        candidates = {}
+        for ip, port in discovered_peers.items():
+            for address in self.fetch_members_from_peer(ip, port):
+                if address not in known:
+                    candidates[address] = None
+        if not candidates or self._stop_event.is_set():
+            return
+
+        addresses = list(candidates)[:setting("membership.max_members")]
+        with ThreadPoolExecutor(max_workers=min(self.SCAN_WORKERS, len(addresses))) as pool:
+            pinged = [address for address in pool.map(self._ping, addresses) if address is not None]
+
+        for ip, port in pinged:
+            self.logger.info(f"Peers know {ip}:{port}, asking it to become our peer")
+            self.send_hello(ip, port)
+
+    def send_hello(self, ip, port):
+        membership = self.context.membership()
+        sender = membership.self_record().to_dict()
+        known = [record.to_dict() for record in membership.alive_records() if record.key != membership.key]
+
+        try:
+            resp = self.context.execute_to_unregistered_node(
+                ip, port, Gossip.COMMANDS.HELLO, sender, known, timeout=setting("membership.hello_timeout")
+            )
+        except Exception as e:
+            self.logger.warning(f"Hello to {ip}:{port} failed: {e}")
+            return False
+
+        result = resp.get('result') if isinstance(resp, dict) and resp.get('status') == 'ok' else None
+        if not isinstance(result, dict) or not result.get('admitted'):
+            self.logger.warning(f"{ip}:{port} did not admit this node")
+            return False
+
+        try:
+            members = [MemberRecord.from_dict(item) for item in result.get('members', [])]
+        except (ValueError, TypeError) as e:
+            self.logger.warning(f"Ignoring bad member list from {ip}:{port}: {e}")
+            return False
+        membership.merge(members[:setting("membership.max_members")])
+        return True

colony/core/roles/gossip/role.py

@@ -65,6 +65,8 @@ class Gossip(Role):
 
     def hello(self, sender, known):
         record = MemberRecord.from_dict(sender)
+        if record.key == self.membership.key:
+            return {"admitted": False, "members": [], "self": True}
         if not self.membership.admit(record):
             return {"admitted": False, "members": []}
 
@@ -137,6 +139,9 @@ class Gossip(Role):
         if not isinstance(result, dict):
             self.logger.debug(f"Hello to {ip}:{port} was answered with an error")
             return FAILED
+        if result.get("self"):
+            self.logger.debug(f"{ip}:{port} is this node")
+            return FAILED
         if not result.get("admitted"):
             self.logger.warning(f"{ip}:{port} refused to admit this node")
             return REFUSED

colony/core/roles/unit/role.py

@@ -1,6 +1,8 @@
 from enum import Enum
+from colony.core.network import advice
 from colony.core.network.hardware import HardwareStatistics
 from colony.core.roles.role import Role, Command
+from colony.role_registry import RolePriority
 
 
 class Unit(Role):
@@ -8,6 +10,7 @@ class Unit(Role):
         SHOW_STATS = 'Unit.ShowStats'
         SHOW_ROLES = 'Unit.ShowRoles'
         SHOW_HARDWARE = 'Unit.ShowHardware'
+        SHOW_CLUSTER = 'Unit.ShowCluster'
         REQUEST_ROLE = 'Unit.RequestRole'
 
     def __init__(self, context):
@@ -15,6 +18,7 @@ class Unit(Role):
             self.COMMANDS.SHOW_STATS.value: Command(self.show_stats),
             self.COMMANDS.SHOW_ROLES.value: Command(self.show_roles),
             self.COMMANDS.SHOW_HARDWARE.value: Command(self.show_hardware),
+            self.COMMANDS.SHOW_CLUSTER.value: Command(self.show_cluster),
             self.COMMANDS.REQUEST_ROLE.value: Command(self.request_role),
         }
 
@@ -23,6 +27,20 @@ class Unit(Role):
     def show_hardware(self):
         return HardwareStatistics.collect().to_dict()
 
+    def show_cluster(self):
+        roles = self.context.get_role_registry().roles
+        return {
+            "members": self.context.membership().overview(),
+            "roles": [
+                {
+                    "name": name,
+                    "priority": roles[name].priority.name.lower(),
+                    "required": None if roles[name].priority == RolePriority.GENERAL else advice.required_holders(roles[name]),
+                }
+                for name in sorted(roles, key=lambda name: (roles[name].priority.value, name))
+            ],
+        }
+
     def show_roles(self):
         return list(self.context.self().roles)
 

colonyctl.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+import sys
+
+from colony.cli import main
+
+if __name__ == "__main__":
+    sys.exit(main())