#!/usr/bin/env python3 """Run each GitHub Actions job in a fresh exe.dev sandbox. From https://exe.dev/deploy/github-actions. Python 3 standard library only. Every few seconds, and whenever GitHub sends a workflow_job webhook, it reconciles: * jobs waiting for the LABEL runner, minus runners already waiting for a job = sandboxes to start: copy BASE_VM with cp --sandbox, register a just-in-time runner for it, start the runner; * a sandbox whose runner has finished its one job is deleted, and so are spare runners nobody needs any more and sandboxes alive for too long. GitHub's own job list is the source of truth, so a missed or forged webhook can't start sandboxes nobody asked for. It only touches VMs whose names start with JOB_PREFIX. """ import hashlib, hmac, json, os, secrets, socket, sys, threading, time import urllib.error, urllib.request from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ENV = os.environ GITHUB_API = ENV["GITHUB_API"].rstrip("/") # the integration, e.g. https://my-ci-github.int.exe.xyz EXE_API = ENV["EXE_API"].rstrip("/") # e.g. https://my-ci-exe.int.exe.xyz REPOS = ENV["REPOS"].split() # owner/repo [owner/repo ...] BASE_VM = ENV["BASE_VM"] LABEL = ENV.get("LABEL", "exe") LABELS = ["self-hosted", "linux", "x64", LABEL] JOB_PREFIX = ENV.get("JOB_PREFIX", socket.gethostname() + "-job-") MAX_RUNNERS = int(ENV.get("MAX_RUNNERS", "10")) CP_FLAGS = ENV.get("CP_FLAGS", "--cpu=2 --memory=4GB") # each job sandbox's size IDLE_SECONDS = int(ENV.get("IDLE_SECONDS", "120")) MAX_HOURS = float(ENV.get("MAX_HOURS", "6")) POLL_SECONDS = int(ENV.get("POLL_SECONDS", "30")) WEBHOOK_SECRET = ENV.get("WEBHOOK_SECRET", "").encode() PORT = int(ENV.get("PORT", "8000")) def log(*a): print(time.strftime("%H:%M:%S"), *a, flush=True) def http(method, url, body=None, headers=None): data = body if isinstance(body, (bytes, type(None))) else json.dumps(body).encode() req = urllib.request.Request(url, data=data, method=method, headers=headers or {}) req.add_header("User-Agent", "exe-ci-dispatcher") with urllib.request.urlopen(req, timeout=60) as r: raw = r.read() return json.loads(raw) if raw.strip() else None def gh(method, path, body=None): return http(method, GITHUB_API + path, body, { "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", "Content-Type": "application/json"}) def exe(cmd): """Run an exe.dev command over the HTTPS API; returns its JSON.""" return http("POST", EXE_API + "/exec", cmd.encode()) def exe_ssh(vm, command): """Run a command on a VM through the API; returns its exit code.""" req = urllib.request.Request(EXE_API + "/exec", method="POST", data=f"ssh {vm} {command}".encode(), headers={"Accept": "application/x-ndjson", "User-Agent": "exe-ci-dispatcher"}) code = None with urllib.request.urlopen(req, timeout=60) as r: for line in r: code = json.loads(line).get("exit_code", code) return code def queued_jobs(repo): """Jobs in repo waiting for a runner with our label.""" n = 0 for status in ("queued", "in_progress"): # a running workflow can queue more jobs runs = gh("GET", f"/repos/{repo}/actions/runs?status={status}&per_page=100") for run in runs["workflow_runs"]: jobs = gh("GET", f"/repos/{repo}/actions/runs/{run['id']}/jobs?per_page=100") n += sum(1 for j in jobs["jobs"] if j["status"] == "queued" and LABEL in j["labels"] and set(j["labels"]) <= set(LABELS)) return n def runners(repo): r = gh("GET", f"/repos/{repo}/actions/runners?per_page=100") return {x["name"]: x for x in r["runners"] if x["name"].startswith(JOB_PREFIX)} def job_vms(): vms = exe("ls --json")["vms"] return {v["vm_name"]: v for v in vms if v["vm_name"].startswith(JOB_PREFIX)} def age_seconds(vm): t = datetime.fromisoformat(vm["created_at"].replace("Z", "+00:00")) return time.time() - t.timestamp() def launch(repo): name = JOB_PREFIX + secrets.token_hex(4) t0 = time.time() exe(f"cp {BASE_VM} {name} --sandbox {CP_FLAGS}".strip()) try: jit = gh("POST", f"/repos/{repo}/actions/runners/generate-jitconfig", { "name": name, "runner_group_id": 1, "labels": LABELS, "work_folder": "_work"})["encoded_jit_config"] start = ("'setsid nohup /home/exedev/actions-runner/run.sh --jitconfig " f"{jit} > /home/exedev/runner.log 2>&1 < /dev/null & echo started'") for _ in range(30): # the copy takes a moment to answer try: if exe_ssh(name, start) == 0: break except urllib.error.URLError: pass time.sleep(1) else: raise RuntimeError("runner did not start") log(f"started {name} for {repo} in {time.time() - t0:.1f}s") except Exception as e: log(f"launch {name} failed: {e}; deleting it") remove(name, repo) def remove(name, repo=None, runner=None): try: if runner and repo: gh("DELETE", f"/repos/{repo}/actions/runners/{runner['id']}") except urllib.error.HTTPError as e: log(f"deregister {name}: {e}") try: exe(f"rm {name}") log(f"deleted {name}") except urllib.error.HTTPError as e: log(f"rm {name}: {e}") lock = threading.Lock() wake = threading.Event() finished = set() # runners GitHub says completed a job; their VMs go next def reconcile(): with lock: vms = job_vms() owned, idle, threads = {}, {}, [] for repo in REPOS: for name, r in runners(repo).items(): owned[name] = (repo, r) queued = {repo: queued_jobs(repo) for repo in REPOS} alive = len(vms) for name, vm in vms.items(): repo, r = owned.get(name, (None, None)) if r is None or name in finished: # Its runner ran its one job, or it never registered. remove(name, repo, r) alive -= 1 elif age_seconds(vm) > MAX_HOURS * 3600: remove(name, repo, r) alive -= 1 elif not r["busy"]: idle.setdefault(repo, []).append((age_seconds(vm), name, r)) finished.intersection_update(vms) room = MAX_RUNNERS - alive for repo in REPOS: waiting = sorted(idle.get(repo, []), reverse=True) # oldest first # More runners waiting than jobs (a job was cancelled, or another # runner took it): delete the spares once they've had their chance. while len(waiting) > queued[repo] and waiting[0][0] > IDLE_SECONDS: _, name, r = waiting.pop(0) remove(name, repo, r) room += 1 want = max(min(queued[repo] - len(waiting), room), 0) for _ in range(want): t = threading.Thread(target=launch, args=(repo,)) t.start() threads.append(t) room -= want for t in threads: t.join() def loop(): while True: try: reconcile() except Exception as e: log(f"reconcile: {e!r}") wake.wait(POLL_SECONDS) wake.clear() class Hook(BaseHTTPRequestHandler): def do_POST(self): body = self.rfile.read(int(self.headers.get("Content-Length", 0))) want = "sha256=" + hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest() if not WEBHOOK_SECRET or not hmac.compare_digest( want, self.headers.get("X-Hub-Signature-256", "")): self.send_response(401) self.end_headers() return if self.headers.get("X-GitHub-Event") == "workflow_job": ev = json.loads(body) job = ev["workflow_job"] runner = job.get("runner_name") or "" log(f"webhook: {job['name']} {ev['action']} {runner}") if ev["action"] == "completed" and runner.startswith(JOB_PREFIX): finished.add(runner) wake.set() self.send_response(204) self.end_headers() def log_message(self, *a): pass if __name__ == "__main__": log(f"dispatching {LABEL} jobs for {' '.join(REPOS)} to copies of {BASE_VM}") threading.Thread(target=loop, daemon=True).start() ThreadingHTTPServer(("", PORT), Hook).serve_forever()