mirror of
https://github.com/blw1138/cross-py-builder.git
synced 2026-09-07 21:41:09 -05:00
Brett Williams
8a6f02b7e9
The controller ran silently and surfaced errors as opaque/empty bodies, making it impossible to diagnose failures from either the UI or the server. - Configure Python logging (timestamps, levels) and log job lifecycle events (created / dispatched / done / failed) plus startup and dispatch warnings. - Add a catch-all error handler so unhandled exceptions return a JSON error with the exception type+message instead of an empty response. - Remove the frontend submit bug that hand-set multipart Content-Type without a boundary (and built an opts object it never used); post the FormData directly so the browser sets the correct content type + boundary.
239 lines
9.5 KiB
Python
239 lines
9.5 KiB
Python
import datetime
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
|
|
import requests
|
|
|
|
import ctrl.db as db
|
|
import ctrl.settings as settings
|
|
import ctrl.workers as wrk
|
|
|
|
log = logging.getLogger("ctrl.scheduler")
|
|
|
|
|
|
class Scheduler:
|
|
def __init__(self):
|
|
self._stop = threading.Event()
|
|
self._thread = None
|
|
self._active = 0
|
|
self._active_lock = threading.Lock()
|
|
self._log_lock = threading.Lock()
|
|
self._stream_subscribers = {} # job_id -> set of queue.Queue
|
|
|
|
# ---- SSE helpers -----------------------------------------------------
|
|
def subscribe(self, job_id):
|
|
import queue
|
|
q = queue.Queue()
|
|
self._stream_subscribers.setdefault(job_id, set()).add(q)
|
|
return q
|
|
|
|
def unsubscribe(self, job_id, q):
|
|
subs = self._stream_subscribers.get(job_id)
|
|
if subs:
|
|
subs.discard(q)
|
|
if not subs:
|
|
self._stream_subscribers.pop(job_id, None)
|
|
|
|
def _publish(self, job_id, event, data):
|
|
subs = self._stream_subscribers.get(job_id)
|
|
if subs:
|
|
frame = f"event: {event}\ndata: {data}\n\n"
|
|
for q in list(subs):
|
|
q.put_nowait(frame)
|
|
|
|
# ---- lifecycle -------------------------------------------------------
|
|
def start(self):
|
|
if self._thread and self._thread.is_alive():
|
|
return
|
|
settings.ensure_dirs()
|
|
db.init_db()
|
|
self._stop.clear()
|
|
self._thread = threading.Thread(target=self._loop, name="scheduler", daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
if self._thread:
|
|
self._thread.join(timeout=5)
|
|
|
|
# ---- dispatch loop ---------------------------------------------------
|
|
def _loop(self):
|
|
while not self._stop.is_set():
|
|
try:
|
|
self._tick()
|
|
except Exception as e:
|
|
print(f"[scheduler] tick error: {e}")
|
|
self._stop.wait(settings.SCHEDULER_INTERVAL)
|
|
|
|
def _tick(self):
|
|
with self._active_lock:
|
|
if self._active >= settings.MAX_CONCURRENT:
|
|
return
|
|
free = settings.MAX_CONCURRENT - self._active
|
|
jobs = [j for j in db.list_jobs(limit=200) if j["status"] == "queued"]
|
|
for job in jobs[:free]:
|
|
self._dispatch(job)
|
|
|
|
# ---- job dispatch ----------------------------------------------------
|
|
def _dispatch(self, job):
|
|
job_id = job["id"]
|
|
candidates = [w for w in wrk.probe_all() if wrk.matches(job, w)]
|
|
if not candidates:
|
|
log.warning("Job %s: no worker matched (os=%r cpu=%r)", job_id, job.get("os_req"), job.get("cpu_req"))
|
|
self._append(job_id, f"[scheduler] No worker matched requirements (os='{job['os_req']}', cpu='{job['cpu_req']}').\n")
|
|
return
|
|
|
|
worker = candidates[0]
|
|
with self._active_lock:
|
|
self._active += 1
|
|
try:
|
|
self._run_on_worker(job, worker)
|
|
finally:
|
|
with self._active_lock:
|
|
self._active -= 1
|
|
|
|
def _run_on_worker(self, job, worker):
|
|
job_id = job["id"]
|
|
db.update_job(job_id, status="dispatching", worker_host=worker["url"],
|
|
started_at=datetime.datetime.now().isoformat())
|
|
log.info("Job %s: dispatching to %s (%s %s)", job_id, worker["url"], worker.get("os"), worker.get("cpu"))
|
|
self._append(job_id, f"[scheduler] Dispatching to {worker['url']} ({worker.get('os')} {worker.get('cpu')})\n")
|
|
|
|
try:
|
|
if job["source_type"] == "git":
|
|
worker_job_id = self._request_git(job, worker)
|
|
else:
|
|
worker_job_id = self._request_upload(job, worker)
|
|
except Exception as e:
|
|
log.error("Job %s: dispatch failed: %s", job_id, e)
|
|
db.update_job(job_id, status="failed", finished_at=datetime.datetime.now().isoformat(),
|
|
error=str(e))
|
|
self._append(job_id, f"[scheduler] FAILED: {e}\n")
|
|
return
|
|
|
|
worker_url = worker["url"]
|
|
db.update_job(job_id, status="building", worker_job_id=worker_job_id, worker_url=worker_url)
|
|
self._append(job_id, f"[scheduler] Building on {worker_url} (worker job {worker_job_id})\n")
|
|
|
|
# Poll progress until the worker reports the build finished.
|
|
try:
|
|
self._wait_for_build(job, worker, worker_job_id)
|
|
self._fetch_artifacts(job, worker, worker_job_id)
|
|
db.update_job(job_id, status="done", finished_at=datetime.datetime.now().isoformat())
|
|
log.info("Job %s: done, artifacts saved", job_id)
|
|
self._append(job_id, "[scheduler] Build complete, artifacts saved.\n")
|
|
except Exception as e:
|
|
log.error("Job %s: failed: %s", job_id, e)
|
|
db.update_job(job_id, status="failed", finished_at=datetime.datetime.now().isoformat(),
|
|
error=str(e))
|
|
self._append(job_id, f"[scheduler] FAILED: {e}\n")
|
|
finally:
|
|
self._publish(job["id"], "job", '"done"')
|
|
|
|
def _request_git(self, job, worker):
|
|
job_id = job["id"]
|
|
self._append(job_id, f"[scheduler] Cloning {job['source']} on {worker['url']}\n")
|
|
resp = requests.post(f"{worker['url']}/checkout_git?async=1",
|
|
json={"repo_url": job["source"]},
|
|
timeout=settings.WORKER_STATUS_TIMEOUT,
|
|
stream=False)
|
|
data = self._worker_response(resp, "checkout")
|
|
if not data.get("id"):
|
|
raise RuntimeError("checkout did not return a job id")
|
|
return data["id"]
|
|
|
|
def _request_upload(self, job, worker):
|
|
job_id = job["id"]
|
|
path = self._source_path(job)
|
|
if not path or not os.path.exists(path):
|
|
raise FileNotFoundError(f"Source file not found: {path}")
|
|
self._append(job_id, f"[scheduler] Uploading {os.path.basename(path)} to {worker['url']}\n")
|
|
with open(path, "rb") as f:
|
|
resp = requests.post(f"{worker['url']}/upload?async=1",
|
|
files={"file": (os.path.basename(path), f, "application/zip")},
|
|
timeout=settings.WORKER_STATUS_TIMEOUT,
|
|
stream=False)
|
|
data = self._worker_response(resp, "upload")
|
|
if not data.get("id"):
|
|
raise RuntimeError("upload did not return a job id")
|
|
return data["id"]
|
|
|
|
def _wait_for_build(self, job, worker, worker_job_id):
|
|
"""Poll worker progress until the build completes (progress cleared + worker ready)."""
|
|
job_id = job["id"]
|
|
deadline = datetime.datetime.now() + datetime.timedelta(seconds=settings.WORKER_BUILD_TIMEOUT)
|
|
while datetime.datetime.now() < deadline:
|
|
progress = self._probe_progress(job, worker["url"], worker_job_id)
|
|
if progress is None:
|
|
status = wrk.probe_worker(worker["url"])
|
|
if status and status.get("status") == "ready":
|
|
return
|
|
# progress gone but worker not ready: brief window, keep waiting
|
|
time.sleep(settings.SCHEDULER_INTERVAL)
|
|
continue
|
|
line = progress.get("last_log_line") or progress.get("step") or ""
|
|
if line:
|
|
self._append(job_id, f"[worker] {line}\n")
|
|
time.sleep(settings.SCHEDULER_INTERVAL)
|
|
raise RuntimeError(f"Worker did not finish in {settings.WORKER_BUILD_TIMEOUT}s")
|
|
|
|
@staticmethod
|
|
def _probe_progress(job, worker_url, worker_job_id):
|
|
try:
|
|
resp = requests.get(f"{worker_url}/progress/{worker_job_id}",
|
|
timeout=settings.WORKER_STATUS_TIMEOUT)
|
|
except requests.RequestException:
|
|
return None
|
|
if resp.status_code == 404:
|
|
return None
|
|
try:
|
|
return resp.json()
|
|
except ValueError:
|
|
return None
|
|
|
|
@staticmethod
|
|
def _worker_response(resp, action):
|
|
try:
|
|
data = resp.json()
|
|
except ValueError:
|
|
raise RuntimeError(f"{action} returned non-JSON ({resp.status_code})")
|
|
if resp.status_code >= 400:
|
|
raise RuntimeError(f"{action} failed ({resp.status_code}): {data.get('error', data)}")
|
|
return data
|
|
|
|
def _source_path(self, job):
|
|
jobs_dir = os.path.join(settings.DATA_DIR, "sources")
|
|
name = job.get("source")
|
|
if not name:
|
|
return None
|
|
return os.path.join(jobs_dir, name)
|
|
|
|
def _fetch_artifacts(self, job, worker, worker_job_id):
|
|
job_id = job["id"]
|
|
self._append(job_id, f"[scheduler] Fetching artifacts from {worker['url']}/download/{worker_job_id}\n")
|
|
resp = requests.get(f"{worker['url']}/download/{worker_job_id}", timeout=settings.WORKER_BUILD_TIMEOUT)
|
|
resp.raise_for_status()
|
|
|
|
out_dir = os.path.join(settings.BUILDS_DIR, job_id)
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
base = os.path.splitext(os.path.basename(job.get("source") or job_id))[0]
|
|
fname = f"{base}-{str(worker.get('os')).lower()}-{str(worker.get('cpu')).lower()}.zip"
|
|
dest = os.path.join(out_dir, fname)
|
|
with open(dest, "wb") as f:
|
|
for chunk in resp.iter_content(chunk_size=8192):
|
|
f.write(chunk)
|
|
|
|
db.update_job(job_id, artifacts=[fname])
|
|
|
|
# ---- logging / status -------------------------------------------------
|
|
def _append(self, job_id, text):
|
|
with self._log_lock:
|
|
db.append_log(job_id, text)
|
|
self._publish(job_id, "log", '"%s"' % text.rstrip().replace('"', '\\"'))
|
|
|
|
|
|
def make_scheduler():
|
|
return Scheduler()
|