This commit is contained in:
ciomek committed 2026-09-19 22:22:22 +02:00
1 parent 7f2ed59798
commit 2fa997a600
9 files changed
+107 -19

No files matched your search

+73 -9
View File
@@ -5,7 +5,7 @@ import time
from colony.core.network.requests import CommandRequest from colony.core.network.requests import CommandRequest
from colony.core.roles.unit.role import Unit from colony.core.roles.unit.role import Unit
from colony.settings import configuration from colony.settings import configuration, setting
from colony.transport.whisperer import Whisperer from colony.transport.whisperer import Whisperer
DEFAULT_TIMEOUT = 5 DEFAULT_TIMEOUT = 5
@@ -16,8 +16,8 @@ CLEAR_SCREEN = "\033[2J"
CLEAR_LINE = "\033[K" CLEAR_LINE = "\033[K"
CLEAR_BELOW = "\033[J" CLEAR_BELOW = "\033[J"
def fetch_cluster(host, port, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None): def call(host, port, command, *args, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
payload = CommandRequest.form(Unit.COMMANDS.SHOW_CLUSTER).collapse() payload = CommandRequest.form(command, *args).collapse()
resp = Whisperer(host, port, cert_file, keyfile).send(payload, timeout=timeout) 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): if not isinstance(resp, dict) or resp.get("status") != "ok" or not isinstance(resp.get("result"), dict):
@@ -26,6 +26,28 @@ def fetch_cluster(host, port, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=N
return resp["result"] return resp["result"]
def fetch_cluster(host, port, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
return call(host, port, Unit.COMMANDS.SHOW_CLUSTER, timeout=timeout, cert_file=cert_file, keyfile=keyfile)
def request_role(host, port, role, timeout=DEFAULT_TIMEOUT, cert_file=None, keyfile=None):
return call(host, port, Unit.COMMANDS.REQUEST_ROLE, role, timeout=timeout, cert_file=cert_file, keyfile=keyfile)
def resolve_on_demand_role(payload, name):
roles = payload["roles"]
match = next((role for role in roles if role["name"].lower() == name.lower()), None)
if match is None:
raise LookupError(f"unknown role {name!r}. Roles: {', '.join(role['name'] for role in roles)}")
if not match.get("on_demand"):
on_demand = [role["name"] for role in roles if role.get("on_demand")]
raise LookupError(
f"{match['name']} is assigned automatically, only on-demand roles can be added by hand"
+ (f" ({', '.join(on_demand)})" if on_demand else "")
)
return match["name"]
def format_age(seconds): def format_age(seconds):
seconds = int(seconds) seconds = int(seconds)
if seconds < 60: if seconds < 60:
@@ -89,7 +111,9 @@ def role_states(payload):
for role in payload["roles"]: for role in payload["roles"]:
holders = sorted(address(member) for member in alive if role["name"] in member["roles"]) holders = sorted(address(member) for member in alive if role["name"] in member["roles"])
required = role["required"] required = role["required"]
if required is None: if role.get("on_demand"):
state = "held" if holders else "off"
elif required is None:
state = "general" state = "general"
elif len(holders) < required: elif len(holders) < required:
state = "unmet" state = "unmet"
@@ -142,6 +166,7 @@ def render_status(payload):
else: else:
state = { state = {
"held": "ok", "held": "ok",
"off": "off (on demand)",
"unmet": f"UNMET ({len(holders)}/{role['required']})", "unmet": f"UNMET ({len(holders)}/{role['required']})",
"duplicate": f"DUPLICATE ({len(holders)}/{role['required']})", "duplicate": f"DUPLICATE ({len(holders)}/{role['required']})",
}[role["state"]] }[role["state"]]
@@ -153,26 +178,37 @@ def render_status(payload):
hardware_rows = [[labels[address(member)]] + describe_stats(member.get("stats")) for member in alive] hardware_rows = [[labels[address(member)]] + describe_stats(member.get("stats")) for member in alive]
off = [role["name"] for role in role_states(payload) if role["state"] == "off"]
sections = [ sections = [
title, title,
table(node_headers, node_rows), table(node_headers, node_rows),
table(["ROLE", "PRIORITY", "STATE", "HOLDERS"], role_rows), table(["ROLE", "PRIORITY", "STATE", "HOLDERS"], role_rows),
table(["NODE", "CPU", "RAM", "DISK", "GPU", "UP"], hardware_rows), table(["NODE", "CPU", "RAM", "DISK", "GPU", "UP"], hardware_rows),
] ]
if off:
sections.append(f"Not running (on demand): {', '.join(off)}. Start one with: colonyctl role add {off[0]}")
return "\n\n".join(sections) return "\n\n".join(sections)
def build_parser(): def build_parser():
parser = argparse.ArgumentParser(prog="colonyctl", description="Inspect a running colony node.") parser = argparse.ArgumentParser(prog="colonyctl", description="Inspect and control a running colony node.")
commands = parser.add_subparsers(dest="command", required=True) commands = parser.add_subparsers(dest="command", required=True)
status = commands.add_parser("status", help="show members, roles and hardware scores (live in a terminal)") connection = argparse.ArgumentParser(add_help=False)
status.add_argument("--host", default="127.0.0.1", help="node to ask (default: this machine)") connection.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)") connection.add_argument("--port", type=int, default=None, help="node port (default: listener.port from config)")
status.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT) connection.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
status = commands.add_parser("status", parents=[connection], help="show members, roles and hardware (live in a terminal)")
status.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="seconds between refreshes") 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("--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") status.add_argument("--json", action="store_true", help="print the raw data as JSON once and exit")
role = commands.add_parser("role", help="manage roles on a node").add_subparsers(dest="role_command", required=True)
add = role.add_parser("add", parents=[connection], help="ask the node to run an on-demand role")
add.add_argument("name", help="role name, e.g. WebView (case does not matter)")
commands.add_parser("webview", parents=[connection], help="ask the node to run the web view (same as: role add WebView)")
return parser return parser
@@ -197,11 +233,39 @@ def watch(host, port, args, out):
return 0 return 0
def add_role(args, port, name, out):
try:
payload = fetch_cluster(args.host, port, timeout=args.timeout)
role = resolve_on_demand_role(payload, name)
reply = request_role(args.host, port, role, timeout=args.timeout)
except LookupError as e:
print(f"colonyctl: {e}", file=sys.stderr)
return 1
except (OSError, ValueError, RuntimeError) as e:
print(f"colonyctl: cannot reach {args.host}:{port}: {e}", file=sys.stderr)
return 1
node = next(member for member in payload["members"] if member["self"])
if not reply.get("accepted"):
print(f"colonyctl: {node['hostname']} did not take {role}: {reply.get('reason')}", file=sys.stderr)
return 1
print(f"{role} is running on {node['hostname']} ({reply.get('reason')})", file=out)
if role == "WebView":
print(f"Web view: http://{node['ip']}:{setting('web.port')}", file=out)
return 0
def main(argv=None, out=None): def main(argv=None, out=None):
args = build_parser().parse_args(argv) args = build_parser().parse_args(argv)
out = sys.stdout if out is None else out out = sys.stdout if out is None else out
port = args.port if args.port is not None else configuration.listener.port port = args.port if args.port is not None else configuration.listener.port
if args.command == "role":
return add_role(args, port, args.name, out)
if args.command == "webview":
return add_role(args, port, "WebView", out)
if not args.raw and not args.json and out.isatty(): if not args.raw and not args.json and out.isatty():
return watch(args.host, port, args, out) return watch(args.host, port, args, out)
+7 -2
View File
@@ -10,7 +10,11 @@ HARDWARE_SCORE_KEY = {
def required_holders(role_stat): def required_holders(role_stat):
return 1 return 0 if role_stat.on_demand else 1
def max_holders(role_stat):
return None if role_stat.on_demand else required_holders(role_stat)
def assignable_roles(roles): def assignable_roles(roles):
@@ -62,6 +66,7 @@ def roles_to_release(self_key, records, roles):
key=lambda record: (record.role_since.get(name, float("inf")), record.key), key=lambda record: (record.role_since.get(name, float("inf")), record.key),
) )
keys = [record.key for record in holders] keys = [record.key for record in holders]
if self_key in keys and keys.index(self_key) >= required_holders(roles[name]): limit = max_holders(roles[name])
if limit is not None and self_key in keys and keys.index(self_key) >= limit:
release.append(name) release.append(name)
return release return release
+1
View File
@@ -36,6 +36,7 @@ class Unit(Role):
"name": name, "name": name,
"priority": roles[name].priority.name.lower(), "priority": roles[name].priority.name.lower(),
"required": None if roles[name].priority == RolePriority.GENERAL else advice.required_holders(roles[name]), "required": None if roles[name].priority == RolePriority.GENERAL else advice.required_holders(roles[name]),
"on_demand": roles[name].on_demand,
} }
for name in sorted(roles, key=lambda name: (roles[name].priority.value, name)) for name in sorted(roles, key=lambda name: (roles[name].priority.value, name))
], ],
+12 -3
View File
@@ -26,7 +26,9 @@ function roleStates(payload) {
return payload.roles.map((role) => { return payload.roles.map((role) => {
const holders = alive.filter((member) => member.roles.includes(role.name)).map(address).sort(); const holders = alive.filter((member) => member.roles.includes(role.name)).map(address).sort();
let state; let state;
if (role.required === null) { if (role.on_demand) {
state = holders.length ? "held" : "off";
} else if (role.required === null) {
state = "general"; state = "general";
} else if (holders.length < role.required) { } else if (holders.length < role.required) {
state = "unmet"; state = "unmet";
@@ -116,11 +118,17 @@ function alerts(payload) {
return out; return out;
} }
function onDemandHint(payload) {
const off = roleStates(payload).filter((role) => role.state === "off").map((role) => role.name);
if (!off.length) return "";
return "Not running (on demand): " + off.join(", ") + " · start one with: colonyctl role add " + off[0];
}
function summarize(payload) { function summarize(payload) {
const states = roleStates(payload); const states = roleStates(payload);
const alive = payload.members.filter((member) => member.alive).length; const alive = payload.members.filter((member) => member.alive).length;
const down = payload.members.length - alive; const down = payload.members.length - alive;
const assigned = states.filter((role) => role.state !== "general"); const assigned = states.filter((role) => role.state !== "general" && role.state !== "off");
const held = assigned.filter((role) => role.state !== "unmet"); const held = assigned.filter((role) => role.state !== "unmet");
let text = alive + " node" + (alive === 1 ? "" : "s") + " alive"; let text = alive + " node" + (alive === 1 ? "" : "s") + " alive";
if (down) text += " · " + down + " down"; if (down) text += " · " + down + " down";
@@ -351,6 +359,7 @@ if (typeof document !== "undefined") {
const box = document.getElementById("alerts"); const box = document.getElementById("alerts");
box.replaceChildren(); box.replaceChildren();
alerts(payload).forEach((alert) => box.appendChild(html("span", alert.text, "alert " + alert.level))); alerts(payload).forEach((alert) => box.appendChild(html("span", alert.text, "alert " + alert.level)));
document.getElementById("ondemand").textContent = onDemandHint(payload);
}; };
let latest = null; let latest = null;
@@ -395,5 +404,5 @@ if (typeof document !== "undefined") {
} }
if (typeof module !== "undefined") { if (typeof module !== "undefined") {
module.exports = { address, roleStates, layout, nodeRoles, chipLayout, chipWidth, shorten, formatAge, formatSize, formatUptime, describeStats, barLevel, alerts, summarize }; module.exports = { address, roleStates, layout, nodeRoles, chipLayout, chipWidth, shorten, formatAge, formatSize, formatUptime, describeStats, barLevel, alerts, summarize, onDemandHint };
} }
@@ -25,6 +25,7 @@
<li><span class="swatch duplicate"></span>held twice</li> <li><span class="swatch duplicate"></span>held twice</li>
<li><span class="swatch general"></span>runs on every node</li> <li><span class="swatch general"></span>runs on every node</li>
</ul> </ul>
<p id="ondemand" class="muted"></p>
</main> </main>
<footer class="muted"><span id="updated"></span></footer> <footer class="muted"><span id="updated"></span></footer>
+1 -1
View File
@@ -93,7 +93,7 @@ main { margin-top: 14px; }
.chip.duplicate text { fill: var(--text); font-weight: 600; } .chip.duplicate text { fill: var(--text); font-weight: 600; }
.chip.unmet rect { stroke: var(--unmet); stroke-dasharray: 5 3; fill: color-mix(in srgb, var(--unmet) 12%, transparent); } .chip.unmet rect { stroke: var(--unmet); stroke-dasharray: 5 3; fill: color-mix(in srgb, var(--unmet) 12%, transparent); }
.chip.unmet text { fill: var(--unmet); font-weight: 600; } .chip.unmet text { fill: var(--unmet); font-weight: 600; }
.chip.unknown rect { stroke-dasharray: 2 3; } .chip.unknown rect, .chip.off rect { stroke-dasharray: 2 3; }
.chip.more text { fill: var(--muted); } .chip.more text { fill: var(--muted); }
.center-box { fill: var(--card); stroke: var(--line); stroke-width: 1.5; } .center-box { fill: var(--card); stroke: var(--line); stroke-width: 1.5; }
+2 -1
View File
@@ -31,11 +31,12 @@ class HardwarePreferences:
class RoleStats: class RoleStats:
def __init__(self, cls, priority: RolePriority, hardware_preferences: HardwarePreferences = None): def __init__(self, cls, priority: RolePriority, hardware_preferences: HardwarePreferences = None, on_demand: bool = False):
self.name = cls.__name__ self.name = cls.__name__
self.cls = cls self.cls = cls
self.priority = priority self.priority = priority
self.hardware_preferences = hardware_preferences self.hardware_preferences = hardware_preferences
self.on_demand = on_demand
class RoleRegistry: class RoleRegistry:
+8 -2
View File
@@ -14,10 +14,16 @@ class Discover:
def listen(real_port, host="0.0.0.0", discover_port=None): def listen(real_port, host="0.0.0.0", discover_port=None):
logger = get_logger(Discover.__name__) logger = get_logger(Discover.__name__)
port = discover_port or Discover.DISCOVER_PORT
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((host, discover_port or Discover.DISCOVER_PORT)) try:
s.listen() s.bind((host, port))
s.listen()
except OSError as e:
s.close()
logger.error(f"Cannot listen for discovery on {host}:{port}: {e}. Other nodes cannot find this one by scanning")
return
while True: while True:
try: try:
+2 -1
View File
@@ -40,6 +40,7 @@ available_roles = [
{ {
HardwareType.LONGEVITY: 3, HardwareType.LONGEVITY: 3,
} }
) ),
on_demand=True,
), ),
] ]