fuzzyworld

fuzzyworld-tools / files / runner / fw-runner.py

fw-runner.py

39.1 KB · history · blame

1#!/usr/bin/env python3
2"""fw-runner: CI for the repos on DreamHost, in the style of a GitLab runner.
3
4 fw-runner.py [--once] [--keep] take queued jobs and run them (cron or launchd runs this every 2 minutes)
5 fw-runner.py --plan <dir> [ref] print the jobs a checkout would run, and run nothing
6
7A push to DreamHost queues a job (server/hooks/post-receive). This runner claims jobs over
8ssh with a key that may only use server/runner-api, clones the commit, runs the pipeline in
9throwaway Docker containers, and uploads the logs and pipeline.json. DreamHost turns those
10into static pages (code.fuzzyworld.net/<repo>/ci/). For apps, a successful pipeline on the
11default branch is followed by screenshots of the build (shoot.py in a Playwright container).
12
13Pipelines come from .fw-ci.yml, written in a subset of GitLab CI's format (see SUPPORTED_JOB_KEYS);
14a repo not yet moved over can still use the old name, .gitlab-ci.yml. Without either, a repo
15whose package.json has a "fuzzyworld" section gets one job: `fw deploy --here --dry-run`
16(install, checks, build, and the output checks), and other node repos get their
17typecheck/lint/test scripts.
18
19Configuration (environment, or <dir>/runner.env next to this script):
20 FW_RUNNER_NAME name shown on the CI pages (default: the host name)
21 FW_RUNNER_HOST ssh destination of DreamHost (default fuzzygamer@iad1-shared-b7-30.dreamhost.com)
22 FW_RUNNER_KEY ssh key allowed only to run server/runner-api (default <dir>/.ssh/id_ed25519)
23 FW_RUNNER_WORK working directory for clones (default <dir>/work)
24 FW_RUNNER_SECRETS directory of <repo>.env files, given to default-branch and tag jobs only (default <dir>/secrets)
25 FW_RUNNER_CPUS / FW_RUNNER_MEMORY container limits (default 4 / 4g)
26 FW_RUNNER_TAGS what this runner can do, comma-separated (default docker,chrome). A job whose `tags:` or
27 screenshots need something else is given back to the queue for a runner that has it. The
28 tag `secrets` means "this repo's secrets": a default-branch or tag job tagged `secrets` runs
29 only on a runner that has <repo>.env.
30 FW_RUNNER_POLITE 1 for a runner on a laptop: claim jobs only on AC power, with the CPU mostly idle and
31 Docker running; containers get a low CPU share. Checked before each job, never during one.
32 FW_RUNNER_MIN_IDLE percent of CPU that must be idle for a polite runner to claim a job (default 50)
33"""
34
35import fcntl
36import fnmatch
37import json
38import os
39import re
40import shlex
41import shutil
42import socket
43import subprocess
44import sys
45import threading
46import time
47
48HERE = os.path.dirname(os.path.abspath(__file__))
49# The pipeline file: our own name first; GitLab's name still works for repos not moved over yet.
50CI_FILES = (".fw-ci.yml", ".gitlab-ci.yml")
51
52
53def load_env_file(path):
54 if os.path.isfile(path):
55 for line in open(path):
56 line = line.strip()
57 if line and not line.startswith("#") and "=" in line:
58 k, v = line.split("=", 1)
59 os.environ.setdefault(k.strip(), v.strip().strip('"'))
60
61
62load_env_file(os.path.join(HERE, "runner.env"))
63RUNNER = os.environ.get("FW_RUNNER_NAME") or socket.gethostname().split(".")[0]
64HOST = os.environ.get("FW_RUNNER_HOST", "fuzzygamer@iad1-shared-b7-30.dreamhost.com")
65KEY = os.environ.get("FW_RUNNER_KEY", os.path.join(HERE, ".ssh", "id_ed25519"))
66WORK = os.environ.get("FW_RUNNER_WORK", os.path.join(HERE, "work"))
67SECRETS = os.environ.get("FW_RUNNER_SECRETS", os.path.join(HERE, "secrets"))
68CPUS = os.environ.get("FW_RUNNER_CPUS", "4")
69MEMORY = os.environ.get("FW_RUNNER_MEMORY", "4g")
70TAGS = {t for t in re.split(r"[,\s]+", os.environ.get("FW_RUNNER_TAGS", "docker,chrome")) if t}
71POLITE = os.environ.get("FW_RUNNER_POLITE", "") not in ("", "0", "false")
72MIN_IDLE = float(os.environ.get("FW_RUNNER_MIN_IDLE", "50"))
73APPS_URL = "https://apps.fuzzyworld.net"
74CODE_URL = "https://code.fuzzyworld.net"
75
76DEFAULT_IMAGE = "node:22-bookworm" # for jobs that name no image (GitLab leaves this to the runner)
77# Playwright's image has the browsers but not the Python package; fw-runner adds it once (shoot_image()).
78PLAYWRIGHT = "1.49.1"
79SHOOT_IMAGE = f"fw-shoot:{PLAYWRIGHT}"
80# `fw deploy --dry-run` from the published fuzzyworld-tools, whether or not the app depends on it.
81FW_DEPLOY_CHECK = "npx --yes --package=git+https://code.fuzzyworld.net/fuzzyworld-tools.git#semver:^1.0.0 fw deploy --here --dry-run"
82BEAT_EVERY = 30 # seconds between heartbeats; DreamHost gives a job back after ~2 minutes without one
83BEAT_TIMEOUT = 20 # seconds one heartbeat may take: the lease is ~2 minutes, so a hung ssh must not eat it
84DOCKER_INFO_TIMEOUT = 20 # seconds `docker info` may take (a wedged Docker daemon hangs it)
85LOG_EVERY = 15 # seconds between uploads of a running job's log (live CI pages)
86JOB_TIMEOUT = 30 * 60 # seconds, unless the job sets `timeout`
87MAX_LOG = 20 * 1024 * 1024 # runner-api refuses bigger files
88
89SSH = ["ssh", "-i", KEY, "-o", "IdentitiesOnly=yes", "-o", "BatchMode=yes",
90 "-o", "StrictHostKeyChecking=accept-new", "-o", f"UserKnownHostsFile={os.path.join(HERE, '.ssh', 'known_hosts')}"]
91
92# Keys of a pipeline job that this runner understands. Unsupported keys stop the job with a
93# message saying so: a job is never run with part of its definition silently ignored. Keys in
94# IGNORED_JOB_KEYS don't change what a job does here (every job of a pipeline shares one workspace,
95# so artifacts are already there for later stages).
96SUPPORTED_JOB_KEYS = {"stage", "image", "script", "before_script", "after_script", "variables", "rules",
97 "only", "except", "when", "allow_failure", "extends", "timeout", "needs", "tags"}
98IGNORED_JOB_KEYS = {"artifacts", "cache", "dependencies", "environment", "interruptible", "retry",
99 "coverage", "resource_group", "pages", "id_tokens", "publish"}
100TOP_LEVEL_KEYS = {"stages", "variables", "image", "default", "workflow", "before_script", "after_script", "cache"}
101RESERVED = TOP_LEVEL_KEYS | {"include", "types"}
102
103
104class Unsupported(Exception):
105 pass
106
107
108def log(msg):
109 print(f"{time.strftime('%F %T')} {msg}", flush=True)
110
111
112def now_iso():
113 return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
114
115
116# ---------------------------------------------------------------- DreamHost API
117
118def api(*args, data=None, timeout=300):
119 r = subprocess.run(SSH + [HOST, " ".join(args)], input=data, capture_output=True, timeout=timeout)
120 if r.returncode != 0:
121 raise RuntimeError(f"runner-api {args[0]}: {r.stderr.decode(errors='replace').strip()}")
122 return r.stdout
123
124
125def clone_url(repo):
126 return f"{HOST}:git/{repo}.git"
127
128
129def put(repo, sha, name, data):
130 if isinstance(data, str):
131 data = data.encode()
132 if len(data) > MAX_LOG:
133 data = b"(start of log cut: it was over 20 MB)\n" + data[-(MAX_LOG - 100):]
134 api("put", repo, sha, name, data=data)
135
136
137# ---------------------------------------------------------------- CI variables and expressions
138
139def slug(s):
140 return re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-")[:63].rstrip("-")
141
142
143def expand(value, env):
144 """$VAR and ${VAR} from env; unknown variables become empty, as in GitLab."""
145 if not isinstance(value, str):
146 return value
147 return re.sub(r"\$\{(\w+)\}|\$(\w+)", lambda m: env.get(m.group(1) or m.group(2), ""), value)
148
149
150_TOKEN = re.compile(r"""\s*(?:(?P<op>==|!=|=~|!~|&&|\|\||\(|\))|(?P<var>\$\{?\w+\}?)|(?P<str>"(?:[^"\\]|\\.)*"|'(?:[^'\\]|\\.)*')
151 |(?P<re>/(?:[^/\\]|\\.)*/[a-z]*)|(?P<null>null)|(?P<bad>\S))""", re.X)
152
153
154def evaluate(expr, env):
155 """A GitLab `rules:if` expression: variables, strings, null, regexes, == != =~ !~ && || and ()."""
156 tokens = []
157 for m in _TOKEN.finditer(expr):
158 if m.group("bad"):
159 raise Unsupported(f"can't read the expression {expr!r}")
160 kind = next(k for k in ("op", "var", "str", "re", "null") if m.group(k) is not None)
161 tokens.append((kind, m.group(kind)))
162 pos = 0
163
164 def peek():
165 return tokens[pos] if pos < len(tokens) else (None, None)
166
167 def take():
168 nonlocal pos
169 pos += 1
170 return tokens[pos - 1]
171
172 def value():
173 kind, v = take()
174 if kind == "op" and v == "(":
175 r = disj()
176 if take() != ("op", ")"):
177 raise Unsupported(f"unbalanced parentheses in {expr!r}")
178 return r
179 if kind == "var":
180 name = v.strip("${}")
181 return env.get(name)
182 if kind == "str":
183 return v[1:-1]
184 if kind == "re":
185 body, flags = v[1:].rsplit("/", 1)
186 return re.compile(body, re.I if "i" in flags else 0)
187 if kind == "null":
188 return None
189 raise Unsupported(f"can't read the expression {expr!r}")
190
191 def comparison():
192 left = value()
193 kind, op = peek()
194 if kind == "op" and op in ("==", "!=", "=~", "!~"):
195 take()
196 right = value()
197 if op in ("==", "!="):
198 return (left == right) == (op == "==")
199 matched = bool(isinstance(right, re.Pattern) and left is not None and right.search(left))
200 return matched if op == "=~" else not matched
201 return bool(left)
202
203 def conj():
204 r = comparison()
205 while peek() == ("op", "&&"):
206 take()
207 r = comparison() and r
208 return r
209
210 def disj():
211 r = conj()
212 while peek() == ("op", "||"):
213 take()
214 r = conj() or r
215 return r
216
217 result = disj()
218 if pos != len(tokens):
219 raise Unsupported(f"can't read the expression {expr!r}")
220 return result
221
222
223def rules_when(rules, env, src):
224 """The `when` of the first matching rule, or None if no rule matches."""
225 for rule in rules:
226 if not isinstance(rule, dict):
227 raise Unsupported(f"can't read rule {rule!r}")
228 unknown = set(rule) - {"if", "when", "exists", "changes", "allow_failure", "variables"}
229 if unknown:
230 raise Unsupported(f"rules: {', '.join(sorted(unknown))} not supported")
231 if "if" in rule and not evaluate(rule["if"], env):
232 continue
233 if "exists" in rule:
234 pats = rule["exists"] if isinstance(rule["exists"], list) else rule["exists"].get("paths", [])
235 if not any(glob_exists(src, expand(p, env)) for p in pats):
236 continue
237 # `changes` compares against the previous push in GitLab; every push is treated as a change here.
238 return rule.get("when", "on_success")
239 return None
240
241
242def glob_exists(src, pattern):
243 for root, dirs, files in os.walk(src):
244 dirs[:] = [d for d in dirs if d != ".git"]
245 for f in files:
246 rel = os.path.relpath(os.path.join(root, f), src)
247 if fnmatch.fnmatch(rel, pattern):
248 return True
249 return False
250
251
252def only_except_match(spec, env):
253 """Old-style `only`/`except` with refs: branches, tags, branch names and /regexes/."""
254 refs = spec if isinstance(spec, list) else (spec or {}).get("refs", [])
255 if isinstance(spec, dict) and set(spec) - {"refs"}:
256 raise Unsupported(f"only/except: {', '.join(sorted(set(spec) - {'refs'}))} not supported")
257 name = env.get("CI_COMMIT_REF_NAME", "")
258 for r in refs:
259 if r == "branches" and env.get("CI_COMMIT_BRANCH"):
260 return True
261 if r == "tags" and env.get("CI_COMMIT_TAG"):
262 return True
263 if r in ("pushes", "web", "api"):
264 return True
265 m = re.fullmatch(r"/(.*)/(i?)", r) if isinstance(r, str) else None
266 if m and re.search(m.group(1), name, re.I if m.group(2) else 0):
267 return True
268 if r == name:
269 return True
270 return False
271
272
273# ---------------------------------------------------------------- pipeline definition
274
275def merge(base, over):
276 out = dict(base)
277 for k, v in over.items():
278 out[k] = merge(out[k], v) if isinstance(v, dict) and isinstance(out.get(k), dict) else v
279 return out
280
281
282def resolve_extends(name, defs, seen=()):
283 job = defs[name]
284 ext = job.get("extends")
285 if not ext:
286 return job
287 base = {}
288 for e in ([ext] if isinstance(ext, str) else ext):
289 if e in seen or e not in defs:
290 raise Unsupported(f"{name}: extends {e!r}, which doesn't exist or loops")
291 base = merge(base, resolve_extends(e, defs, seen + (name,)))
292 return merge(base, {k: v for k, v in job.items() if k != "extends"})
293
294
295def script_lines(v):
296 if v is None:
297 return []
298 items = v if isinstance(v, list) else [v]
299 out = []
300 for i in items:
301 out.extend(script_lines(i) if isinstance(i, list) else [str(i)])
302 return out
303
304
305def ci_file(src):
306 return next((f for f in CI_FILES if os.path.exists(os.path.join(src, f))), None)
307
308
309def yaml_pipeline(src, env):
310 """[(job dict)] from the pipeline file, in run order, or raise Unsupported."""
311 import yaml # PyYAML; on TrueNAS it is part of the system Python
312
313 with open(os.path.join(src, ci_file(src))) as f:
314 doc = yaml.safe_load(f) or {}
315 if "include" in doc:
316 raise Unsupported(f"include: is not supported; put the jobs in {ci_file(src)} itself")
317 variables = {k: str(expand(v.get("value", "") if isinstance(v, dict) else v, env))
318 for k, v in (doc.get("variables") or {}).items()}
319 env = {**env, **variables}
320 wf = (doc.get("workflow") or {}).get("rules")
321 if wf is not None and rules_when(wf, env, src) in (None, "never"):
322 return [], "workflow:rules: no pipeline for this push"
323 default = doc.get("default") or {}
324 stages = doc.get("stages") or ["build", "test", "deploy"]
325 stages = [".pre"] + [s for s in stages if s not in (".pre", ".post")] + [".post"]
326 defs = {k: v for k, v in doc.items() if isinstance(v, dict) and k not in RESERVED}
327 jobs = []
328 for name in defs:
329 if name.startswith("."):
330 continue # hidden: only used through extends
331 job = resolve_extends(name, defs)
332 unknown = set(job) - SUPPORTED_JOB_KEYS - IGNORED_JOB_KEYS
333 jenv = {**env, **{k: str(expand(v, env)) for k, v in (job.get("variables") or {}).items()},
334 "CI_JOB_NAME": name, "CI_JOB_STAGE": job.get("stage", "test")}
335 if "rules" in job:
336 when = rules_when(job["rules"], jenv, src)
337 else:
338 only = job.get("only", ["branches", "tags"])
339 when = job.get("when", "on_success") if only_except_match(only, jenv) and not (
340 "except" in job and only_except_match(job["except"], jenv)) else None
341 if when in (None, "never"):
342 continue
343 stage = job.get("stage", "test")
344 image = job.get("image", default.get("image", doc.get("image", DEFAULT_IMAGE)))
345 if isinstance(image, dict):
346 if set(image) - {"name"}:
347 raise Unsupported(f"{name}: image: {', '.join(sorted(set(image) - {'name'}))} not supported")
348 image = image["name"]
349 before = job["before_script"] if "before_script" in job else default.get("before_script", doc.get("before_script"))
350 after = job["after_script"] if "after_script" in job else default.get("after_script", doc.get("after_script"))
351 jobs.append({
352 "name": name, "stage": stage, "image": expand(image, jenv), "env": jenv,
353 "script": script_lines(before) + script_lines(job.get("script")),
354 "after": script_lines(after), "when": when, "allow_failure": bool(job.get("allow_failure", False)),
355 "timeout": parse_timeout(job.get("timeout")),
356 "tags": [str(t) for t in (job.get("tags") or [])],
357 "unsupported": sorted(unknown),
358 "note": "GitLab Pages: the build runs as a check here; `fw deploy` publishes apps." if name == "pages" else "",
359 })
360 order = {s: i for i, s in enumerate(stages)}
361 missing = sorted({j["stage"] for j in jobs if j["stage"] not in order})
362 if missing:
363 raise Unsupported(f"stage(s) {', '.join(missing)} not listed under stages:")
364 jobs.sort(key=lambda j: order[j["stage"]])
365 return jobs, "" if jobs else "no job's rules or only/except match this push"
366
367
368def parse_timeout(v):
369 if not v:
370 return JOB_TIMEOUT
371 total = 0
372 for n, unit in re.findall(r"(\d+)\s*(h|hours?|m|min(?:utes?)?|s|sec(?:onds?)?)", str(v)):
373 total += int(n) * {"h": 3600, "m": 60, "s": 1}[unit[0]]
374 return total or JOB_TIMEOUT
375
376
377def package_pipeline(src, env):
378 """The pipeline for a repo without a pipeline file, from package.json."""
379 try:
380 with open(os.path.join(src, "package.json")) as f:
381 pkg = json.load(f)
382 except (OSError, ValueError):
383 return [], "no .fw-ci.yml, and no readable package.json"
384 if (pkg.get("fuzzyworld") or {}).get("app"):
385 # The same steps `fw deploy` takes, without uploading: install, prebuild, checks, build, output checks.
386 return [job("deploy-check", "test", ["corepack enable", FW_DEPLOY_CHECK], env)], ""
387 scripts = pkg.get("scripts") or {}
388 checks = [s for s in ("typecheck", "lint", "test") if scripts.get(s) and "no test specified" not in scripts[s]]
389 if not checks:
390 return [], "no .fw-ci.yml, and package.json has no typecheck, lint or test script"
391 install = install_command(src)
392 pm = install.split()[0]
393 return [job("checks", "test", ["corepack enable", install] + [f"{pm} run {c}" for c in checks], env)], ""
394
395
396def install_command(src):
397 if os.path.exists(os.path.join(src, "pnpm-lock.yaml")):
398 return "pnpm install --frozen-lockfile"
399 if os.path.exists(os.path.join(src, "yarn.lock")):
400 return "yarn install --frozen-lockfile"
401 if os.path.exists(os.path.join(src, "package-lock.json")):
402 return "npm ci"
403 return "npm install"
404
405
406def job(name, stage, script, env):
407 return {"name": name, "stage": stage, "image": DEFAULT_IMAGE, "env": {**env, "CI_JOB_NAME": name, "CI_JOB_STAGE": stage},
408 "script": script, "after": [], "when": "on_success", "allow_failure": False,
409 "timeout": JOB_TIMEOUT, "tags": [], "unsupported": [], "note": ""}
410
411
412def ci_env(j, src):
413 ref, sha, repo, default = j["ref"], j["sha"], j["repo"], j["default_branch"]
414 name = ref.split("/", 2)[2]
415 msg = git(src, "log", "-1", "--format=%B", sha)
416 env = {
417 "CI": "true", "FUZZYWORLD_CI": "true", "CI_SERVER_NAME": "fuzzyworld", "CI_RUNNER_DESCRIPTION": RUNNER,
418 "CI_PROJECT_NAME": repo, "CI_PROJECT_PATH": repo, "CI_PROJECT_DIR": f"/builds/{repo}",
419 "CI_PROJECT_URL": f"{CODE_URL}/{repo}", "CI_REPOSITORY_URL": f"{CODE_URL}/{repo}.git",
420 "CI_DEFAULT_BRANCH": default, "CI_COMMIT_SHA": sha, "CI_COMMIT_SHORT_SHA": sha[:8],
421 "CI_COMMIT_REF_NAME": name, "CI_COMMIT_REF_SLUG": slug(name),
422 "CI_COMMIT_MESSAGE": msg, "CI_COMMIT_TITLE": msg.split("\n", 1)[0],
423 "CI_PIPELINE_SOURCE": "push", "CI_PIPELINE_ID": j["id"], "CI_PAGES_URL": f"{APPS_URL}/{repo}",
424 }
425 if ref.startswith("refs/tags/"):
426 env["CI_COMMIT_TAG"] = name
427 else:
428 env["CI_COMMIT_BRANCH"] = name
429 return env
430
431
432# ---------------------------------------------------------------- tags
433
434def release(job_id, tries=3):
435 """Give a claimed job back. A network blip must not drop the job into the generic error path (which would
436 upload an "error" result for a job that never ran): retry, then give up quietly. Unreleased, the job goes
437 back in the queue by itself when the heartbeat has been quiet for ~2 minutes."""
438 for n in range(tries):
439 try:
440 api("release", job_id, RUNNER)
441 return
442 except Exception as e:
443 log(f"{job_id}: release failed ({e})" + ("; retrying" if n + 1 < tries else "; the lease will expire"))
444 if n + 1 < tries:
445 time.sleep(5)
446
447
448class Declined(Exception):
449 """This runner can't run the job; it went back in the queue for another."""
450
451
452def missing_tags(j, jobs, src, protected, has_secrets):
453 """What this pipeline needs that this runner lacks: job `tags:`, `chrome` if it will take screenshots,
454 and `secrets` (the repo's <repo>.env) for a default-branch or tag job tagged `secrets`."""
455 need = {t for x in jobs for t in x["tags"]}
456 if protected and not j["ref"].startswith("refs/tags/") and shot_plan(src):
457 need.add("chrome")
458 # `secrets` is not a runner tag: the runner has them if it has the repo's file, and a job off the
459 # default branch never gets them, so for that job the tag is met by any runner.
460 wants_secrets = "secrets" in need
461 need.discard("secrets")
462 return sorted(need - TAGS) + (["secrets"] if wants_secrets and protected and not has_secrets else [])
463
464
465# ---------------------------------------------------------------- running
466
467def git(cwd, *args):
468 return subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True, check=True).stdout.strip()
469
470
471def docker_run(name, image, workdir, env, script, timeout, logf, env_file=None, extra=(), progress=None):
472 """Run a shell script in a fresh container; returns (exit code, seconds). Output goes to logf;
473 progress() is called every LOG_EVERY seconds while it runs."""
474 cmd = ["docker", "run", "--rm", "--name", name, "--cpus", CPUS, "--memory", MEMORY,
475 "-v", f"{workdir}:/builds/{env['CI_PROJECT_NAME']}", "-w", f"/builds/{env['CI_PROJECT_NAME']}", *extra]
476 if POLITE:
477 cmd[2:2] = ["--cpu-shares", "128"] # the default is 1024: other containers and the Docker VM's own work win
478 for k, v in env.items():
479 cmd += ["-e", f"{k}={v}"]
480 if env_file:
481 cmd += ["--env-file", env_file]
482 lines = ["set -e"] + [f"printf '%s\\n' {shlex.quote('$ ' + line)}\n{line}" for line in script]
483 cmd += ["--entrypoint", "/bin/sh", image, "-c", "\n".join(lines)]
484 start = time.time()
485 p = subprocess.Popen(cmd, stdout=logf, stderr=subprocess.STDOUT)
486 while True:
487 try:
488 code = p.wait(timeout=LOG_EVERY)
489 break
490 except subprocess.TimeoutExpired:
491 if time.time() - start >= timeout:
492 subprocess.run(["docker", "kill", name], capture_output=True)
493 p.wait()
494 logf.write(f"\n\nfw-runner: stopped after {timeout} seconds (the job's timeout)\n".encode())
495 code = 124
496 break
497 if progress:
498 try:
499 progress()
500 except Exception as e: # a failed upload must not stop the job
501 log(f"progress upload failed: {e}")
502 return code, time.time() - start
503
504
505def clear_workdir(workdir):
506 """Remove a job's workspace. Containers ran as root, so it may hold root-owned files: remove it from a container too."""
507 if os.path.isdir(workdir):
508 subprocess.run(["docker", "run", "--rm", "-v", f"{workdir}:/w", "alpine:3.20", "rm", "-rf", "/w/src"],
509 capture_output=True)
510 shutil.rmtree(workdir, ignore_errors=True)
511
512
513def run_pipeline(j, keep=False):
514 repo, sha = j["repo"], j["sha"]
515 workdir = os.path.join(WORK, j["id"])
516 src = os.path.join(workdir, "src")
517 clear_workdir(workdir) # left behind if this runner crashed on the same job (a requeued job can come back to it)
518 os.makedirs(workdir, exist_ok=True)
519 pipeline = {"repo": repo, "sha": sha, "ref": j["ref"].split("/", 2)[2], "runner": RUNNER,
520 "started": now_iso(), "status": "running", "jobs": []}
521 t0 = time.time()
522 try:
523 env_git = {**os.environ, "GIT_SSH_COMMAND": " ".join(shlex.quote(a) for a in SSH)}
524 subprocess.run(["git", "clone", "--quiet", "--no-checkout", clone_url(repo), src],
525 check=True, env=env_git, capture_output=True)
526 git(src, "checkout", "--quiet", "--detach", sha)
527 # Jobs that fetch (e.g. to build every branch) reach the repo the way GitLab jobs did: over https,
528 # read-only. The runner's own ssh key never goes into a container. (Private repos can't be fetched this way.)
529 git(src, "remote", "set-url", "origin", f"{CODE_URL}/{repo}.git")
530 subprocess.run(["git", "-C", src, "submodule", "update", "--init", "--recursive", "--quiet"],
531 env=env_git, capture_output=True)
532 env = ci_env(j, src)
533 try:
534 if ci_file(src):
535 jobs, why = yaml_pipeline(src, env)
536 else:
537 jobs, why = package_pipeline(src, env)
538 except Unsupported as e:
539 jobs, why = None, str(e)
540 except Exception as e: # e.g. YAML syntax
541 jobs, why = None, f"{ci_file(src)} could not be read: {e}"
542 if jobs is None:
543 pipeline.update(status="error", message=why, duration=time.time() - t0)
544 put(repo, sha, "pipeline.json", json.dumps(pipeline))
545 return pipeline
546 if not jobs:
547 log(f"{repo} {sha[:12]}: nothing to run ({why})")
548 return None
549 protected = j["ref"] == f"refs/heads/{j['default_branch']}" or j["ref"].startswith("refs/tags/")
550 secret = os.path.join(SECRETS, f"{repo}.env")
551 env_file = secret if protected and os.path.isfile(secret) else None
552 lacking = missing_tags(j, jobs, src, protected, os.path.isfile(secret))
553 if lacking:
554 # Nothing has run and nothing is uploaded: another runner takes it (or it waits for one).
555 release(j["id"])
556 raise Declined(f"needs {', '.join(lacking)}, which {RUNNER} lacks")
557 pipeline["jobs"] = [{"name": x["name"], "stage": x["stage"], "status": "pending"} for x in jobs]
558 put(repo, sha, "pipeline.json", json.dumps(pipeline))
559
560 failed = False
561 for i, x in enumerate(jobs):
562 rec = pipeline["jobs"][i]
563 rec["log"] = f"{i:02d}-{re.sub(r'[^A-Za-z0-9._-]+', '-', x['name'])[:60]}.log"
564 if x["unsupported"]:
565 rec.update(status="error", message=f"not supported here: {', '.join(x['unsupported'])}", duration=0)
566 put(repo, sha, rec["log"], f"This job uses {', '.join(x['unsupported'])}, which fw-runner doesn't support.\n")
567 failed = failed or not x["allow_failure"]
568 continue
569 run_it = x["when"] == "always" or (x["when"] == "on_success" and not failed) or (x["when"] == "on_failure" and failed)
570 if x["when"] == "manual":
571 rec.update(status="skipped", message="manual job (not run automatically)", duration=0)
572 continue
573 if not run_it:
574 rec.update(status="skipped", message="an earlier job failed", duration=0)
575 continue
576 log_path = os.path.join(workdir, rec["log"])
577 rec.update(status="running", started=now_iso())
578 put(repo, sha, "pipeline.json", json.dumps(pipeline))
579
580 def upload_log(path=log_path, name=rec["log"]):
581 with open(path, "rb") as f:
582 put(repo, sha, name, f.read())
583 with open(log_path, "wb") as logf:
584 logf.write(f"fw-runner {RUNNER}: {x['name']} ({x['stage']}) in {x['image']}\n".encode())
585 if x["note"]:
586 logf.write(f"note: {x['note']}\n".encode())
587 logf.flush()
588 code, secs = docker_run(f"fw-{j['id']}-{i}", x["image"], src, x["env"], x["script"],
589 x["timeout"], logf, env_file, progress=lambda: (logf.flush(), upload_log()))
590 if x["after"]:
591 logf.write(b"\n--- after_script\n")
592 logf.flush()
593 docker_run(f"fw-{j['id']}-{i}-after", x["image"], src, x["env"], x["after"], 300, logf, env_file)
594 ok = code == 0
595 rec.pop("started", None)
596 rec.update(status="success" if ok else ("success" if x["allow_failure"] else "failed"),
597 duration=round(secs), **({} if ok else {"message": f"exit code {code}" + (" (allowed to fail)" if x["allow_failure"] else "")}))
598 failed = failed or (not ok and not x["allow_failure"])
599 upload_log()
600 put(repo, sha, "pipeline.json", json.dumps(pipeline))
601 pipeline.update(status="failed" if failed else "success", duration=round(time.time() - t0))
602 put(repo, sha, "pipeline.json", json.dumps(pipeline))
603
604 if not failed and protected and not j["ref"].startswith("refs/tags/"):
605 screenshots(j, src, workdir)
606 return pipeline
607 finally:
608 if not keep:
609 clear_workdir(workdir)
610
611
612# ---------------------------------------------------------------- screenshots
613
614SIZES = {"phone": (390, 844), "tablet": (820, 1180), "desktop": (1280, 800)}
615
616
617def shoot_image():
618 if subprocess.run(["docker", "image", "inspect", SHOOT_IMAGE], capture_output=True).returncode != 0:
619 dockerfile = (f"FROM mcr.microsoft.com/playwright/python:v{PLAYWRIGHT}-noble\n"
620 f"RUN pip install --no-cache-dir --break-system-packages playwright=={PLAYWRIGHT}\n")
621 subprocess.run(["docker", "build", "-q", "-t", SHOOT_IMAGE, "-"], input=dockerfile.encode(), check=True, capture_output=True)
622 return SHOOT_IMAGE
623
624
625def shot_plan(src):
626 """(package.json's "fuzzyworld" section, app, screenshot config) for an app that wants screenshots, else None."""
627 try:
628 with open(os.path.join(src, "package.json")) as f:
629 cfg = json.load(f).get("fuzzyworld") or {}
630 except (OSError, ValueError):
631 return None
632 app, shots = cfg.get("app"), cfg.get("screenshots", [{"path": "", "size": "phone"}])
633 if not app or shots is False or shots == []:
634 return None
635 return cfg, app, shots
636
637
638def screenshots(j, src, workdir):
639 """Build the app and photograph it, for apps (package.json "fuzzyworld.app") on the default branch."""
640 plan = shot_plan(src)
641 if not plan:
642 return
643 cfg, app, shots = plan
644 repo, sha = j["repo"], j["sha"]
645 env = {**ci_env(j, src), "FW_APP": app, "FW_BASE": f"/{app}/"}
646 log_path = os.path.join(workdir, "screenshots.log")
647 shots_dir = os.path.join(workdir, "shots")
648 os.makedirs(shots_dir, exist_ok=True)
649 with open(log_path, "wb") as logf:
650 # 1. Build exactly as `fw deploy` would (fresh install, prebuild, checks, build, output checks).
651 script = ["corepack enable", FW_DEPLOY_CHECK]
652 if isinstance(shots, str): # the app takes its own pictures: a command that writes PNGs to $FW_SHOTS_DIR
653 script += [f"FW_SHOTS_DIR=/builds/{repo}/.fw-shots {shots}"]
654 code, _ = docker_run(f"fw-{j['id']}-build", DEFAULT_IMAGE, src, env, script, JOB_TIMEOUT, logf)
655 if code != 0:
656 log(f"{repo}: screenshot build failed (exit {code}); see {log_path}")
657 return
658 if isinstance(shots, str):
659 own = os.path.join(src, ".fw-shots")
660 files = sorted(f for f in os.listdir(own) if f.endswith(".png")) if os.path.isdir(own) else []
661 for f in files:
662 shutil.copy(os.path.join(own, f), os.path.join(shots_dir, f))
663 entries = [{"file": f, "label": os.path.splitext(f)[0]} for f in files]
664 else:
665 out = cfg.get("out") or next((d for d in ("dist", "public", "build") if os.path.exists(os.path.join(src, d, "index.html"))), None)
666 if not out:
667 return
668 spec = [{"path": s.get("path", ""), "size": s.get("size", "phone"), "wait": int(s.get("wait", 1500))}
669 for s in shots if isinstance(s, dict) and s.get("size", "phone") in SIZES]
670 entries = [{"file": f"{i + 1:02d}-{s['size']}.png", "label": s["path"] or s["size"], **s} for i, s in enumerate(spec)]
671 with open(os.path.join(shots_dir, "spec.json"), "w") as f:
672 json.dump({"app": app, "shots": [{**e, "w": SIZES[e["size"]][0], "h": SIZES[e["size"]][1]} for e in entries]}, f)
673 # 2. Serve the build at /<app>/ and photograph it, in a Playwright container without network access.
674 code, _ = docker_run(f"fw-{j['id']}-shoot", shoot_image(), src, env,
675 [f"python3 /fw/shoot.py /shots /builds/{repo}/{out}"],
676 600, logf, extra=["--network", "none", "-v", f"{HERE}:/fw:ro", "-v", f"{shots_dir}:/shots"])
677 if code != 0:
678 log(f"{repo}: screenshots failed (exit {code}); see {log_path}")
679 return
680 taken = [e for e in entries if os.path.exists(os.path.join(shots_dir, e["file"]))]
681 if not taken:
682 return
683 for e in taken:
684 with open(os.path.join(shots_dir, e["file"]), "rb") as f:
685 api("put-shot", app, e["file"], data=f.read())
686 meta = {"commit": sha, "taken": now_iso(), "shots": [{"file": e["file"], "label": e["label"]} for e in taken]}
687 api("put-shot", app, "shots.json", data=json.dumps(meta).encode())
688 api("shots-done", app, repo)
689 log(f"{repo}: {len(taken)} screenshot(s) of {app}")
690
691
692# ---------------------------------------------------------------- main
693
694def plan(src, ref):
695 """--plan: what a push of this checkout would run (for checking a .fw-ci.yml)."""
696 sha = git(src, "rev-parse", "HEAD")
697 j = {"id": "plan", "repo": os.path.basename(os.path.abspath(src)), "sha": sha, "ref": ref, "default_branch": "main"}
698 env = ci_env(j, src)
699 try:
700 jobs, why = yaml_pipeline(src, env) if ci_file(src) else package_pipeline(src, env)
701 except Unsupported as e:
702 print(f"error: {e}")
703 return 1
704 if not jobs:
705 print(f"nothing to run: {why}")
706 for x in jobs:
707 print(f"[{x['stage']}] {x['name']} image={x['image']} when={x['when']}"
708 + (f" tags={','.join(x['tags'])}" if x["tags"] else "")
709 + (f" UNSUPPORTED: {', '.join(x['unsupported'])}" if x["unsupported"] else "")
710 + (f"\n {x['note']}" if x["note"] else ""))
711 for line in x["script"]:
712 print(f" $ {line}")
713 return 0
714
715
716class Heartbeat(threading.Thread):
717 """Tells DreamHost every BEAT_EVERY seconds that this runner is still working on a job. Without it the
718 job goes back in the queue after ~2 minutes, so a closed laptop doesn't hold a pipeline for hours."""
719
720 def __init__(self, job_id):
721 super().__init__(daemon=True)
722 self.job_id, self.stopped = job_id, threading.Event()
723
724 def run(self):
725 while not self.stopped.wait(BEAT_EVERY):
726 try:
727 api("beat", self.job_id, RUNNER, timeout=BEAT_TIMEOUT)
728 except Exception as e: # a network blip: the lease is ~2 minutes, so the next beat may still be in time
729 log(f"heartbeat failed: {e}")
730
731 def stop(self):
732 self.stopped.set()
733
734
735def parse_top_idle(out):
736 """The idle percentage from the last "CPU usage:" line of macOS `top -l 2` output, or None."""
737 m = re.findall(r"CPU usage:.*?([\d.]+)% idle", out)
738 return float(m[-1]) if m else None
739
740
741def cpu_idle_percent():
742 """Percent of CPU idle over the last second or so, or None if this system doesn't say."""
743 try:
744 if sys.platform == "darwin":
745 out = subprocess.run(["top", "-l", "2", "-n", "0", "-s", "1"], capture_output=True, text=True, timeout=20).stdout
746 return parse_top_idle(out)
747 def times():
748 f = [int(x) for x in open("/proc/stat").readline().split()[1:]]
749 return f[3] + f[4], sum(f) # idle + iowait, total
750 i0, t0 = times()
751 time.sleep(1)
752 i1, t1 = times()
753 return 100.0 * (i1 - i0) / (t1 - t0) if t1 > t0 else None
754 except (OSError, ValueError, subprocess.SubprocessError):
755 return None
756
757
758def on_ac_power():
759 """False only when the system says it is on battery; True when it is on AC or has no battery to report."""
760 try:
761 if sys.platform == "darwin":
762 out = subprocess.run(["pmset", "-g", "batt"], capture_output=True, text=True, timeout=10).stdout
763 return "Battery Power" not in out
764 base = "/sys/class/power_supply"
765 mains = [os.path.join(base, d, "online") for d in os.listdir(base) if open(os.path.join(base, d, "type")).read().strip() == "Mains"]
766 return not mains or any(open(m).read().strip() == "1" for m in mains)
767 except (OSError, subprocess.SubprocessError):
768 return True
769
770
771def busy_reason():
772 """Why a polite runner should not claim a job right now, or "" if it may."""
773 if not on_ac_power():
774 return "on battery"
775 try:
776 if subprocess.run(["docker", "info"], capture_output=True, timeout=DOCKER_INFO_TIMEOUT).returncode != 0:
777 return "Docker is not running"
778 except subprocess.TimeoutExpired:
779 return "Docker is not answering"
780 except FileNotFoundError:
781 return "Docker is not installed"
782 idle = cpu_idle_percent()
783 if idle is not None and idle < MIN_IDLE:
784 return f"CPU busy ({100 - idle:.0f}% used)"
785 return ""
786
787
788def say_once(reason):
789 """Log a skip reason only when it changes, so a polite runner polling every 2 minutes doesn't fill the log."""
790 path = os.path.join(HERE, ".last-skip")
791 try:
792 last = open(path).read()
793 except OSError:
794 last = ""
795 if reason != last:
796 log(f"not claiming: {reason}" if reason else "ready to claim again")
797 with open(path, "w") as f:
798 f.write(reason)
799
800
801def main(argv):
802 if argv[1:2] == ["--plan"]:
803 return plan(argv[2], argv[3] if len(argv) > 3 else "refs/heads/main")
804 once, keep = "--once" in argv, "--keep" in argv
805 os.makedirs(WORK, exist_ok=True)
806 # One runner process at a time. (TrueNAS cron also wraps it in `flock -n .lock`; launchd on a Mac has no flock.)
807 lock = open(os.path.join(HERE, ".runner.lock"), "a")
808 try:
809 fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
810 except OSError:
811 return 0
812 started_as = os.path.getmtime(__file__)
813 while True:
814 # runner/install.sh replaced this file: continue as the new version (between jobs, never during one).
815 if os.path.getmtime(__file__) != started_as:
816 log("fw-runner.py was updated; restarting")
817 lock.close()
818 os.execv(sys.executable, [sys.executable, os.path.abspath(__file__)] + argv[1:])
819 if POLITE:
820 reason = busy_reason()
821 say_once(reason)
822 if reason:
823 return 0
824 raw = api("claim", RUNNER, ",".join(sorted(TAGS)) or "-").decode().strip()
825 if not raw:
826 return 0
827 j = json.loads(raw)
828 log(f"claimed {j['id']}: {j['repo']} {j['ref']} {j['sha'][:12]}")
829 beat = Heartbeat(j["id"])
830 beat.start()
831 try:
832 p = run_pipeline(j, keep)
833 log(f"{j['id']}: {p['status'] if p else 'nothing to run'}")
834 except Declined as e:
835 beat.stop()
836 log(f"{j['id']}: {j['repo']} given back: {e}")
837 continue
838 except Exception as e:
839 log(f"{j['id']}: runner error: {e}")
840 try:
841 put(j["repo"], j["sha"], "pipeline.json", json.dumps({
842 "repo": j["repo"], "sha": j["sha"], "ref": j["ref"].split("/", 2)[2], "runner": RUNNER,
843 "started": now_iso(), "status": "error", "message": f"runner error: {e}", "jobs": []}))
844 except Exception:
845 pass
846 beat.stop()
847 try:
848 api("done", j["repo"], j["sha"], j["id"], RUNNER)
849 except Exception as e: # refused when the lease was lost and another runner holds the job: expected, not a crash
850 log(f"{j['id']}: not marked done ({e})")
851 if once:
852 return 0
853
854
855if __name__ == "__main__":
856 sys.exit(main(sys.argv))