216 lines
7.7 KiB
Python
216 lines
7.7 KiB
Python
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"
|
||
|
||
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_size(gb):
|
||
if gb >= 1000:
|
||
return f"{gb / 1024:.1f} TB"
|
||
if gb >= 100:
|
||
return f"{gb:.0f} GB"
|
||
return f"{gb:.1f} GB"
|
||
|
||
|
||
def format_uptime(hours):
|
||
if hours < 1:
|
||
return f"{round(hours * 60)}m"
|
||
if hours < 48:
|
||
return f"{round(hours)}h"
|
||
return f"{int(hours // 24)}d {round(hours % 24)}h"
|
||
|
||
|
||
def describe_stats(stats):
|
||
stats = stats or {}
|
||
|
||
def usage(used_key, total_key):
|
||
if used_key in stats and total_key in stats:
|
||
return f"{format_size(stats[used_key])} / {format_size(stats[total_key])}"
|
||
return "-"
|
||
|
||
cpu = "-"
|
||
if "cpu_percent" in stats:
|
||
cpu = f"{stats['cpu_percent']:.0f}%"
|
||
if stats.get("cpu_cores") and stats.get("cpu_threads"):
|
||
cpu += f" · {stats['cpu_cores']}C/{stats['cpu_threads']}T"
|
||
|
||
gpu = usage("gpu_mem_used_gb", "gpu_mem_total_gb")
|
||
if gpu != "-" and stats.get("gpu_count", 1) > 1:
|
||
gpu = f"{stats['gpu_count']}× {gpu}"
|
||
|
||
uptime = format_uptime(stats["uptime_hours"]) if "uptime_hours" in stats else "-"
|
||
return [cpu, usage("mem_used_gb", "mem_total_gb"), usage("disk_used_gb", "disk_total_gb"), gpu, uptime]
|
||
|
||
|
||
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])
|
||
|
||
hardware_rows = [[labels[address(member)]] + describe_stats(member.get("stats")) for member in alive]
|
||
|
||
sections = [
|
||
title,
|
||
table(node_headers, node_rows),
|
||
table(["ROLE", "PRIORITY", "STATE", "HOLDERS"], role_rows),
|
||
table(["NODE", "CPU", "RAM", "DISK", "GPU", "UP"], hardware_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
|