358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""
|
|
BuildRunner — manages build lifecycle.
|
|
|
|
Wraps PipelineEngine with:
|
|
- Build ID generation
|
|
- State tracking (queued, running, complete, failed)
|
|
- Async task management (so /api/build returns immediately)
|
|
- Optional persistence via a storage backend
|
|
|
|
Usage:
|
|
runner = BuildRunner(bus, registry, nodes)
|
|
build_id = await runner.start(BuildBody(cmd="make -j4", dir="/tmp/foo"))
|
|
state = runner.state(build_id)
|
|
all_builds = runner.list_builds()
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from backend.events.bus import EventBus
|
|
from backend.events.schema import EventType
|
|
from backend.pipeline.engine import PipelineEngine
|
|
from backend.cache.minio_cache import MinioCache
|
|
|
|
|
|
# Optional storage backend
|
|
try:
|
|
from backend.storage.sqlite_db import STORAGE
|
|
except Exception:
|
|
STORAGE = None
|
|
|
|
|
|
@dataclass
|
|
class BuildRecord:
|
|
build_id: str
|
|
cmd: str
|
|
dir: str
|
|
project: Optional[str] = None
|
|
arch: str = "x86_64"
|
|
target: str = "linux-gnu"
|
|
toolchain: str = "gcc"
|
|
started_at: float = field(default_factory=time.time)
|
|
ended_at: Optional[float] = None
|
|
status: str = "queued" # queued | running | complete | failed
|
|
actions: List[Dict[str, str]] = field(default_factory=list)
|
|
error: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class BuildRunner:
|
|
"""Manages build lifecycle and tracks all builds in memory (or via storage)."""
|
|
|
|
def __init__(self, bus: EventBus, registry, nodes: List[Dict[str, Any]],
|
|
storage=None):
|
|
self.bus = bus
|
|
self.registry = registry
|
|
self.nodes = nodes
|
|
self.storage = storage if storage is not None else STORAGE # may still be None
|
|
self._builds: Dict[str, BuildRecord] = {}
|
|
self._tasks: Dict[str, asyncio.Task] = {}
|
|
self._engines: Dict[str, PipelineEngine] = {} # build_id -> engine
|
|
self._cache: Optional[MinioCache] = None # lazy init
|
|
|
|
# On startup, load any persisted builds into the in-memory cache
|
|
if self.storage:
|
|
self.load_from_storage()
|
|
|
|
# Wire engines into debugger + pipeline_control so they can
|
|
# pause/resume/step the most-recently-started build.
|
|
self._wire_debugger()
|
|
|
|
def _wire_debugger(self):
|
|
"""Register our engine getter with the debugger + pipeline_control modules."""
|
|
try:
|
|
from backend.api import debugger as dbg
|
|
from backend.api import pipeline_control as pc
|
|
# Replace their _ENGINE singletons with a proxy that always
|
|
# points at the most recent running build's engine.
|
|
proxy = _EngineProxy(self)
|
|
dbg.set_engine(proxy)
|
|
pc.set_engine(proxy)
|
|
except Exception as e:
|
|
# Don't crash if debugger/pipeline_control aren't importable
|
|
pass
|
|
|
|
# -------------------------------------------------
|
|
# CACHE
|
|
# -------------------------------------------------
|
|
def _get_cache(self) -> Optional[MinioCache]:
|
|
"""Lazily construct a MinioCache. Returns None if MinIO is unreachable
|
|
or not configured."""
|
|
if self._cache is not None:
|
|
return self._cache
|
|
try:
|
|
self._cache = MinioCache("localhost:9000", "minioadmin", "minioadmin")
|
|
# Force connection test
|
|
self._cache._ensure_client()
|
|
return self._cache
|
|
except Exception:
|
|
self._cache = None
|
|
return None
|
|
|
|
# -------------------------------------------------
|
|
# START A BUILD
|
|
# -------------------------------------------------
|
|
async def start(self, body) -> str:
|
|
"""Queue a build and return its ID. The actual run happens in a background task."""
|
|
build_id = str(uuid.uuid4())[:8]
|
|
record = BuildRecord(
|
|
build_id=build_id,
|
|
cmd=body.cmd,
|
|
dir=body.dir,
|
|
project=getattr(body, "project", None),
|
|
arch=getattr(body, "arch", "x86_64"),
|
|
target=getattr(body, "target", "linux-gnu"),
|
|
toolchain=getattr(body, "toolchain", "gcc"),
|
|
)
|
|
record.status = "running"
|
|
self._builds[build_id] = record
|
|
self._persist(record)
|
|
|
|
# Launch in background
|
|
task = asyncio.create_task(self._run_build(build_id))
|
|
self._tasks[build_id] = task
|
|
return build_id
|
|
|
|
# -------------------------------------------------
|
|
# BUILD EXECUTION
|
|
# -------------------------------------------------
|
|
async def _run_build(self, build_id: str):
|
|
record = self._builds[build_id]
|
|
try:
|
|
# Construct a project spec the engine can consume
|
|
project = self._build_project_spec(record)
|
|
|
|
# Choose execution mode
|
|
if record.cmd and record.cmd.startswith("make"):
|
|
# Real build — use the actual engine
|
|
engine = PipelineEngine(
|
|
nodes=self.nodes,
|
|
node_registry=self.registry,
|
|
event_bus=self.bus,
|
|
cache=self._get_cache(),
|
|
build_id=build_id,
|
|
)
|
|
self._engines[build_id] = engine
|
|
try:
|
|
results = await engine.run(project)
|
|
finally:
|
|
# Engine done — drop from active set
|
|
self._engines.pop(build_id, None)
|
|
|
|
# Record results
|
|
record.actions = [{"name": n, "state": s} for n, s in results]
|
|
record.status = "failed" if any(s == "failed" for _, s in results) else "complete"
|
|
else:
|
|
# Synthetic pipeline for non-shell build commands (demo mode)
|
|
await self._run_synthetic(build_id, record)
|
|
record.status = "complete"
|
|
|
|
except Exception as e:
|
|
record.status = "failed"
|
|
record.error = f"{type(e).__name__}: {e}"
|
|
self.bus.emit(
|
|
EventType.FAILURE,
|
|
action=None,
|
|
node=None,
|
|
state="failed",
|
|
reason=f"build_crashed:{record.error}",
|
|
meta={"build_id": build_id},
|
|
)
|
|
finally:
|
|
record.ended_at = time.time()
|
|
self._persist(record)
|
|
|
|
def _build_project_spec(self, record: BuildRecord) -> Dict[str, Any]:
|
|
"""Convert a BuildRecord into a project spec the engine understands."""
|
|
return {
|
|
"source": record.project or "local",
|
|
"default_target": record.target,
|
|
"build_dir": record.dir, # passed through to actions as `dir`
|
|
"build_env": {
|
|
"ARCH": record.arch,
|
|
"TARGET": record.target,
|
|
"CC": record.toolchain,
|
|
"DISTCC_HOSTS": " ".join(
|
|
n.get("host", "") for n in self.nodes if n.get("state") == "online"
|
|
),
|
|
},
|
|
"targets": {
|
|
record.target: record.cmd,
|
|
},
|
|
}
|
|
|
|
async def _run_synthetic(self, build_id: str, record: BuildRecord):
|
|
"""Fallback synthetic pipeline — emits a realistic-looking stream of
|
|
events without actually running any commands. Used when the build
|
|
command isn't a real shell command (e.g. demo / smoke test)."""
|
|
import random
|
|
actions = [
|
|
("fetch_source", []),
|
|
("configure", ["fetch_source"]),
|
|
("build_kernel", ["configure"]),
|
|
("build_modules", ["configure"]),
|
|
("build_initramfs", ["configure"]),
|
|
("build_debian", ["build_kernel", "build_modules"]),
|
|
("package", ["build_debian"]),
|
|
("release", ["package"]),
|
|
]
|
|
critical = {"fetch_source", "configure", "build_kernel", "build_debian", "package", "release"}
|
|
|
|
for action, deps in actions:
|
|
node_name = random.choice([n["name"] for n in self.nodes if n.get("state") != "offline"]) if self.nodes else "localhost"
|
|
|
|
self.bus.emit(EventType.TASK_UPDATE, action=action, node=node_name, state="scheduled",
|
|
meta={"deps": deps, "critical": action in critical, "target": record.target, "build_id": build_id})
|
|
await asyncio.sleep(0.2)
|
|
|
|
self.bus.emit(EventType.TASK_UPDATE, action=action, node=node_name, state="running",
|
|
meta={"deps": deps, "critical": action in critical, "target": record.target, "build_id": build_id})
|
|
await asyncio.sleep(random.uniform(0.4, 1.0))
|
|
|
|
cache_hit = random.random() < 0.25
|
|
self.bus.emit(EventType.TASK_UPDATE, action=action, node=node_name, state="done",
|
|
meta={"deps": deps, "cache": "hit" if cache_hit else None, "critical": action in critical, "target": record.target, "build_id": build_id})
|
|
if cache_hit:
|
|
self.bus.emit(EventType.CACHE_UPDATE, action=action, node=node_name, state="hit",
|
|
meta={"build_id": build_id})
|
|
record.actions.append({"name": action, "state": "done"})
|
|
await asyncio.sleep(0.1)
|
|
|
|
# -------------------------------------------------
|
|
# QUERY
|
|
# -------------------------------------------------
|
|
def state(self, build_id: str) -> Optional[Dict[str, Any]]:
|
|
rec = self._builds.get(build_id)
|
|
if rec:
|
|
return rec.to_dict()
|
|
if self.storage:
|
|
try:
|
|
return self.storage.get_build(build_id)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
def list_builds(self) -> List[Dict[str, Any]]:
|
|
# Merge in-memory + storage (in-memory takes precedence)
|
|
in_mem = {b.build_id: b.to_dict() for b in self._builds.values()}
|
|
if not self.storage:
|
|
return list(in_mem.values())
|
|
try:
|
|
stored = {b["build_id"]: b for b in self.storage.list_builds()}
|
|
except Exception:
|
|
stored = {}
|
|
# Merge: in-memory wins (it has fresher state for running builds)
|
|
merged = {**stored, **in_mem}
|
|
return sorted(merged.values(), key=lambda b: b.get("started_at", 0), reverse=True)
|
|
|
|
def cancel(self, build_id: str) -> bool:
|
|
task = self._tasks.get(build_id)
|
|
if task and not task.done():
|
|
task.cancel()
|
|
rec = self._builds.get(build_id)
|
|
if rec:
|
|
rec.status = "cancelled"
|
|
rec.ended_at = time.time()
|
|
self._persist(rec)
|
|
return True
|
|
return False
|
|
|
|
# -------------------------------------------------
|
|
# DEBUGGER CONTROL
|
|
# -------------------------------------------------
|
|
def get_engine(self, build_id: Optional[str] = None) -> Optional[PipelineEngine]:
|
|
"""Look up the running engine for a build.
|
|
If build_id is None, returns the most recently started engine."""
|
|
if build_id:
|
|
return self._engines.get(build_id)
|
|
# Return the most recent (last-inserted) engine
|
|
if not self._engines:
|
|
return None
|
|
return list(self._engines.values())[-1]
|
|
|
|
def active_build_ids(self) -> List[str]:
|
|
"""Return IDs of builds with engines currently running."""
|
|
return list(self._engines.keys())
|
|
|
|
# -------------------------------------------------
|
|
# PERSISTENCE
|
|
# -------------------------------------------------
|
|
def _persist(self, record: BuildRecord):
|
|
if self.storage:
|
|
try:
|
|
self.storage.save_build(record.to_dict())
|
|
except Exception:
|
|
pass # storage errors are non-fatal
|
|
|
|
def load_from_storage(self):
|
|
"""On startup, load any persisted builds into the in-memory cache."""
|
|
if not self.storage:
|
|
return
|
|
try:
|
|
builds = self.storage.list_builds()
|
|
for b in builds:
|
|
# Don't overwrite running builds (they're stale from a crashed process)
|
|
if b.get("status") == "running":
|
|
b["status"] = "crashed"
|
|
b["error"] = "process restarted while build was running"
|
|
rec = BuildRecord(**{k: v for k, v in b.items() if k in BuildRecord.__dataclass_fields__})
|
|
self._builds[rec.build_id] = rec
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class _EngineProxy:
|
|
"""Stand-in object that the debugger + pipeline_control modules can call
|
|
pause/resume/step on. Forwards calls to the BuildRunner's most-recent
|
|
active engine, so debugger buttons affect the currently-running build."""
|
|
|
|
def __init__(self, runner: BuildRunner):
|
|
self._runner = runner
|
|
|
|
def _engine(self) -> Optional[PipelineEngine]:
|
|
return self._runner.get_engine()
|
|
|
|
def pause(self):
|
|
eng = self._engine()
|
|
if eng:
|
|
eng.pause()
|
|
return True
|
|
return False
|
|
|
|
def resume(self):
|
|
eng = self._engine()
|
|
if eng:
|
|
eng.resume()
|
|
return True
|
|
return False
|
|
|
|
def step(self):
|
|
eng = self._engine()
|
|
if eng:
|
|
eng.step()
|
|
return True
|
|
return False
|
|
|
|
def step_back(self):
|
|
# Step-back is replay-only; live engine can't undo
|
|
return False
|
|
|
|
def is_active(self) -> bool:
|
|
return self._engine() is not None
|