61 lines
5.5 KiB
Python
61 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""The fleet's orchestrator loop (runs on the Mac in the background, one pass a minute): advances every live box
|
|
through its stages without a human, pulls the results of each finished stage into ~/Desktop/fleet/<instance>/, runs the
|
|
collector after a matrix or an Ember ladder lands, and refreshes the fleet page every 5 minutes.
|
|
|
|
phase 1: setup_done -> box-matrix.sh -> matrix_done -> box-ember.sh -> ember_done -> box-prover.sh (joins phase 2)
|
|
phase 2: setup_done -> box-prover.sh
|
|
setup_failed: retried once (the toolchain CDN class), then marked failed and left for a human
|
|
Writes ~/Desktop/fleet/autorun.log; every change is a line there and in the page log.
|
|
"""
|
|
import json, os, sys, time, subprocess, datetime
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import fleet
|
|
ROOT = fleet.ROOT; LOG = os.path.join(ROOT, "autorun.log")
|
|
def now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
def log(s, page=False):
|
|
line = f"{now()} {s}"; open(LOG, "a").write(line + "\n"); print(line, flush=True)
|
|
if page: subprocess.run([sys.executable, os.path.join(fleet.HERE, "page.py"), "log", s])
|
|
def probe(b):
|
|
rc, out, err = fleet.ssh(b, "tail -1 /root/fleet/setup.log 2>/dev/null | grep -o '^RESULT setup_[a-z]*'; grep -c . /root/fleet/out/matrix.log 2>/dev/null; grep -o '^RESULT matrix_[a-z]*' /root/fleet/out/matrix.log 2>/dev/null | tail -1; grep -c . /root/fleet/out/ember.log 2>/dev/null; grep -o '^RESULT ember_done' /root/fleet/out/ember.log 2>/dev/null | tail -1; pgrep -c -f '[b]ash in/box-|[p]ython3 -u /root/fleet/in/box-prover'; grep -E '^RESULT (point|miner|step|choice|claim|submitted|paid|segment_refused|seg [0-9]+ (chain|shards))' /root/fleet/out/matrix.log /root/fleet/out/ember.log /root/fleet/out/prover.log 2>/dev/null | tail -1 | cut -c1-200", timeout=40)
|
|
if rc != 0 and not out.strip(): return None
|
|
l = out.split("\n")
|
|
g = lambda i: l[i].strip() if i < len(l) else ""
|
|
return {"setup": g(0), "matrix_lines": g(1), "matrix": g(2), "ember_lines": g(3), "ember": g(4), "running": g(5), "last": g(6)}
|
|
def start(b, script, env=""):
|
|
fleet.run(script, [b["label"]]) if not env else fleet.run_env(script, [b["label"]], env)
|
|
last_publish = 0; retried = set()
|
|
while True:
|
|
reg = fleet.load(); changed = False
|
|
live = [(iid, b) for iid, b in reg.items() if b.get("state") not in ("destroyed", "failed") and b.get("ssh_ok") and b.get("phase") != "5"]
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
with ThreadPoolExecutor(max_workers=16) as ex: probes = dict(zip([i for i, _ in live], ex.map(lambda x: probe(x[1]), live)))
|
|
for iid, b in live:
|
|
p = probes.get(iid)
|
|
if p is None: continue
|
|
stage = b.get("stage", "setup")
|
|
if p["last"]: fleet.patch(iid, last_line=p["last"])
|
|
if p["running"] not in ("0", ""): # a stage script is running on the box: never start another (the 5070 ran two matrices at once, 12:23Z)
|
|
if stage == "setup" and p["setup"] == "RESULT setup_done": fleet.patch(iid, stage="matrix" if b["phase"] == "1" else "prover", state="running")
|
|
continue
|
|
if stage == "setup":
|
|
if p["setup"] == "RESULT setup_done":
|
|
nxt = "matrix" if b["phase"] == "1" else "prover"
|
|
fleet.patch(iid, stage=nxt, state="running", doing=("phase 1 matrix: idle, miner, stock server, patched server alone and beside the miner" if nxt == "matrix" else "phase 2: node + miner + segment prover on the devnet"))
|
|
fleet.run("box-matrix.sh" if nxt == "matrix" else "box-prover.sh", [b["label"]]); log(f"{b['label']}: setup done, {nxt} started", page=True); changed = True
|
|
elif p["setup"] == "RESULT setup_failed":
|
|
if iid not in retried: retried.add(iid); fleet.setup([b["label"]]); log(f"{b['label']}: setup failed once, retried")
|
|
else: fleet.patch(iid, state="failed", doing="setup failed twice; see setup.log"); log(f"{b['label']}: setup failed twice", page=True)
|
|
elif stage == "matrix":
|
|
if p["matrix"] == "RESULT matrix_done":
|
|
fleet.pull([b["label"]]); subprocess.run([sys.executable, os.path.join(fleet.HERE, "collect.py")], capture_output=True)
|
|
fleet.patch(iid, stage="ember", doing="Ember two-knob ladder: 6 power steps, 4 clock caps, 75 s each"); fleet.run("box-ember.sh", [b["label"]]); log(f"{b['label']}: matrix done ({p['last'][:80]}), Ember ladder started", page=True); changed = True
|
|
elif p["matrix"] == "RESULT matrix_failed":
|
|
fleet.pull([b["label"]]); fleet.patch(iid, stage="ember", doing="matrix failed (see matrix.log); Ember ladder started"); fleet.run("box-ember.sh", [b["label"]]); log(f"{b['label']}: matrix FAILED ({p['last'][:80]}), Ember started", page=True)
|
|
elif stage == "ember":
|
|
if p["ember"] == "RESULT ember_done" or (p["ember_lines"].isdigit() and int(p["ember_lines"]) > 0 and "ember_failed" in p["last"]):
|
|
fleet.pull([b["label"]]); subprocess.run([sys.executable, os.path.join(fleet.HERE, "collect.py")], capture_output=True)
|
|
fleet.patch(iid, stage="prover", phase="2", doing="phase 2: node + miner + segment prover on the devnet"); fleet.run("box-prover.sh", [b["label"]]); log(f"{b['label']}: Ember done ({p['last'][:80]}), prover started", page=True); changed = True
|
|
if time.time() - last_publish > 300 or changed:
|
|
subprocess.run([sys.executable, os.path.join(fleet.HERE, "page.py"), "publish"], capture_output=True); last_publish = time.time()
|
|
time.sleep(60)
|