[feat]: add selectable Modal and GB200 Kubernetes GPU CI
- Preserve existing CI and add configurable GPU backend routing - Limit execution to 2 active PRs, 4 GPUs per PR, and 8 GPUs total - Add persistent admission, cancellation recovery, and backend statuses - Document deployment and add infrastructure regression tests
This commit is contained in:
Satyam Srivastava
@@ -0,0 +1 @@
|
||||
"""Trusted, opt-in GPU CI dispatch. Never import this package from a PR checkout."""
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Operator CLI. Use the isolated entrypoint from an immutable installation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .dispatcher import (cancellation_signals, execute, make_store, recover,
|
||||
run_directory, upload_artifacts)
|
||||
from .pipeline import render_pipeline
|
||||
from .policy import DEFAULT_REPOSITORY, backend_name, build_request
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def load_config(path: Path) -> dict[str, Any]:
|
||||
config = json.loads(path.read_text())
|
||||
if not isinstance(config, dict):
|
||||
raise ValueError("Operator config must be a JSON object")
|
||||
for key in ("state_path", "artifacts_dir"):
|
||||
value = config.get(key)
|
||||
if not isinstance(value, str) or not Path(value).is_absolute():
|
||||
raise ValueError(f"{key} must be an absolute local path")
|
||||
for key, default in (("max_active_prs", 2), ("max_gpus_per_pr", 4), ("max_gpus", 8),
|
||||
("queue_timeout_seconds", 21600)):
|
||||
value = config.get(key, default)
|
||||
if type(value) is not int or value < 1:
|
||||
raise ValueError(f"{key} must be a positive integer")
|
||||
config[key] = value
|
||||
# The initial rollout deliberately cannot raise the user's agreed ceilings.
|
||||
for key, ceiling in (("max_active_prs", 2), ("max_gpus_per_pr", 4), ("max_gpus", 8)):
|
||||
if config[key] > ceiling:
|
||||
raise ValueError(f"{key} exceeds the reviewed ceiling {ceiling}")
|
||||
if config.get("repository", DEFAULT_REPOSITORY) != DEFAULT_REPOSITORY:
|
||||
raise ValueError("This worker installation is allowlisted only for hao-ai-lab/FastVideo")
|
||||
backend_name(config.get("default_backend"))
|
||||
return config
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
for name in ("run", "render", "upload", "status", "recover"):
|
||||
sub = subparsers.add_parser(name)
|
||||
sub.add_argument("--config", type=Path, required=True)
|
||||
if name in ("render", "upload"):
|
||||
sub.add_argument("--canonical", type=Path, default=ROOT / ".buildkite/pipeline.yml")
|
||||
if name == "recover":
|
||||
sub.add_argument("--build-id", required=True)
|
||||
args = parser.parse_args(argv)
|
||||
try:
|
||||
config = load_config(args.config)
|
||||
if args.command == "status":
|
||||
print(json.dumps(make_store(config).snapshot(), indent=2))
|
||||
return 0
|
||||
if args.command == "recover":
|
||||
recover(config, args.build_id)
|
||||
return 0
|
||||
backend = backend_name(os.environ.get("CI_GPU_BACKEND"), config.get("default_backend", "slurm"))
|
||||
if args.command == "upload" and backend == "slurm":
|
||||
if config.get("default_backend", "slurm") != "slurm":
|
||||
raise ValueError("Slurm overrides are disabled after promoting another backend: legacy statuses are unqualified")
|
||||
legacy = config.get("slurm_uploader")
|
||||
if (not isinstance(legacy, list) or not legacy or
|
||||
not all(isinstance(x, str) and x for x in legacy) or not Path(legacy[0]).is_absolute()):
|
||||
raise ValueError("Configure slurm_uploader as the existing trusted uploader argv")
|
||||
return subprocess.run(legacy, check=False).returncode
|
||||
if args.command in ("render", "upload"):
|
||||
import yaml
|
||||
canonical = yaml.safe_load(args.canonical.read_text())
|
||||
rendered = render_pipeline(canonical, backend, os.environ.get("TEST_SCOPE", ""),
|
||||
config.get("queue", "gpu-ci-dispatch"))
|
||||
payload = json.dumps(rendered, indent=2) + "\n"
|
||||
if args.command == "render":
|
||||
print(payload, end="")
|
||||
return 0
|
||||
return subprocess.run(["buildkite-agent", "pipeline", "upload", "--no-interpolation"],
|
||||
input=payload, text=True, check=False).returncode
|
||||
request = build_request(os.environ, config, ROOT)
|
||||
cancel = threading.Event()
|
||||
with cancellation_signals(cancel):
|
||||
code = execute(request, config, cancel=cancel)
|
||||
if config.get("upload_artifacts", True):
|
||||
upload_artifacts(run_directory(config, request["build_id"]))
|
||||
return code
|
||||
except (OSError, ValueError, RuntimeError, KeyError, subprocess.SubprocessError) as error:
|
||||
print(f"GPU CI dispatcher error: {error}", file=sys.stderr)
|
||||
return 97
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,307 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
"""Persistent GPU admission for cooperating processes on one trusted host.
|
||||
|
||||
Keep this SQLite database on local disk, not a shared/network filesystem.
|
||||
Reservations include backend handles before external creation. Neither stale
|
||||
heartbeats nor cancellation free active reservations: the caller must stop the
|
||||
backend worker, confirm it stopped, and then call ``release``. The dispatcher
|
||||
must also serialize creation and recovery for each build outside this ledger.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
class AdmissionStore:
|
||||
"""Bound distinct PRs and GPU reservations across all selected backends.
|
||||
|
||||
``pr_key`` identifies a repository and PR together. ``None`` or an empty
|
||||
string gives a non-PR build its own slot. Once admitted, all unfinished
|
||||
builds for a PR share and retain its slot, even between lanes. Completed
|
||||
build and lane records remain as tombstones; retries must use new lane IDs.
|
||||
"""
|
||||
|
||||
def __init__(self, path: str | Path, max_prs: int = 2, max_gpus_per_pr: int = 4, max_gpus: int = 8):
|
||||
for name, value in (("max_prs", max_prs), ("max_gpus_per_pr", max_gpus_per_pr), ("max_gpus", max_gpus)):
|
||||
if type(value) is not int or value < 1:
|
||||
raise ValueError(f"{name} must be a positive integer")
|
||||
if str(path) == ":memory:":
|
||||
raise ValueError("Admission requires a persistent database on local disk")
|
||||
self.path = Path(path)
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.max_prs = max_prs
|
||||
self.max_gpus_per_pr = max_gpus_per_pr
|
||||
self.max_gpus = max_gpus
|
||||
with self._transaction() as connection:
|
||||
connection.execute("""
|
||||
CREATE TABLE IF NOT EXISTS settings (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
schema_version INTEGER NOT NULL,
|
||||
max_prs INTEGER NOT NULL,
|
||||
max_gpus_per_pr INTEGER NOT NULL,
|
||||
max_gpus INTEGER NOT NULL
|
||||
)
|
||||
""")
|
||||
connection.execute("INSERT OR IGNORE INTO settings VALUES (1, 1, ?, ?, ?)",
|
||||
(max_prs, max_gpus_per_pr, max_gpus))
|
||||
settings = connection.execute("SELECT * FROM settings WHERE id = 1").fetchone()
|
||||
expected = (1, max_prs, max_gpus_per_pr, max_gpus)
|
||||
actual = tuple(settings[name] for name in ("schema_version", "max_prs", "max_gpus_per_pr", "max_gpus"))
|
||||
if actual != expected:
|
||||
raise ValueError("Admission database schema or limits conflict with this process")
|
||||
connection.execute("""
|
||||
CREATE TABLE IF NOT EXISTS builds (
|
||||
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
build_id TEXT NOT NULL UNIQUE,
|
||||
pr_key TEXT,
|
||||
workload_key TEXT NOT NULL,
|
||||
backend TEXT NOT NULL,
|
||||
commit_sha TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN ('queued', 'admitted', 'finished')),
|
||||
cancelled INTEGER NOT NULL DEFAULT 0 CHECK (cancelled IN (0, 1)),
|
||||
created_at REAL NOT NULL,
|
||||
heartbeat_at REAL NOT NULL,
|
||||
admitted_at REAL,
|
||||
finished_at REAL
|
||||
)
|
||||
""")
|
||||
connection.execute("""
|
||||
CREATE TABLE IF NOT EXISTS allocations (
|
||||
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
build_id TEXT NOT NULL REFERENCES builds(build_id),
|
||||
lane_id TEXT NOT NULL,
|
||||
gpus INTEGER NOT NULL CHECK (gpus > 0),
|
||||
handle_json TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN ('waiting', 'active', 'released', 'cancelled')),
|
||||
created_at REAL NOT NULL,
|
||||
acquired_at REAL,
|
||||
released_at REAL,
|
||||
UNIQUE (build_id, lane_id)
|
||||
)
|
||||
""")
|
||||
connection.execute("CREATE INDEX IF NOT EXISTS builds_workload ON builds(workload_key, state)")
|
||||
connection.execute("CREATE INDEX IF NOT EXISTS allocations_state ON allocations(state, sequence)")
|
||||
|
||||
@contextmanager
|
||||
def _transaction(self) -> Iterator[sqlite3.Connection]:
|
||||
connection = sqlite3.connect(str(self.path), timeout=30, isolation_level=None)
|
||||
connection.row_factory = sqlite3.Row
|
||||
try:
|
||||
connection.execute("PRAGMA foreign_keys = ON")
|
||||
connection.execute("BEGIN IMMEDIATE")
|
||||
yield connection
|
||||
connection.commit()
|
||||
except BaseException:
|
||||
connection.rollback()
|
||||
raise
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
@staticmethod
|
||||
def _identifier(value: str, name: str) -> None:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
raise ValueError(f"{name} must be a nonempty string")
|
||||
|
||||
@staticmethod
|
||||
def _build(connection: sqlite3.Connection, build_id: str) -> sqlite3.Row:
|
||||
build = connection.execute("SELECT * FROM builds WHERE build_id = ?", (build_id, )).fetchone()
|
||||
if build is None:
|
||||
raise KeyError(f"Unknown build: {build_id}")
|
||||
return build
|
||||
|
||||
def register(self, build_id: str, pr_key: str | None, backend: str, commit: str) -> None:
|
||||
"""Register once; an identical duplicate never resets state or queue age."""
|
||||
for name, value in (("build_id", build_id), ("backend", backend), ("commit", commit)):
|
||||
self._identifier(value, name)
|
||||
if pr_key is not None and not isinstance(pr_key, str):
|
||||
raise ValueError("pr_key must be a string or None")
|
||||
pr_key = pr_key or None
|
||||
if pr_key is not None:
|
||||
self._identifier(pr_key, "pr_key")
|
||||
workload_key = f"pr:{pr_key}" if pr_key is not None else f"build:{build_id}"
|
||||
with self._transaction() as connection:
|
||||
existing = connection.execute("SELECT * FROM builds WHERE build_id = ?", (build_id, )).fetchone()
|
||||
if existing is not None:
|
||||
if (existing["pr_key"], existing["backend"], existing["commit_sha"]) != (pr_key, backend, commit):
|
||||
raise ValueError(f"Build identity changed: {build_id}")
|
||||
return
|
||||
active = connection.execute("SELECT 1 FROM builds WHERE workload_key = ? AND state = 'admitted'",
|
||||
(workload_key, )).fetchone()
|
||||
now = time.time()
|
||||
connection.execute("""
|
||||
INSERT INTO builds (build_id, pr_key, workload_key, backend, commit_sha,
|
||||
state, created_at, heartbeat_at, admitted_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (build_id, pr_key, workload_key, backend, commit,
|
||||
"admitted" if active else "queued", now, now, now if active else None))
|
||||
|
||||
def try_admit(self, build_id: str) -> bool:
|
||||
"""Admit the oldest waiting PRs, reserving space for earlier waiters."""
|
||||
with self._transaction() as connection:
|
||||
build = self._build(connection, build_id)
|
||||
if build["cancelled"] or build["state"] == "finished":
|
||||
return False
|
||||
if build["state"] == "admitted":
|
||||
return True
|
||||
active_count = connection.execute(
|
||||
"SELECT COUNT(DISTINCT workload_key) FROM builds WHERE state = 'admitted'").fetchone()[0]
|
||||
available = self.max_prs - active_count
|
||||
if available <= 0:
|
||||
return False
|
||||
eligible = connection.execute("""
|
||||
SELECT workload_key FROM builds WHERE state = 'queued' AND cancelled = 0
|
||||
GROUP BY workload_key ORDER BY MIN(sequence) LIMIT ?
|
||||
""", (available, )).fetchall()
|
||||
if build["workload_key"] not in {row["workload_key"] for row in eligible}:
|
||||
return False
|
||||
connection.execute("""
|
||||
UPDATE builds SET state = 'admitted', admitted_at = ?
|
||||
WHERE workload_key = ? AND state = 'queued' AND cancelled = 0
|
||||
""", (time.time(), build["workload_key"]))
|
||||
return True
|
||||
|
||||
def try_acquire(self, build_id: str, lane_id: str, gpus: int, handle: dict[str, Any]) -> bool:
|
||||
"""Persist a handle and reserve GPUs before the caller creates a worker.
|
||||
|
||||
Pending requests keep FIFO order. A PR whose oldest request cannot fit
|
||||
its own budget does not block the other PR. Once a request fits its PR
|
||||
budget it waits for global capacity without smaller requests bypassing
|
||||
it. Repeating an active request returns True with the same reservation;
|
||||
released/cancelled lane IDs cannot acquire again.
|
||||
"""
|
||||
self._identifier(lane_id, "lane_id")
|
||||
if type(gpus) is not int or not 1 <= gpus <= min(self.max_gpus_per_pr, self.max_gpus):
|
||||
raise ValueError("Requested GPUs must fit both the per-PR and total GPU limits")
|
||||
if not isinstance(handle, dict):
|
||||
raise ValueError("Backend handle must be a JSON object")
|
||||
try:
|
||||
encoded_handle = json.dumps(handle, sort_keys=True, separators=(",", ":"), allow_nan=False)
|
||||
except (TypeError, ValueError) as error:
|
||||
raise ValueError("Backend handle must be JSON serializable") from error
|
||||
with self._transaction() as connection:
|
||||
build = self._build(connection, build_id)
|
||||
allocation = connection.execute("SELECT * FROM allocations WHERE build_id = ? AND lane_id = ?",
|
||||
(build_id, lane_id)).fetchone()
|
||||
if allocation is not None:
|
||||
if allocation["gpus"] != gpus or allocation["handle_json"] != encoded_handle:
|
||||
raise ValueError(f"Lane reservation identity changed: {build_id}/{lane_id}")
|
||||
if build["cancelled"] or build["state"] != "admitted":
|
||||
return False
|
||||
if allocation is not None and allocation["state"] != "waiting":
|
||||
return allocation["state"] == "active"
|
||||
if allocation is None:
|
||||
connection.execute("""
|
||||
INSERT INTO allocations (build_id, lane_id, gpus, handle_json, state, created_at)
|
||||
VALUES (?, ?, ?, ?, 'waiting', ?)
|
||||
""", (build_id, lane_id, gpus, encoded_handle, time.time()))
|
||||
|
||||
usage = connection.execute("""
|
||||
SELECT b.workload_key, SUM(a.gpus) AS gpus FROM allocations a
|
||||
JOIN builds b ON b.build_id = a.build_id WHERE a.state = 'active'
|
||||
GROUP BY b.workload_key
|
||||
""").fetchall()
|
||||
gpu_usage = {row["workload_key"]: row["gpus"] for row in usage}
|
||||
total_gpus = sum(gpu_usage.values())
|
||||
waiting = connection.execute("""
|
||||
SELECT a.*, b.workload_key FROM allocations a
|
||||
JOIN builds b ON b.build_id = a.build_id
|
||||
WHERE a.state = 'waiting' AND b.state = 'admitted' AND b.cancelled = 0
|
||||
ORDER BY a.sequence
|
||||
""").fetchall()
|
||||
seen_workloads = set()
|
||||
for request in waiting:
|
||||
workload_key = request["workload_key"]
|
||||
if workload_key in seen_workloads:
|
||||
continue
|
||||
seen_workloads.add(workload_key)
|
||||
if gpu_usage.get(workload_key, 0) + request["gpus"] > self.max_gpus_per_pr:
|
||||
continue
|
||||
if (request["build_id"], request["lane_id"]) != (build_id, lane_id):
|
||||
return False
|
||||
if total_gpus + gpus > self.max_gpus:
|
||||
return False
|
||||
connection.execute("""
|
||||
UPDATE allocations SET state = 'active', acquired_at = ?
|
||||
WHERE build_id = ? AND lane_id = ?
|
||||
""", (time.time(), build_id, lane_id))
|
||||
return True
|
||||
return False
|
||||
|
||||
def release(self, build_id: str, lane_id: str) -> None:
|
||||
"""Release only after the caller confirms the worker has stopped.
|
||||
|
||||
Missing/already released reservations are harmless for reconciliation.
|
||||
This method does not stop workers or infer their status from timestamps.
|
||||
"""
|
||||
with self._transaction() as connection:
|
||||
self._build(connection, build_id)
|
||||
connection.execute("""
|
||||
UPDATE allocations SET state = 'released', released_at = ?
|
||||
WHERE build_id = ? AND lane_id = ? AND state IN ('active', 'waiting')
|
||||
""", (time.time(), build_id, lane_id))
|
||||
|
||||
def finish(self, build_id: str) -> None:
|
||||
"""Finish or cancel a queued build, refusing any active reservations."""
|
||||
with self._transaction() as connection:
|
||||
build = self._build(connection, build_id)
|
||||
if build["state"] == "finished":
|
||||
return
|
||||
active = connection.execute("SELECT 1 FROM allocations WHERE build_id = ? AND state = 'active'",
|
||||
(build_id, )).fetchone()
|
||||
if active is not None:
|
||||
raise RuntimeError(f"Cannot finish build with active GPU allocations: {build_id}")
|
||||
now = time.time()
|
||||
connection.execute("""
|
||||
UPDATE allocations SET state = 'cancelled', released_at = ?
|
||||
WHERE build_id = ? AND state = 'waiting'
|
||||
""", (now, build_id))
|
||||
connection.execute("UPDATE builds SET state = 'finished', finished_at = ? WHERE build_id = ?",
|
||||
(now, build_id))
|
||||
|
||||
def heartbeat(self, build_id: str) -> None:
|
||||
"""Record liveness for operators; no timeout ever frees a reservation."""
|
||||
with self._transaction() as connection:
|
||||
self._build(connection, build_id)
|
||||
connection.execute("UPDATE builds SET heartbeat_at = ? WHERE build_id = ? AND state != 'finished'",
|
||||
(time.time(), build_id))
|
||||
|
||||
def request_cancel(self, build_id: str) -> None:
|
||||
"""Block further admission/launch reservations without freeing GPUs."""
|
||||
with self._transaction() as connection:
|
||||
self._build(connection, build_id)
|
||||
connection.execute("UPDATE builds SET cancelled = 1 WHERE build_id = ?", (build_id, ))
|
||||
connection.execute("""
|
||||
UPDATE allocations SET state = 'cancelled', released_at = ?
|
||||
WHERE build_id = ? AND state = 'waiting'
|
||||
""", (time.time(), build_id))
|
||||
|
||||
def is_cancelled(self, build_id: str) -> bool:
|
||||
with self._transaction() as connection:
|
||||
return bool(self._build(connection, build_id)["cancelled"])
|
||||
|
||||
def snapshot(self) -> dict[str, Any]:
|
||||
"""Return a consistent reconciliation view, including terminal records."""
|
||||
with self._transaction() as connection:
|
||||
builds = [dict(row) for row in connection.execute("SELECT * FROM builds ORDER BY sequence")]
|
||||
allocations = [dict(row) for row in connection.execute("SELECT * FROM allocations ORDER BY sequence")]
|
||||
for build in builds:
|
||||
build["commit"] = build.pop("commit_sha")
|
||||
build["cancelled"] = bool(build["cancelled"])
|
||||
for allocation in allocations:
|
||||
allocation["handle"] = json.loads(allocation.pop("handle_json"))
|
||||
return {
|
||||
"limits": {
|
||||
"max_prs": self.max_prs,
|
||||
"max_gpus_per_pr": self.max_gpus_per_pr,
|
||||
"max_gpus": self.max_gpus,
|
||||
},
|
||||
"builds": builds,
|
||||
"allocations": allocations,
|
||||
}
|
||||
@@ -0,0 +1,395 @@
|
||||
"""Trusted execution adapters. No PR code executes in the controller process.
|
||||
|
||||
Modal API: https://modal.com/docs/sdk/py/latest/Sandbox
|
||||
The controller persists deterministic handles before calling ``run``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any
|
||||
|
||||
REPOSITORY = "https://github.com/hao-ai-lab/FastVideo.git"
|
||||
ENV_ALLOWLIST = {
|
||||
"FASTVIDEO_GOLDEN_TEST_FILES", "FASTVIDEO_SSIM_TEST_FILES", "BUILDKITE_BRANCH",
|
||||
"BUILDKITE_SOURCE", "BUILDKITE_PULL_REQUEST", "BUILDKITE_BUILD_NUMBER", "BUILDKITE_BUILD_URL",
|
||||
"BUILDKITE_COMMIT", "BUILDKITE_REPO", "BUILDKITE_BUILD_ID", "BUILDKITE_JOB_ID",
|
||||
}
|
||||
|
||||
|
||||
def _image(value: str) -> str:
|
||||
if not re.fullmatch(r"[A-Za-z0-9._/:\-]+@sha256:[0-9a-f]{64}", value):
|
||||
raise ValueError("Backend image must be pinned by sha256 digest")
|
||||
return value
|
||||
|
||||
|
||||
def _subpath(value: str) -> str:
|
||||
path = PurePosixPath(value)
|
||||
if not value or path.is_absolute() or ".." in path.parts or value in {".", "/"}:
|
||||
raise ValueError("PVC mounts require a dedicated relative CI subpath")
|
||||
return value
|
||||
|
||||
|
||||
class Backend:
|
||||
name = "abstract"
|
||||
|
||||
def __init__(self, config: dict, trusted_dir: Path, run_dir: Path):
|
||||
self.config = config
|
||||
self.trusted_dir = trusted_dir
|
||||
self.run_dir = run_dir
|
||||
self.image = _image(config["image"])
|
||||
self.run_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def handle(self, build_id: str, lane_id: str) -> dict:
|
||||
identity = hashlib.sha256(f"{build_id}\0{lane_id}".encode()).hexdigest()[:32]
|
||||
return {"backend": self.name, "name": f"fv-ci-{identity}", "run_id": identity}
|
||||
|
||||
def _environment(self, request: dict, lane: dict) -> dict[str, str]:
|
||||
if request["repository"] != REPOSITORY or not re.fullmatch(r"[0-9a-f]{40}", request["commit"]):
|
||||
raise ValueError("Worker requires the fixed repository and an immutable commit")
|
||||
if type(lane["gpus"]) is not int or lane["gpus"] not in range(1, 5):
|
||||
raise ValueError("A lane must request between one and four GPUs")
|
||||
script = lane["script"]
|
||||
if script != ".buildkite/scripts/unit_test.sh" and not re.fullmatch(
|
||||
r"\.buildkite/scripts/lanes/[a-z_]+\.sh", script):
|
||||
raise ValueError("Invalid lane script")
|
||||
extras = lane.get("extras", ["test"])
|
||||
if not extras or any(not re.fullmatch(r"[a-z][a-z0-9-]*", extra) for extra in extras):
|
||||
raise ValueError("Invalid dependency extras")
|
||||
if str(lane["fa4"]) not in {"0", "1"}:
|
||||
raise ValueError("Invalid attention policy")
|
||||
build_id = str(request.get("buildkite_build_id") or request["build_id"])
|
||||
job_id = str(request.get("buildkite_job_id") or request.get("env", {}).get("BUILDKITE_JOB_ID")
|
||||
or request["build_id"])
|
||||
if any(not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,199}", value) for value in (build_id, job_id)):
|
||||
raise ValueError("Invalid Buildkite build/job identity")
|
||||
pr_number = str(request.get("pr_number") or "false")
|
||||
if pr_number != "false" and not re.fullmatch(r"[1-9][0-9]*", pr_number):
|
||||
raise ValueError("Invalid PR number")
|
||||
env = {key: str(value) for key, value in request.get("env", {}).items() if key in ENV_ALLOWLIST}
|
||||
env.update({
|
||||
"BUILDKITE_COMMIT": request["commit"], "BUILDKITE_REPO": REPOSITORY,
|
||||
"BUILDKITE_BUILD_ID": build_id, "BUILDKITE_JOB_ID": job_id,
|
||||
"BUILDKITE_PULL_REQUEST": pr_number, "FASTVIDEO_CONTAINER_IMAGE_REF": self.image,
|
||||
"FASTVIDEO_CI_REPOSITORY": REPOSITORY, "FASTVIDEO_CI_COMMIT": request["commit"],
|
||||
"FASTVIDEO_CI_PR_NUMBER": str(request.get("pr_number") or ""),
|
||||
"FASTVIDEO_CI_GPUS": str(lane["gpus"]), "FASTVIDEO_CI_SCRIPT": script,
|
||||
"FASTVIDEO_CI_EXTRAS": ",".join(extras), "FASTVIDEO_FA4": str(lane["fa4"]),
|
||||
"FASTVIDEO_CI_KERNEL": "1" if lane.get("kernel", True) else "0",
|
||||
"FASTVIDEO_CI_LOCAL_ONLY": "1", "FASTVIDEO_SSIM_BOOTSTRAP_MODE": "0",
|
||||
"TEST_SCOPE": request.get("scope", "direct"),
|
||||
})
|
||||
return env
|
||||
|
||||
def run(self, handle: dict, request: dict, lane: dict, cancel: threading.Event) -> int:
|
||||
raise NotImplementedError
|
||||
|
||||
def stop(self, handle: dict) -> bool:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class KubernetesBackend(Backend):
|
||||
name = "kubernetes"
|
||||
|
||||
def __init__(self, config: dict, trusted_dir: Path, run_dir: Path):
|
||||
super().__init__(config, trusted_dir, run_dir)
|
||||
if config.get("namespace", "vllm") != "vllm":
|
||||
raise ValueError("Kubernetes CI is restricted to namespace vllm")
|
||||
for kind in ("cache", "artifacts"):
|
||||
claim = config.get(f"{kind}_pvc")
|
||||
if claim:
|
||||
if claim in {"lustre-pvc-vllm", "nfs-pvc-vllm"}:
|
||||
raise ValueError("Personal development PVCs cannot be mounted in CI")
|
||||
_subpath(config.get(f"{kind}_subpath", ""))
|
||||
|
||||
def handle(self, build_id: str, lane_id: str) -> dict:
|
||||
result = super().handle(build_id, lane_id)
|
||||
result.update({"namespace": "vllm", "context": self.config.get("context")})
|
||||
return result
|
||||
|
||||
def _command(self, handle: dict, *args: str) -> list[str]:
|
||||
if handle.get("namespace") != "vllm" or handle.get("context") != self.config.get("context"):
|
||||
raise ValueError("Handle does not match configured Kubernetes target")
|
||||
command = ["kubectl", "--request-timeout=20s", "--namespace=vllm"]
|
||||
if handle.get("context"):
|
||||
command += ["--context", handle["context"]]
|
||||
return [*command, *args]
|
||||
|
||||
def _kubectl(self, handle: dict, *args: str, body: dict | None = None) -> str:
|
||||
result = subprocess.run(self._command(handle, *args), input=json.dumps(body) if body else None,
|
||||
text=True, capture_output=True, timeout=35, check=False)
|
||||
if result.returncode:
|
||||
raise RuntimeError(f"kubectl {args[0]} failed: {result.stderr.strip()}")
|
||||
return result.stdout
|
||||
|
||||
def _job(self, handle: dict) -> dict | None:
|
||||
raw = self._kubectl(handle, "get", "job", handle["name"], "--ignore-not-found", "-o", "json")
|
||||
return json.loads(raw) if raw.strip() else None
|
||||
|
||||
def _pods(self, handle: dict) -> list[dict]:
|
||||
raw = self._kubectl(handle, "get", "pods", "-l", f"job-name={handle['name']}", "-o", "json")
|
||||
return json.loads(raw)["items"]
|
||||
|
||||
def manifest(self, handle: dict, request: dict, lane: dict) -> dict:
|
||||
env = [{"name": key, "value": value} for key, value in self._environment(request, lane).items()]
|
||||
if self.config.get("hf_secret"):
|
||||
env.append({"name": "HF_TOKEN", "valueFrom": {"secretKeyRef": {
|
||||
"name": self.config["hf_secret"], "key": self.config.get("hf_secret_key", "HF_TOKEN")}}})
|
||||
volumes = [{"name": "workspace", "emptyDir": {}}, {"name": "shm", "emptyDir": {
|
||||
"medium": "Memory", "sizeLimit": self.config.get("shared_memory", "8Gi")}}]
|
||||
mounts = [{"name": "workspace", "mountPath": "/workspace"}, {"name": "shm", "mountPath": "/dev/shm"}]
|
||||
for kind, path in (("cache", "/ci-cache"), ("artifacts", "/workspace/artifacts")):
|
||||
if self.config.get(f"{kind}_pvc"):
|
||||
subpath = _subpath(self.config[f"{kind}_subpath"])
|
||||
if kind == "artifacts":
|
||||
subpath += "/" + handle["name"]
|
||||
volumes.append({"name": kind, "persistentVolumeClaim": {"claimName": self.config[f"{kind}_pvc"]}})
|
||||
mounts.append({"name": kind, "mountPath": path, "subPath": subpath, "readOnly": kind == "cache"})
|
||||
resources = {"cpu": str(self.config.get("cpu", "8")), "memory": self.config.get("memory", "64Gi"),
|
||||
"nvidia.com/gpu": str(lane["gpus"])}
|
||||
labels = {"app.kubernetes.io/managed-by": "fastvideo-gpu-ci", "fastvideo-ci/run-id": handle["run_id"]}
|
||||
manifest = {"apiVersion": "batch/v1", "kind": "Job", "metadata": {
|
||||
"name": handle["name"], "namespace": "vllm", "labels": labels}, "spec": {
|
||||
"backoffLimit": 0, "activeDeadlineSeconds": int(lane["timeout_seconds"]),
|
||||
"ttlSecondsAfterFinished": 86400, "template": {"metadata": {"labels": labels}, "spec": {
|
||||
"restartPolicy": "Never", "automountServiceAccountToken": False,
|
||||
"hostNetwork": False, "hostPID": False, "hostIPC": False,
|
||||
"nodeSelector": {"kubernetes.io/arch": "arm64", "nvidia.com/gpu.product": "NVIDIA-GB200"},
|
||||
"tolerations": [{"key": "nvidia.com/gpu", "operator": "Exists", "effect": "NoSchedule"}],
|
||||
"containers": [{"name": "worker", "image": self.image,
|
||||
"command": ["/bin/bash", "-c", (self.trusted_dir / "worker.sh").read_text()],
|
||||
"env": env, "resources": {"requests": resources, "limits": resources},
|
||||
"securityContext": {"privileged": False, "allowPrivilegeEscalation": False,
|
||||
"capabilities": {"drop": ["ALL"], "add": [
|
||||
"CHOWN", "DAC_OVERRIDE", "FOWNER", "SETGID", "SETUID"]}},
|
||||
"volumeMounts": mounts}], "volumes": volumes}}}}
|
||||
if self.config.get("artifacts_pvc"):
|
||||
# Only this fixed init command sees the CI artifact parent. PR code
|
||||
# mounts its own run directory and cannot alter adjacent runs.
|
||||
manifest["spec"]["template"]["spec"]["initContainers"] = [{
|
||||
"name": "prepare-artifacts", "image": self.image,
|
||||
"command": ["mkdir", "-p", "/ci-artifacts/" + handle["name"]],
|
||||
"securityContext": {"allowPrivilegeEscalation": False, "capabilities": {"drop": ["ALL"]}},
|
||||
"volumeMounts": [{"name": "artifacts", "mountPath": "/ci-artifacts",
|
||||
"subPath": _subpath(self.config["artifacts_subpath"])}]}]
|
||||
return manifest
|
||||
|
||||
@staticmethod
|
||||
def _owned(job: dict, handle: dict) -> bool:
|
||||
return job.get("metadata", {}).get("labels", {}).get("fastvideo-ci/run-id") == handle["run_id"]
|
||||
|
||||
def run(self, handle: dict, request: dict, lane: dict, cancel: threading.Event) -> int:
|
||||
if self._job(handle) is not None or self._pods(handle):
|
||||
raise RuntimeError("Resource already exists; explicit reconciliation is required")
|
||||
manifest = self.manifest(handle, request, lane)
|
||||
(self.run_dir / f"{handle['name']}.job.json").write_text(json.dumps(manifest, indent=2))
|
||||
if not self.config.get("artifacts_pvc"):
|
||||
(self.run_dir / f"{handle['name']}.artifacts.txt").write_text(
|
||||
"No dedicated artifacts PVC configured. Controller logs/status survive; worker artifacts are ephemeral.\n")
|
||||
if cancel.is_set():
|
||||
return 130
|
||||
marker = self.run_dir / f"{handle['name']}.creating"
|
||||
marker.touch()
|
||||
self._kubectl(handle, "create", "-f", "-", body=manifest)
|
||||
marker.unlink()
|
||||
deadline = time.monotonic() + int(lane["timeout_seconds"])
|
||||
stream = None
|
||||
log_file = (self.run_dir / f"{handle['name']}.log").open("w")
|
||||
try:
|
||||
while True:
|
||||
if cancel.is_set() or time.monotonic() >= deadline:
|
||||
if not self.stop(handle):
|
||||
raise RuntimeError("Cannot confirm Kubernetes job cancellation")
|
||||
return 130 if cancel.is_set() else 124
|
||||
job = self._job(handle)
|
||||
if job is None or not self._owned(job, handle):
|
||||
raise RuntimeError("Kubernetes job disappeared or its ownership changed")
|
||||
pods = self._pods(handle)
|
||||
if len(pods) > 1:
|
||||
raise RuntimeError("Unexpected replacement or duplicate CI pods")
|
||||
for pod in pods:
|
||||
owners = pod.get("metadata", {}).get("ownerReferences", [])
|
||||
if not any(owner.get("uid") == job["metadata"]["uid"] for owner in owners):
|
||||
raise RuntimeError("Pod does not belong to the expected job")
|
||||
statuses = pod.get("status", {}).get("containerStatuses", [])
|
||||
worker = next((status for status in statuses if status["name"] == "worker"), {})
|
||||
state = worker.get("state", {})
|
||||
if stream is None and ("running" in state or "terminated" in state):
|
||||
stream = subprocess.Popen(self._command(handle, "logs", "-f", pod["metadata"]["name"],
|
||||
"-c", "worker", "--timestamps"),
|
||||
stdout=log_file, stderr=subprocess.STDOUT, text=True)
|
||||
terminated = state.get("terminated")
|
||||
if terminated is not None:
|
||||
exit_code = terminated.get("exitCode")
|
||||
if type(exit_code) is not int:
|
||||
raise RuntimeError("Worker termination has no numeric exit status")
|
||||
(self.run_dir / f"{handle['name']}.status.json").write_text(json.dumps(pod["status"], indent=2))
|
||||
# Streaming may fail independently of the pod. Fetch a
|
||||
# final complete log before accepting even a zero exit.
|
||||
final_log = self._kubectl(handle, "logs", pod["metadata"]["name"], "-c", "worker", "--timestamps")
|
||||
if not final_log.strip():
|
||||
raise RuntimeError("Worker completed but Kubernetes logs are empty")
|
||||
(self.run_dir / f"{handle['name']}.final.log").write_text(final_log)
|
||||
return exit_code
|
||||
if pod.get("status", {}).get("phase") == "Failed":
|
||||
raise RuntimeError("Pod failed without a worker exit status")
|
||||
if any(condition.get("type") == "Failed" and condition.get("status") == "True"
|
||||
for condition in job.get("status", {}).get("conditions", [])):
|
||||
raise RuntimeError("Kubernetes job failed before worker completion")
|
||||
cancel.wait(2)
|
||||
finally:
|
||||
if stream is not None:
|
||||
try:
|
||||
stream.wait(timeout=10)
|
||||
except subprocess.TimeoutExpired:
|
||||
stream.terminate()
|
||||
try:
|
||||
stream.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
stream.kill()
|
||||
stream.wait()
|
||||
log_file.close()
|
||||
|
||||
def stop(self, handle: dict) -> bool:
|
||||
job = self._job(handle)
|
||||
if job is not None:
|
||||
if not self._owned(job, handle):
|
||||
return False
|
||||
self._kubectl(handle, "delete", "job", handle["name"], "--cascade=foreground",
|
||||
"--wait=true", "--timeout=20s")
|
||||
(self.run_dir / f"{handle['name']}.creating").unlink(missing_ok=True)
|
||||
elif (self.run_dir / f"{handle['name']}.creating").exists():
|
||||
# A timed-out create may still commit remotely. Keep its GPU lease.
|
||||
return False
|
||||
# A missing Job alone is insufficient: orphaned Pods may still use GPUs.
|
||||
return self._job(handle) is None and not self._pods(handle)
|
||||
|
||||
|
||||
class ModalBackend(Backend):
|
||||
name = "modal"
|
||||
|
||||
def __init__(self, config: dict, trusted_dir: Path, run_dir: Path):
|
||||
super().__init__(config, trusted_dir, run_dir)
|
||||
if config.get("gpu", "L40S") != "L40S":
|
||||
raise ValueError("Modal default must remain L40S; Hopper lanes are selected by reviewed lane policy")
|
||||
self._sandboxes: dict[str, Any] = {}
|
||||
|
||||
@staticmethod
|
||||
def _modal() -> Any:
|
||||
return importlib.import_module("modal")
|
||||
|
||||
@staticmethod
|
||||
def _gpu_type(lane: dict) -> str:
|
||||
gpu_type = lane.get("modal_gpu", "L40S")
|
||||
if gpu_type not in ("L40S", "H100"):
|
||||
raise ValueError("Reviewed Modal hardware must be L40S or H100")
|
||||
return gpu_type
|
||||
|
||||
def _environment(self, request: dict, lane: dict) -> dict[str, str]:
|
||||
self._gpu_type(lane)
|
||||
# Attention/reference policy differs from the GB200 lanes. Copy the
|
||||
# lane record so concurrent tasks never mutate shared policy or image.
|
||||
modal_lane = {**lane, "fa4": lane.get("modal_fa4", lane["fa4"])}
|
||||
return super()._environment(request, modal_lane)
|
||||
|
||||
def handle(self, build_id: str, lane_id: str) -> dict:
|
||||
result = super().handle(build_id, lane_id)
|
||||
result["app"] = self.config.get("app", "fastvideo-gpu-ci")
|
||||
return result
|
||||
|
||||
def _lookup(self, handle: dict) -> Any:
|
||||
if handle["app"] != self.config.get("app", "fastvideo-gpu-ci"):
|
||||
raise ValueError("Handle does not match configured Modal app")
|
||||
modal = self._modal()
|
||||
try:
|
||||
return modal.Sandbox.from_name(handle["app"], handle["name"])
|
||||
except modal.exception.NotFoundError:
|
||||
return None
|
||||
|
||||
def run(self, handle: dict, request: dict, lane: dict, cancel: threading.Event) -> int:
|
||||
env = self._environment(request, lane)
|
||||
if self._lookup(handle) is not None:
|
||||
raise RuntimeError("Sandbox already exists; explicit reconciliation is required")
|
||||
modal = self._modal()
|
||||
image = modal.Image.from_registry(self.image).add_local_file(
|
||||
str(self.trusted_dir / "worker.sh"), "/opt/fastvideo-ci/worker.sh")
|
||||
app = modal.App.lookup(handle["app"], create_if_missing=True)
|
||||
secrets = [modal.Secret.from_name(self.config["hf_secret"])] if self.config.get("hf_secret") else []
|
||||
(self.run_dir / f"{handle['name']}.artifacts.txt").write_text(
|
||||
"Modal adapter retains controller logs/status. Worker artifacts are ephemeral.\n")
|
||||
if cancel.is_set():
|
||||
return 130
|
||||
marker = self.run_dir / f"{handle['name']}.creating"
|
||||
marker.touch()
|
||||
sandbox = modal.Sandbox.create(
|
||||
"/bin/bash", "/opt/fastvideo-ci/worker.sh", app=app, name=handle["name"], image=image,
|
||||
env=env, secrets=secrets, gpu=f"{self._gpu_type(lane)}:{lane['gpus']}",
|
||||
timeout=int(lane["timeout_seconds"]),
|
||||
cpu=float(self.config.get("cpu", 8)), memory=int(self.config.get("memory_mib", 65536)),
|
||||
tags={"fastvideo-ci-run-id": handle["run_id"]})
|
||||
self._sandboxes[handle["name"]] = sandbox
|
||||
marker.unlink()
|
||||
(self.run_dir / f"{handle['name']}.sandbox.json").write_text(json.dumps({"id": sandbox.object_id}))
|
||||
return self._watch(sandbox, handle, lane, cancel)
|
||||
|
||||
def _watch(self, sandbox: Any, handle: dict, lane: dict, cancel: threading.Event) -> int:
|
||||
errors: list[Exception] = []
|
||||
|
||||
def capture(stream: Any, suffix: str) -> None:
|
||||
try:
|
||||
with (self.run_dir / f"{handle['name']}.{suffix}.log").open("w") as output:
|
||||
for line in stream:
|
||||
output.write(line)
|
||||
output.flush()
|
||||
except Exception as error:
|
||||
errors.append(error)
|
||||
|
||||
threads = [threading.Thread(target=capture, args=(sandbox.stdout, "stdout"), daemon=True),
|
||||
threading.Thread(target=capture, args=(sandbox.stderr, "stderr"), daemon=True)]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
deadline = time.monotonic() + int(lane["timeout_seconds"])
|
||||
try:
|
||||
while True:
|
||||
if cancel.is_set() or time.monotonic() >= deadline:
|
||||
if not self.stop(handle):
|
||||
raise RuntimeError("Cannot confirm Modal sandbox cancellation")
|
||||
return 130 if cancel.is_set() else 124
|
||||
code = sandbox.poll()
|
||||
if code is not None:
|
||||
if type(code) is not int:
|
||||
raise RuntimeError("Sandbox returned a nonnumeric exit status")
|
||||
(self.run_dir / f"{handle['name']}.status.json").write_text(json.dumps({"exit_code": code}))
|
||||
for thread in threads:
|
||||
thread.join(timeout=5)
|
||||
if errors or any(thread.is_alive() for thread in threads):
|
||||
raise RuntimeError("Modal log collection did not complete")
|
||||
return code
|
||||
cancel.wait(2)
|
||||
finally:
|
||||
# Do not detach a running sandbox: root must confirm stop before
|
||||
# freeing its shared GPU lease, including after SDK/network errors.
|
||||
for thread in threads:
|
||||
thread.join(timeout=1)
|
||||
|
||||
def stop(self, handle: dict) -> bool:
|
||||
sandbox = self._sandboxes.get(handle["name"]) or self._lookup(handle)
|
||||
marker = self.run_dir / f"{handle['name']}.creating"
|
||||
if sandbox is None:
|
||||
return not marker.exists()
|
||||
if sandbox.get_tags().get("fastvideo-ci-run-id") != handle["run_id"]:
|
||||
return False
|
||||
code = sandbox.terminate(wait=True)
|
||||
confirmed = type(code) is int and sandbox.poll() is not None
|
||||
if confirmed:
|
||||
marker.unlink(missing_ok=True)
|
||||
sandbox.detach()
|
||||
self._sandboxes.pop(handle["name"], None)
|
||||
return confirmed
|
||||
@@ -0,0 +1,25 @@
|
||||
{
|
||||
"repository": "https://github.com/hao-ai-lab/FastVideo.git",
|
||||
"default_backend": "slurm",
|
||||
"slurm_uploader": ["/opt/fastvideo-ci-runner/pipeline-upload"],
|
||||
"queue": "gpu-ci-dispatch",
|
||||
"state_path": "/var/lib/fastvideo-gpu-ci/admission.sqlite3",
|
||||
"artifacts_dir": "/var/lib/fastvideo-gpu-ci/runs",
|
||||
"max_active_prs": 2,
|
||||
"max_gpus_per_pr": 4,
|
||||
"max_gpus": 8,
|
||||
"queue_timeout_seconds": 21600,
|
||||
"upload_artifacts": true,
|
||||
"kubernetes": {
|
||||
"namespace": "vllm",
|
||||
"image": "ghcr.io/hao-ai-lab/fastvideo/fastvideo-dev@sha256:REPLACE_WITH_REVIEWED_ARM64_SM100_DIGEST",
|
||||
"cpu": "8",
|
||||
"memory": "64Gi",
|
||||
"shared_memory": "8Gi"
|
||||
},
|
||||
"modal": {
|
||||
"app": "fastvideo-gpu-ci",
|
||||
"gpu": "L40S",
|
||||
"image": "ghcr.io/hao-ai-lab/fastvideo/fastvideo-dev@sha256:REPLACE_WITH_REVIEWED_AMD64_SM89_DIGEST"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
"""Suite lifecycle and local-host recovery for opt-in Buildkite GPU backends."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import hashlib
|
||||
import json
|
||||
import signal
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import Future, ThreadPoolExecutor
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterator
|
||||
|
||||
from .admission import AdmissionStore
|
||||
|
||||
INFRASTRUCTURE_FAILURE = 97
|
||||
|
||||
|
||||
def run_directory(config: dict[str, Any], build_id: str) -> Path:
|
||||
digest = hashlib.sha256(build_id.encode()).hexdigest()[:32]
|
||||
return Path(config["artifacts_dir"]) / digest
|
||||
|
||||
|
||||
@contextmanager
|
||||
def build_lock(directory: Path) -> Iterator[None]:
|
||||
"""Serialize launch and recovery; never stop a resource during its creation."""
|
||||
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
with (directory / "dispatcher.lock").open("a") as lock:
|
||||
try:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError as error:
|
||||
raise RuntimeError("A dispatcher is still running; cancel its Buildkite job first") from error
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
|
||||
|
||||
|
||||
def make_store(config: dict[str, Any]) -> AdmissionStore:
|
||||
return AdmissionStore(config["state_path"], config.get("max_active_prs", 2),
|
||||
config.get("max_gpus_per_pr", 4), config.get("max_gpus", 8))
|
||||
|
||||
|
||||
def make_backend(config: dict[str, Any], backend: str, directory: Path) -> Any:
|
||||
from .backends import KubernetesBackend, ModalBackend
|
||||
cls = {"vllm": KubernetesBackend, "modal": ModalBackend}.get(backend)
|
||||
if cls is None:
|
||||
raise ValueError("The new dispatcher supports only modal and vllm")
|
||||
backend_config = config["kubernetes" if backend == "vllm" else "modal"]
|
||||
return cls(backend_config, Path(__file__).resolve().parent, directory)
|
||||
|
||||
|
||||
def _run_lane(backend: Any, handle: dict[str, Any], request: dict[str, Any], lane: dict[str, Any],
|
||||
cancel: threading.Event) -> tuple[int, bool, str]:
|
||||
error = ""
|
||||
try:
|
||||
code = backend.run(handle, request, lane, cancel)
|
||||
if type(code) is not int or not 0 <= code <= 255:
|
||||
raise RuntimeError("Backend did not return a valid process exit status")
|
||||
except Exception as exception:
|
||||
code, error = INFRASTRUCTURE_FAILURE, str(exception)
|
||||
try:
|
||||
stopped = backend.stop(handle)
|
||||
except Exception as exception:
|
||||
stopped = False
|
||||
error += f"; stop failed: {exception}"
|
||||
if not stopped:
|
||||
code = INFRASTRUCTURE_FAILURE
|
||||
error += "; worker termination unconfirmed; reservation retained"
|
||||
return code, stopped, error
|
||||
|
||||
|
||||
def execute(request: dict[str, Any], config: dict[str, Any], *, backend: Any = None,
|
||||
cancel: threading.Event | None = None, poll_seconds: float = 2.0) -> int:
|
||||
"""Run one suite; independent processes share the PR/GPU admission ledger."""
|
||||
cancel = cancel or threading.Event()
|
||||
build_id = request["build_id"]
|
||||
directory = run_directory(config, build_id)
|
||||
store = make_store(config)
|
||||
if any(lane["gpus"] > min(store.max_gpus_per_pr, store.max_gpus) for lane in request["lanes"]):
|
||||
raise ValueError("Selected lane does not fit the configured GPU budget")
|
||||
backend = backend or make_backend(config, request["backend"], directory)
|
||||
results: dict[str, Any] = {}
|
||||
code = 0
|
||||
with build_lock(directory):
|
||||
# Never reattach to an ambiguous prior launch. Buildkite retries have
|
||||
# a new job ID; abandoned attempts need explicit backend reconciliation.
|
||||
if any(item["build_id"] == build_id for item in store.snapshot()["builds"]):
|
||||
raise RuntimeError("Build attempt already exists; inspect/recover it instead of replaying it")
|
||||
store.register(build_id, request["pr_key"], request["backend"], request["commit"])
|
||||
(directory / "request.json").write_text(json.dumps(request, indent=2) + "\n")
|
||||
deadline = time.monotonic() + config.get("queue_timeout_seconds", 21600)
|
||||
# A docs-only merge plan requires no GPUs and must not wait behind
|
||||
# unrelated PRs. Arrival order is not evidence of commit recency, so
|
||||
# never cancel other SHAs merely because this request arrived later.
|
||||
admitted = not request["lanes"]
|
||||
while not admitted:
|
||||
store.heartbeat(build_id)
|
||||
if cancel.is_set() or store.is_cancelled(build_id) or time.monotonic() >= deadline:
|
||||
cancel.set()
|
||||
store.request_cancel(build_id)
|
||||
store.finish(build_id)
|
||||
code = INFRASTRUCTURE_FAILURE
|
||||
break
|
||||
admitted = store.try_admit(build_id)
|
||||
if not admitted:
|
||||
cancel.wait(poll_seconds)
|
||||
pending = list(request["lanes"]) if admitted else []
|
||||
golden_selected = any(lane["key"] == "golden-gate" for lane in pending)
|
||||
running: dict[Future[tuple[int, bool, str]], dict[str, Any]] = {}
|
||||
# Lane concurrency is additionally bounded by transactional GPU leases.
|
||||
with ThreadPoolExecutor(max_workers=4) as pool:
|
||||
while pending or running:
|
||||
store.heartbeat(build_id)
|
||||
if store.is_cancelled(build_id):
|
||||
cancel.set()
|
||||
if cancel.is_set():
|
||||
store.request_cancel(build_id)
|
||||
for lane in pending:
|
||||
results[lane["key"]] = {"state": "cancelled"}
|
||||
pending.clear()
|
||||
code = code or INFRASTRUCTURE_FAILURE
|
||||
for future, lane in list(running.items()):
|
||||
if not future.done():
|
||||
continue
|
||||
lane_code, stopped, error = future.result()
|
||||
if stopped:
|
||||
store.release(build_id, lane["key"])
|
||||
else:
|
||||
cancel.set()
|
||||
results[lane["key"]] = {"state": "passed" if lane_code == 0 else "failed",
|
||||
"exit_code": lane_code, "stopped": stopped, "error": error}
|
||||
print(f"--- {lane['key']}: exit {lane_code}", flush=True)
|
||||
code = code or lane_code
|
||||
del running[future]
|
||||
for lane in list(pending):
|
||||
if cancel.is_set() or len(running) >= 4:
|
||||
break
|
||||
if golden_selected and not lane["fastcheck"] and lane["key"] != "golden-gate":
|
||||
golden = results.get("golden-gate")
|
||||
if golden is None:
|
||||
continue
|
||||
if golden["state"] != "passed":
|
||||
results[lane["key"]] = {"state": "skipped", "reason": "golden gate failed"}
|
||||
pending.remove(lane)
|
||||
continue
|
||||
handle = backend.handle(build_id, lane["key"])
|
||||
if store.try_acquire(build_id, lane["key"], lane["gpus"], handle):
|
||||
print(f"--- Starting {lane['key']} on {request['backend']} ({lane['gpus']} GPUs)", flush=True)
|
||||
future = pool.submit(_run_lane, backend, handle, request, lane, cancel)
|
||||
running[future] = lane
|
||||
pending.remove(lane)
|
||||
if pending or running:
|
||||
# wait() on an already-set Event spins; cancellation still
|
||||
# polls bounded backend shutdowns at the normal interval.
|
||||
time.sleep(poll_seconds)
|
||||
active = [a for a in store.snapshot()["allocations"] if a["build_id"] == build_id and a["state"] == "active"]
|
||||
if not active:
|
||||
store.finish(build_id)
|
||||
else:
|
||||
code = INFRASTRUCTURE_FAILURE
|
||||
summary = {"build_id": build_id, "backend": request["backend"], "commit": request["commit"],
|
||||
"exit_code": code, "recovery_required": bool(active), "lanes": results}
|
||||
(directory / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
|
||||
print(json.dumps(summary, indent=2), flush=True)
|
||||
return code
|
||||
|
||||
|
||||
def recover(config: dict[str, Any], build_id: str, *, backend: Any = None) -> None:
|
||||
"""Stop abandoned workers first, then release reservations. Never TTL-unlock."""
|
||||
store = make_store(config)
|
||||
build = next((b for b in store.snapshot()["builds"] if b["build_id"] == build_id), None)
|
||||
if build is None:
|
||||
raise ValueError("Unknown build ID")
|
||||
directory = run_directory(config, build_id)
|
||||
with build_lock(directory):
|
||||
store.request_cancel(build_id)
|
||||
backend = backend or make_backend(config, build["backend"], directory)
|
||||
uncertain = []
|
||||
for allocation in store.snapshot()["allocations"]:
|
||||
if allocation["build_id"] != build_id or allocation["state"] != "active":
|
||||
continue
|
||||
try:
|
||||
stopped = backend.stop(allocation["handle"])
|
||||
except Exception:
|
||||
stopped = False
|
||||
if stopped:
|
||||
store.release(build_id, allocation["lane_id"])
|
||||
else:
|
||||
uncertain.append(allocation["lane_id"])
|
||||
if uncertain:
|
||||
raise RuntimeError(f"Could not confirm termination; reservations retained: {uncertain}")
|
||||
store.finish(build_id)
|
||||
|
||||
|
||||
def upload_artifacts(directory: Path) -> None:
|
||||
"""Only upload controller-owned logs/metadata, never arbitrary PR globs."""
|
||||
for path in sorted(directory.iterdir()):
|
||||
if path.is_file() and not path.is_symlink() and path.suffix in (".json", ".log"):
|
||||
subprocess.run(["buildkite-agent", "artifact", "upload", path.name], cwd=directory, check=True)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def cancellation_signals(event: threading.Event) -> Iterator[None]:
|
||||
previous: dict[signal.Signals, Any] = {}
|
||||
def handler(signum: int, frame: Any) -> None:
|
||||
event.set()
|
||||
for signum in (signal.SIGINT, signal.SIGTERM):
|
||||
previous[signum] = signal.signal(signum, handler)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
for signum, old_handler in previous.items():
|
||||
signal.signal(signum, old_handler)
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Launch with python -I to exclude job-controlled PYTHONPATH and user packages."""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
|
||||
from scripts.gpu_ci.__main__ import main # noqa: E402
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,69 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
"""Render an opt-in GPU backend without changing the canonical Slurm graph.
|
||||
|
||||
The caller is the operator-owned pipeline uploader. It must load this module
|
||||
from its reviewed installation, never from a pull request checkout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
BACKENDS = frozenset({"slurm", "modal", "vllm"})
|
||||
STATUS_SUFFIXES = {
|
||||
"fastcheck": "fastcheck-passed",
|
||||
"full": "full-suite-passed",
|
||||
"merge": "full-suite-passed",
|
||||
"direct": "direct-test-completed",
|
||||
"scheduled": "scheduled-ssim-passed",
|
||||
}
|
||||
|
||||
|
||||
def render_pipeline(
|
||||
canonical: dict[str, Any],
|
||||
backend: str,
|
||||
scope: str,
|
||||
queue: str = "gpu-ci-dispatch",
|
||||
) -> dict[str, Any]:
|
||||
"""Return the original graph or one trusted, backend-specific suite step.
|
||||
|
||||
Only enumerated backend and scope values reach the generated step. Request
|
||||
SHA, PR identity, direct lane, and merge plan remain build metadata and are
|
||||
validated independently by the trusted dispatcher. Arbitrary canonical
|
||||
graph commands, environment, hooks, and notifications are not copied into
|
||||
the new control plane.
|
||||
"""
|
||||
backend = backend or "slurm"
|
||||
scope = scope or "fastcheck"
|
||||
if backend not in BACKENDS:
|
||||
raise ValueError(f"Unsupported GPU CI backend: {backend!r}")
|
||||
if scope not in STATUS_SUFFIXES:
|
||||
raise ValueError(f"Unsupported GPU CI scope: {scope!r}")
|
||||
if backend == "slurm":
|
||||
return copy.deepcopy(canonical)
|
||||
if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}", queue):
|
||||
raise ValueError("GPU CI queue must be an operator-configured queue name")
|
||||
return {
|
||||
"notify": [{
|
||||
"github_commit_status": {
|
||||
"context": f"gpu-ci/{backend}/{STATUS_SUFFIXES[scope]}"
|
||||
}
|
||||
}],
|
||||
"steps": [{
|
||||
"label": f"GPU CI ({backend}, {scope})",
|
||||
"key": f"gpu-ci-{backend}-{scope}",
|
||||
"command": "/opt/fastvideo-gpu-ci/run",
|
||||
# Admission waiting happens inside this command and uses this
|
||||
# budget. An unscheduled Buildkite job has no command timeout yet.
|
||||
"timeout_in_minutes": 480,
|
||||
"env": {
|
||||
"CI_GPU_BACKEND": backend,
|
||||
"TEST_SCOPE": scope,
|
||||
},
|
||||
"agents": {
|
||||
"queue": queue
|
||||
},
|
||||
}],
|
||||
}
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Reviewed lane policy and validation at the trusted dispatcher boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Mapping
|
||||
|
||||
BACKENDS = ("slurm", "modal", "vllm")
|
||||
SCOPES = ("fastcheck", "full", "merge", "direct", "scheduled")
|
||||
DEFAULT_REPOSITORY = "https://github.com/hao-ai-lab/FastVideo.git"
|
||||
|
||||
|
||||
def _lane(key: str, public_type: str, script: str, gpus: int = 1, *, fastcheck: bool = False,
|
||||
extras: tuple[str, ...] = ("test",), fa4: str = "0") -> dict[str, Any]:
|
||||
return {"key": key, "public_type": public_type, "script": script, "gpus": gpus,
|
||||
"fastcheck": fastcheck, "extras": list(extras), "fa4": fa4, "timeout_seconds": 5400,
|
||||
"kernel": key != "dreamverse",
|
||||
"modal_gpu": "H100" if key in {"encoder", "kernel-tests", "training-vsa"} else "L40S",
|
||||
"modal_fa4": "0" if key in {"transformer", "training", "distillation", "self-forcing",
|
||||
"lora-training", "training-vsa", "train-framework"} else "1"}
|
||||
|
||||
|
||||
LANES = (
|
||||
_lane("encoder", "encoder", "lanes/encoder.sh", fastcheck=True),
|
||||
_lane("vae", "vae", "lanes/vae.sh", fastcheck=True),
|
||||
_lane("transformer", "transformer", "lanes/transformer.sh", fastcheck=True),
|
||||
_lane("kernel-tests", "kernel_tests", "lanes/kernel_tests.sh", fastcheck=True),
|
||||
_lane("unit", "unit_test", "unit_test.sh", fastcheck=True),
|
||||
_lane("dreamverse", "dreamverse_app", "lanes/dreamverse.sh", fastcheck=True,
|
||||
extras=("test", "dreamverse")),
|
||||
_lane("golden-gate", "golden_gate", "lanes/golden_gate.sh"),
|
||||
_lane("ssim", "ssim", "lanes/ssim.sh", 4, fa4="1"),
|
||||
_lane("lora-inference", "inference_lora", "lanes/inference_lora.sh"),
|
||||
_lane("lora-extraction", "lora_extraction", "lanes/lora_extraction.sh"),
|
||||
_lane("training", "training", "lanes/training.sh", 4),
|
||||
_lane("distillation", "distillation_dmd", "lanes/distillation_dmd.sh", 2),
|
||||
_lane("self-forcing", "self_forcing", "lanes/self_forcing.sh", 2),
|
||||
_lane("lora-training", "training_lora", "lanes/training_lora.sh", 2),
|
||||
_lane("training-vsa", "training_vsa", "lanes/training_vsa.sh", 2),
|
||||
_lane("inference-vmoba", "inference_vmoba", "lanes/inference_vmoba.sh"),
|
||||
_lane("performance", "performance", "lanes/performance.sh", 2),
|
||||
_lane("api-server", "api_server", "lanes/api_server.sh"),
|
||||
_lane("train-framework", "train_framework", "lanes/train_framework.sh"),
|
||||
_lane("eval", "eval", "lanes/eval.sh", extras=("test", "eval-full")),
|
||||
)
|
||||
LANES_BY_KEY = {lane["key"]: lane for lane in LANES}
|
||||
|
||||
|
||||
def backend_name(value: str | None, default: str = "slurm") -> str:
|
||||
backend = value or default
|
||||
if backend not in BACKENDS:
|
||||
raise ValueError(f"Unknown CI_GPU_BACKEND: {backend!r}")
|
||||
return backend
|
||||
|
||||
|
||||
def _identifier(value: str, label: str) -> str:
|
||||
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,99}", value):
|
||||
raise ValueError(f"Invalid {label}")
|
||||
return value
|
||||
|
||||
|
||||
def build_request(env: Mapping[str, str], config: Mapping[str, Any], root: Path) -> dict[str, Any]:
|
||||
"""Copy only the explicitly accepted Buildkite fields; never forward its token."""
|
||||
backend = backend_name(env.get("CI_GPU_BACKEND"), config.get("default_backend", "slurm"))
|
||||
if backend == "slurm":
|
||||
raise ValueError("Slurm uses the existing trusted uploader and dispatcher")
|
||||
repository = config.get("repository", DEFAULT_REPOSITORY)
|
||||
if not re.fullmatch(r"https://github\.com/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+\.git", repository):
|
||||
raise ValueError("repository must be a fixed HTTPS GitHub .git URL")
|
||||
supplied_repo = env.get("BUILDKITE_REPO", "")
|
||||
ssh_repo = "git@github.com:" + repository.removeprefix("https://github.com/")
|
||||
if supplied_repo not in (repository, ssh_repo):
|
||||
raise ValueError("Buildkite repository does not match operator policy")
|
||||
commit = env.get("BUILDKITE_COMMIT", "")
|
||||
if not re.fullmatch(r"[0-9a-f]{40}", commit):
|
||||
raise ValueError("BUILDKITE_COMMIT must be an immutable 40-character SHA")
|
||||
build_id = _identifier(env.get("BUILDKITE_BUILD_ID", ""), "BUILDKITE_BUILD_ID")
|
||||
job_id = _identifier(env.get("BUILDKITE_JOB_ID", ""), "BUILDKITE_JOB_ID")
|
||||
pr = env.get("BUILDKITE_PULL_REQUEST", "false")
|
||||
if pr == "false" and env.get("PR_NUMBER"):
|
||||
pr = env["PR_NUMBER"]
|
||||
if pr not in ("", "false") and not re.fullmatch(r"[1-9][0-9]*", pr):
|
||||
raise ValueError("Invalid PR number")
|
||||
scope = env.get("TEST_SCOPE") or "fastcheck"
|
||||
if scope not in SCOPES:
|
||||
raise ValueError(f"Invalid TEST_SCOPE: {scope!r}")
|
||||
# Reference publication requires a separately trusted publishing job.
|
||||
if env.get("FASTVIDEO_SSIM_BOOTSTRAP_MODE", "0") not in ("", "0", "false"):
|
||||
raise ValueError("SSIM bootstrap publication is not supported by the isolated GPU runner")
|
||||
request = {"build_id": f"{build_id}.{job_id}", "buildkite_build_id": build_id,
|
||||
"repository": repository, "commit": commit, "pr_number": pr or "false",
|
||||
"pr_key": f"{repository}#{pr}" if pr not in ("", "false") else None,
|
||||
"backend": backend, "scope": scope, "env": {"TEST_SCOPE": scope,
|
||||
"BUILDKITE_PULL_REQUEST": pr or "false", "BUILDKITE_COMMIT": commit,
|
||||
"BUILDKITE_REPO": repository, "BUILDKITE_BUILD_ID": build_id,
|
||||
"BUILDKITE_JOB_ID": job_id}}
|
||||
for key in ("BUILDKITE_BRANCH", "BUILDKITE_SOURCE", "BUILDKITE_BUILD_URL"):
|
||||
value = env.get(key, "")
|
||||
if len(value) > 2048 or any(ord(c) < 32 for c in value):
|
||||
raise ValueError(f"Invalid {key}")
|
||||
request["env"][key] = value
|
||||
if scope == "direct":
|
||||
requested = env.get("TEST_TYPE", "").removesuffix("_ci")
|
||||
selected = [lane for lane in LANES if requested in (lane["public_type"], lane["key"])]
|
||||
if len(selected) != 1:
|
||||
raise ValueError(f"Unknown direct TEST_TYPE: {requested!r}")
|
||||
elif scope == "merge":
|
||||
encoded = env.get("MERGE_TEST_PLAN", "")
|
||||
if not re.fullmatch(r",(?:none|[a-z]+(?:-[a-z]+)*(?:,[a-z]+(?:-[a-z]+)*)*),", encoded):
|
||||
raise ValueError("Missing or malformed trusted MERGE_TEST_PLAN")
|
||||
names = encoded[1:-1].split(",")
|
||||
if names == ["none"]:
|
||||
names = []
|
||||
if any(name not in LANES_BY_KEY or LANES_BY_KEY[name]["fastcheck"] for name in names):
|
||||
raise ValueError("MERGE_TEST_PLAN contains an unknown or non-integration lane")
|
||||
if len(names) != len(set(names)):
|
||||
raise ValueError("Duplicate MERGE_TEST_PLAN lane")
|
||||
selected = [lane for lane in LANES if lane["key"] in names]
|
||||
elif scope == "scheduled":
|
||||
requested = env.get("TEST_TYPE", "ssim").removesuffix("_ci")
|
||||
if requested != "ssim":
|
||||
raise ValueError("Scheduled scope supports SSIM; performance uses direct scope")
|
||||
selected = [LANES_BY_KEY[requested]]
|
||||
else:
|
||||
selected = [lane for lane in LANES if scope == "full" or lane["fastcheck"]]
|
||||
for key, lane_key, directory in (("FASTVIDEO_GOLDEN_TEST_FILES", "golden-gate", "golden_gate"),
|
||||
("FASTVIDEO_SSIM_TEST_FILES", "ssim", "ssim")):
|
||||
if lane_key not in {lane["key"] for lane in selected}:
|
||||
continue
|
||||
merge_key = "MERGE_GOLDEN_TESTS" if lane_key == "golden-gate" else "MERGE_SSIM_TESTS"
|
||||
value = env.get(merge_key, "") if scope == "merge" else "all"
|
||||
if not value:
|
||||
raise ValueError(f"Missing trusted {key} for merge scope")
|
||||
if value != "all":
|
||||
names = value.split(",")
|
||||
if len(names) != len(set(names)) or any(
|
||||
not re.fullmatch(r"test_[a-z0-9_]+\.py", name) for name in names
|
||||
):
|
||||
raise ValueError(f"Invalid or unknown {key}")
|
||||
# Existence belongs to the isolated exact-SHA checkout. The trusted
|
||||
# controller need not be redeployed whenever a PR adds a test file.
|
||||
request["env"][key] = value
|
||||
request["lanes"] = [{**lane, "script": ".buildkite/scripts/" + lane["script"]} for lane in selected]
|
||||
return request
|
||||
Executable
+6
@@ -0,0 +1,6 @@
|
||||
#!/bin/sh
|
||||
# Install this file at /opt/fastvideo-gpu-ci/run on the trusted dispatch host.
|
||||
set -eu
|
||||
exec /opt/fastvideo-gpu-ci/venv/bin/python -I \
|
||||
/opt/fastvideo-gpu-ci/source/scripts/gpu_ci/entrypoint.py run \
|
||||
--config /etc/fastvideo-gpu-ci.json
|
||||
Executable
+6
@@ -0,0 +1,6 @@
|
||||
#!/bin/sh
|
||||
# Install as the trusted bootstrap for each participating Buildkite pipeline.
|
||||
set -eu
|
||||
exec /opt/fastvideo-gpu-ci/venv/bin/python -I \
|
||||
/opt/fastvideo-gpu-ci/source/scripts/gpu_ci/entrypoint.py upload \
|
||||
--config /etc/fastvideo-gpu-ci.json
|
||||
Executable
+109
@@ -0,0 +1,109 @@
|
||||
#!/usr/bin/env bash
|
||||
# Installed with the trusted controller; never load this entrypoint from a PR.
|
||||
set -euo pipefail
|
||||
umask 077
|
||||
|
||||
[[ ${FASTVIDEO_CI_REPOSITORY:-} == https://github.com/hao-ai-lab/FastVideo.git ]]
|
||||
[[ ${FASTVIDEO_CI_COMMIT:-} =~ ^[0-9a-f]{40}$ ]]
|
||||
[[ ${FASTVIDEO_CI_GPUS:-} =~ ^[1-4]$ ]]
|
||||
[[ ${FASTVIDEO_CI_EXTRAS:-} =~ ^[a-z][a-z0-9-]*(,[a-z][a-z0-9-]*)*$ ]]
|
||||
[[ ${FASTVIDEO_CI_SCRIPT:-} == .buildkite/scripts/unit_test.sh ||
|
||||
${FASTVIDEO_CI_SCRIPT:-} =~ ^\.buildkite/scripts/lanes/[a-z_]+\.sh$ ]]
|
||||
[[ ${FASTVIDEO_CI_KERNEL:-} =~ ^[01]$ ]]
|
||||
[[ ${FASTVIDEO_FA4:-} =~ ^[01]$ ]]
|
||||
|
||||
export PATH="/opt/venv/bin:/root/.local/bin:$PATH"
|
||||
export VIRTUAL_ENV=/opt/venv
|
||||
export HF_HOME=/workspace/hf
|
||||
export HF_HUB_CACHE="$HF_HOME/hub"
|
||||
export UV_CACHE_DIR=/workspace/uv-cache
|
||||
export UV_LINK_MODE=copy
|
||||
export WANDB_MODE=offline
|
||||
export FASTVIDEO_CI_LOCAL_ONLY=1
|
||||
export FASTVIDEO_SSIM_BOOTSTRAP_MODE=0
|
||||
export MASTER_ADDR=127.0.0.1
|
||||
export MASTER_PORT=29500
|
||||
export PYTHONUNBUFFERED=1
|
||||
mkdir -p /workspace/artifacts "$HF_HUB_CACHE"
|
||||
|
||||
# Results are diagnostic only. The controller trusts the container exit status.
|
||||
finish() {
|
||||
worker_rc=$?
|
||||
trap - EXIT
|
||||
printf '{"exit_code":%s,"commit":"%s"}\n' "$worker_rc" "$FASTVIDEO_CI_COMMIT" \
|
||||
> /workspace/artifacts/worker-result.json
|
||||
if [ -d /workspace/repository/fastvideo/tests/ssim/generated_videos ]; then
|
||||
cp -R /workspace/repository/fastvideo/tests/ssim/generated_videos \
|
||||
/workspace/artifacts/ssim-generated 2>/dev/null || true
|
||||
fi
|
||||
exit "$worker_rc"
|
||||
}
|
||||
trap finish EXIT
|
||||
trap 'exit 143' TERM
|
||||
trap 'exit 130' INT
|
||||
|
||||
# The optional cache is a dedicated, read-only HF Hub subtree. Keep mutable
|
||||
# refs and locks private; link immutable blobs instead of duplicating weights.
|
||||
if [ -d /ci-cache ]; then
|
||||
python - <<'PY'
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
source = Path('/ci-cache').resolve()
|
||||
destination = Path('/workspace/hf/hub')
|
||||
for current, directories, filenames in os.walk(source, followlinks=False):
|
||||
if Path(current) == source:
|
||||
directories[:] = [name for name in directories if name.startswith(('models--', 'datasets--', 'spaces--'))]
|
||||
filenames = [] # Never copy a token file from a misconfigured cache root.
|
||||
directories[:] = [name for name in directories if name != '.locks'
|
||||
and not (Path(current) / name).is_symlink()]
|
||||
relative = Path(current).relative_to(source)
|
||||
target_dir = destination / relative
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
for name in filenames:
|
||||
original = Path(current) / name
|
||||
resolved = original.resolve()
|
||||
if not resolved.is_relative_to(source) or not resolved.is_file():
|
||||
raise RuntimeError('Cache contains a file outside its mounted subtree')
|
||||
target = target_dir / name
|
||||
if 'blobs' in relative.parts or original.is_symlink():
|
||||
target.symlink_to(resolved)
|
||||
else:
|
||||
shutil.copyfile(original, target)
|
||||
PY
|
||||
fi
|
||||
|
||||
mkdir /workspace/repository
|
||||
cd /workspace/repository
|
||||
git init --quiet
|
||||
git remote add origin "$FASTVIDEO_CI_REPOSITORY"
|
||||
if ! git fetch --no-tags --depth=1 origin "$FASTVIDEO_CI_COMMIT"; then
|
||||
[[ ${FASTVIDEO_CI_PR_NUMBER:-} =~ ^[1-9][0-9]*$ ]]
|
||||
git fetch --no-tags --depth=1 origin "refs/pull/${FASTVIDEO_CI_PR_NUMBER}/head"
|
||||
fi
|
||||
git checkout --quiet --detach FETCH_HEAD
|
||||
test "$(git rev-parse HEAD)" = "$FASTVIDEO_CI_COMMIT"
|
||||
git -c protocol.file.allow=never submodule update --init --recursive --depth=1
|
||||
|
||||
# Installing dependencies after the kernel can silently replace the PR kernel.
|
||||
printf 'fastvideo-kernel\n' > /workspace/kernel-excludes
|
||||
uv pip install --excludes /workspace/kernel-excludes -e ".[${FASTVIDEO_CI_EXTRAS}]"
|
||||
if [ "$FASTVIDEO_CI_KERNEL" = 1 ]; then
|
||||
python fastvideo/tests/modal/kernel_build_cache.py install
|
||||
fi
|
||||
|
||||
python - <<'PY'
|
||||
import os
|
||||
import torch
|
||||
|
||||
expected = int(os.environ['FASTVIDEO_CI_GPUS'])
|
||||
if not torch.cuda.is_available() or torch.cuda.device_count() != expected:
|
||||
raise RuntimeError(f'Expected exactly {expected} CUDA devices')
|
||||
for index in range(expected):
|
||||
print(f'CUDA {index}: {torch.cuda.get_device_name(index)}', flush=True)
|
||||
PY
|
||||
|
||||
# SSIM runs concurrent pytest subprocesses; a single global JUnit destination
|
||||
# would race. Preserve each lane's native metrics plus the controller logs.
|
||||
bash "$FASTVIDEO_CI_SCRIPT"
|
||||
Reference in New Issue
Block a user