#!/usr/bin/env python3 """Probe each team receiver's /check/ directly and report SLA. Probes are SEQUENTIAL per team on purpose: the team receivers are sync Flask apps, so 8 concurrent /check requests make them time out and report false failures. Keep max_workers=1. This is still far faster than the panel's /api/scoreboard, which probes every team on one shared refresher thread. python3 panel/sla_probe.py # all teams python3 panel/sla_probe.py 1 2 # specific teams """ import base64 import json import sys import urllib.error import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import teams as orch def check(recv_port, au, ap, name): url = f"http://127.0.0.1:{recv_port}/check/{name}" tok = base64.b64encode(f"{au}:{ap}".encode()).decode() req = urllib.request.Request(url, headers={"Authorization": f"Basic {tok}"}) try: with urllib.request.urlopen(req, timeout=30) as r: body = json.loads(r.read().decode()) return name, bool(body.get("success")), "" except urllib.error.HTTPError as e: return name, False, f"HTTP {e.code}" except Exception as e: return name, False, str(e)[:60] def main(): want = [int(a) for a in sys.argv[1:]] teams = [t for t in orch.list_teams() if not want or t["index"] in want] enabled = [c["name"] for c in orch.enabled_challenges()] grand_ok = grand_all = 0 for t in teams: idx = t["index"] port = t["ports"]["receiver"] au, ap = t.get("admin_user", ""), t.get("admin_pass", "") res = [check(port, au, ap, n) for n in enabled] up = [n for n, ok, _ in res if ok] down = [(n, e) for n, ok, e in res if not ok] grand_ok += len(up); grand_all += len(res) print(f"team{idx} ({t.get('label')}) SLA {len(up)}/{len(res)}") if down: for n, e in down: print(f" DOWN {n}: {e}") print(f"\nTOTAL {grand_ok}/{grand_all} " f"({100.0 * grand_ok / grand_all if grand_all else 0:.1f}%)") if __name__ == "__main__": main()