From 0e89919342ac5c50d12d473175e0faa8e78bdde5 Mon Sep 17 00:00:00 2001 From: Brett Williams Date: Sun, 30 Aug 2026 21:55:46 -0500 Subject: [PATCH] Add controller with web UI and retire CLI - Add ctrl/ package: Flask app (waitress), SQLite job store, background scheduler that probes configured workers, dispatches via /upload and /checkout_git, fetches artifacts, and serves a single-page UI - Supporting API: /api/jobs (create/list/detail/cancel), /api/workers, /api/capabilities, artifact download, and SSE log stream - Workers keep the existing HTTP API; add /progress/ endpoint and per-step build tracking for live status - Retire agent_manager.py and the cross-py-builder CLI; controller UI is the primary frontend - Record design in DESIGN.md; add waitress dependency Note: /progress is exposed on the worker but the synchronous /upload and /checkout endpoints block until a build completes, so per-step worker progress is not yet relayed live in the UI. Live streaming needs an async-start worker endpoint as a follow-up. --- DESIGN.md | 201 ++++++++++++++++++ cross_py_builder/agent_manager.py | 339 ------------------------------ cross_py_builder/build_agent.py | 59 +++++- ctrl/__init__.py | 0 ctrl/app.py | 158 ++++++++++++++ ctrl/db.py | 112 ++++++++++ ctrl/scheduler.py | 191 +++++++++++++++++ ctrl/settings.py | 56 +++++ ctrl/static/app.js | 197 +++++++++++++++++ ctrl/static/index.html | 68 ++++++ ctrl/static/style.css | 143 +++++++++++++ ctrl/workers.py | 50 +++++ requirements.txt | 3 +- setup.py | 2 +- 14 files changed, 1237 insertions(+), 342 deletions(-) create mode 100644 DESIGN.md delete mode 100755 cross_py_builder/agent_manager.py create mode 100644 ctrl/__init__.py create mode 100644 ctrl/app.py create mode 100644 ctrl/db.py create mode 100644 ctrl/scheduler.py create mode 100644 ctrl/settings.py create mode 100644 ctrl/static/app.js create mode 100644 ctrl/static/index.html create mode 100644 ctrl/static/style.css create mode 100644 ctrl/workers.py diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..c037f29 --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,201 @@ +# Cross-Py-Builder — Controller + Web UI Design + +Target architecture: keep the existing worker (`build_agent.py`) HTTP API as-is, and add a +**controller** LXC that runs a web app. The controller stores every submitted request as a +**job row**, polls workers to pick a target, dispatches the build, records download artifacts, +and exposes a browser UI to submit, track progress, and download results. + +The workers (`build_agent.py`) keep their HTTP API; only one small read-only endpoint +(`/progress/`) is added for live build progress. Everything else is new controller code. + +--- + +## 1. Components + +``` + Browser + │ (HTTP, port 8080) + ▼ + ┌───────────────────────────────── ─┐ ┌──────────────────────┐ + │ CONTROLLER LXC │ │ WORKER A (ubuntu x64)│ + │ │ │ │ + │ Flask app (waitress) │ │ build_agent.py │ + │ ├─ web UI (submit/track/download)│ │ /upload /status │ + │ ├─ job API │ ───► │ /download /checkout │ + │ ├─ scheduler (string executor) │ └──────────────────────┘ + │ └─ SQLite (jobs.db) │ ┌──────────────────────┐ + │ │ │ WORKER B (win arm64)│ + └───────────────────────────────── ─┘ │ build_agent.py │ + └──────────────────────┘ + ctrl/ controller package + ctrl/static/ frontend (HTML/JS/CSS) + ctrl/jobs.db SQLite database (runtime) + builds/ downloaded artifacts (runtime) +``` + +**Worker discovery:** static config (list of `host:port`) in the controller settings. On each +job the controller queries every worker `/status`, filters to `status=="ready"` and matching +CPU/OS, and picks one. This replaces Zeroconf on the controller; workers can stop advertising, +and the Zeroconf code path can be removed with the CLI retirement (§7). + +We deliberately do **not** rewrite the agent's HTTP API, and do **not** add auth in v1 (same +trusted-LAN posture as today). The worker endpoint contract used by the controller: + +| Worker endpoint | Method | Purpose | +|---|---|---| +| `/status` | GET | readiness, os, cpu, agent_version, hostname, ip | +| `/upload` | POST | multipart `file` zip → starts build, returns job JSON | +| `/checkout_git` | POST | JSON `{"repo_url": ...}` → starts build | +| `/download/` | GET | zip of `dist/` for a built job | +| `/delete_cache` | GET | clear cached jobs | +| `/healthz` | GET | liveness (already added) | +| `/progress/` | GET | **NEW (small agent change):** live build snapshot (status, current step, elapsed, last log line) for SSE relay | + +--- + +## 2. Data model + +**SQLite table `jobs`** (one row per submitted build; single table keeps v1 simple): + +| column | type | notes | +|---|---|---| +| `id` | text PK | controller job id (e.g. `JOB-<8 hex>`) — distinct from worker job id | +| `status` | text | `queued → dispatching → building → done \| failed \| cancelled` | +| `source_type` | text | `upload` \| `git` | +| `source` | text | uploaded filename or repo URL | +| `os_req` | text? | requested OS filter (nullable) | +| `cpu_req` | text? | requested CPU filter (nullable) | +| `worker_host` | text | worker chosen (host:port) | +| `worker_job_id` | text? | job id returned by the worker | +| `worker_url` | text? | base URL of worker (for download link) | +| `created_at` | text | ISO timestamp | +| `started_at` | text? | | +| `finished_at` | text? | | +| `log` | text | aggregated progress lines (appended, newline-delimited) | +| `error` | text? | last error detail | +| `artifacts` | text | JSON list of downloaded artifact paths/names | + +Artifacts are downloaded by the controller into `builds//` (`---.zip`) +so results survive worker container restarts and remain available after the worker cleans up. + +--- + +## 3. REST API (controller) + +| Method | Path | Purpose | +|---|---|---| +| GET | `/` | web UI (single page) | +| GET | `/api/workers` | live `/status` for each configured worker | +| GET | `/api/capabilities` | distinct (os, cpu) across ready workers, for dropdowns | +| POST | `/api/jobs` | create job: multipart upload `file` OR JSON `{repo_url, os, cpu}` | +| GET | `/api/jobs` | list jobs (newest first, with status) | +| GET | `/api/jobs/` | job detail incl. live `log` | +| POST | `/api/jobs//cancel` | set status → `cancelled` (best-effort) | +| GET | `/api/jobs//artifacts/` | download a built artifact | +| GET | `/api/jobs//stream` | SSE log stream for live tail | + +--- + +## 4. Scheduler (dispatch loop) + +A single background thread owns all dispatch work. + +1. Pop the oldest job with `status == "queued"` (FIFO). +2. From worker config, query `/status` (short timeout). Filter: + - `status` contains `ready` (not building/updating) + - if `os_req` set, worker `os` contains `os_req` (case-insensitive) + - if `cpu_req` set, worker `cpu` contains `cpu_req` + - exclude the local/controller LXC if present, to avoid self-builds +3. Choose first match (optionally: prefer workers with more free disk — v1: first match). +4. Mark job `dispatching`; set `worker_host`. +5. Send to worker: + - upload → `POST /upload` with the stored file + - git → `POST /checkout_git` with `{repo_url}` + - on success, read the worker job JSON → store `worker_job_id`, `worker_url`, set `building` + - on hard failure, mark `failed` (§ failure handling below) +6. **Progress relay** — while the dispatch call is in flight (the worker is building), poll + `GET /progress/` on an interval and append each snapshot to the job log + + push it to connected SSE clients. This gives live per-step progress in the UI, enabled by + the small agent addition (§1). +7. On success (dispatch call returns): compute download URL `worker_url + /download/`, + fetch the zip, save to `builds//`, record artifact in the job row, set `done`. + +**Concurrency/limits:** scheduler processes one job at a time (workers can already build only one +at a time — `409`). Multiple queued jobs simply wait. This makes NIC/disk behavior predictable +and avoids the current CLI dumping N builds in parallel. + +**Failure handling:** if dispatch fails (worker down/409/500), mark job `failed` and record +`error` + log tail. No auto-retry in v1 (confirmed decision) — surface the error and let the +user resubmit to an eligible worker. + +--- + +## 5. Frontend (single-page, no build-step) + +- Ask for `/api/capabilities` → render OS + CPU dropdowns (or "any"). +- Upload a zip **or** paste a git URL; submit. +- Job list refreshes via periodic `/api/jobs`. +- Detail view: status badge + log (poll, or SSE stream). Download button appears when `done`. +- No npm/JS toolchain — vanilla JS + ``ed CSS; keeps the controller LXC dependency-light. + +--- + +## 6. Project layout (new/changed files) + +``` +setup.py (add ctrl package + entry point `cross-py-controller`) +requirements.txt (add waitress; keep Flask/requests) +cross_py_builder/ (agent unchanged except new /progress endpoint) +ctrl/ + __init__.py + app.py (Flask factory, routes, waitress runner) + db.py (SQLite init + helpers) + scheduler.py (dispatch + progress-poll loop, worker client) + workers.py (worker config load + /status probe) + settings.py (env-driven config: DB path, builds dir, worker list) + static/ + index.html + app.js + style.css +``` + +Worker config example (env or JSON file): +``` +CROSS_PY_WORKERS='["10.0.0.11:9001","10.0.0.12:9001"]' +CROSS_PY_DATA=/var/lib/cross-py-controller # holds jobs.db + builds/ +CROSS_PY_PORT=8080 +``` + +--- + +## 7. What we reuse vs. retire + +- **Reuse on workers:** agent HTTP API, plus the new `/progress/` endpoint. +- **Retire the CLI entirely (confirmed decision):** `agent_manager.py` is removed. The + controller web UI is the sole frontend. This drops the Zeroconf-dependent manager code paths + (multi-parallel submits, hardcoded `DEFAULT_PORT`, the self-update flow) and leaves a single + interface to maintain. + +## 8. Build environment model (per-build venv) + +Workers keep creating a fresh venv and installing requirements per job. This is the isolated +model — no cross-job env contamination, and a bad or malicious `requirements.txt` can't poison +a shared environment. Tradeoff: each job pays ~2-5 min of pip installs (PyInstaller + deps) +and repeats downloads/disk churn. + +Mitigation ladder, if wall-clock ever hurts (preserve isolation in all cases): +1. Persistent pip-cache volume on the worker (`~/.cache/pip`) — fast, no isolation loss. +2. Local wheel mirror / `--find-links` — same, plus offline-friendly. +3. Prebuilt PyInstaller baked into the worker LXC snapshot (skip its reinstall per job). + +Do **not** move to a shared persistent venv until a concrete cross-job dependency problem +appears. + +## 9. Deferred / next-phase (explicitly out of scope for v1) + +- Auth / tokens on controller and worker endpoints +- Worker add/remove/update/restart/shutdown via the UI +- Auto-retry on a different worker; retry-with-backoff for queued jobs +- Persisting/streaming full-build logs from worker (`build-.log` is on the worker today) +- Storing source zip centrally for re-runs +- Auto-scaling/provisioning of worker LXCs diff --git a/cross_py_builder/agent_manager.py b/cross_py_builder/agent_manager.py deleted file mode 100755 index 2ba1f3f..0000000 --- a/cross_py_builder/agent_manager.py +++ /dev/null @@ -1,339 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import concurrent.futures -import os -import socket -import tempfile -import time -import zipfile -from datetime import datetime, timedelta - -import requests -from packaging.version import Version -from tabulate import tabulate - -from .build_agent import build_agent_version -from .zeroconf_server import ZeroconfServer - -DEFAULT_PORT = 9001 -TIMEOUT = 10 - -def find_server_ips(): - ZeroconfServer.configure("_crosspybuilder._tcp.local.", socket.gethostname(), DEFAULT_PORT) - hostnames = [] - try: - ZeroconfServer.start(listen_only=True) - time.sleep(.3) # give it time to find network - hostnames = ZeroconfServer.found_ip_addresses() - except KeyboardInterrupt: - pass - finally: - ZeroconfServer.stop() - - # get known hosts - # with open("../known_hosts", "r") as file: - # lines = file.readlines() - # hostnames.extend(lines) - - return hostnames - - -def get_all_servers_status(): - table_data = [] - server_ips = find_server_ips() - - with concurrent.futures.ThreadPoolExecutor() as executor: - futures = [] - for server in server_ips: - ip = server.split(':')[0] - port = server.split(":")[-1].strip() if ":" in server else DEFAULT_PORT - futures.append(executor.submit(get_worker_status, ip, port)) - - for future in concurrent.futures.as_completed(futures): - try: - result = future.result() # Get the result of the thread - table_data.append(result) - except Exception as e: - print(f"Error fetching status from server: {e}") # Handle potential errors - return table_data - - -def get_worker_status(hostname, port=DEFAULT_PORT): - """Fetch worker status from the given hostname.""" - try: - response = requests.get(f"http://{hostname}:{port}/status", timeout=TIMEOUT) - status = response.json() - status['port'] = port - if status['hostname'] != hostname and status['ip'] != hostname: - status['ip'] = socket.gethostbyname(hostname) - return status - except requests.exceptions.RequestException as e: - return {"hostname": hostname, "port": port, "status": "offline"} - - -def zip_project(source_dir, output_zip): - """Zips the given directory.""" - excluded_dirs = {"venv", ".venv", "dist", "build"} - with zipfile.ZipFile(output_zip, 'w', zipfile.ZIP_DEFLATED) as zipf: - for root, dirs, files in os.walk(source_dir): - - dirs[:] = [d for d in dirs if d.lower() not in excluded_dirs] - for file in files: - file_path = os.path.join(root, file) - zipf.write(file_path, os.path.relpath(file_path, source_dir)) - - -def send_build_request(server_ip, server_port=DEFAULT_PORT, zip_file=None, git_url=None, download_after=True, version=None): - - if zip_file: - upload_url = f"http://{server_ip}:{server_port}/upload" - print(f"Submitting build request to URL: {upload_url} - Please wait. This may take a few minutes...") - with open(zip_file, 'rb') as f: - response = requests.post(upload_url, files={"file": f}) - elif git_url: - checkout_url = f"http://{server_ip}:{server_port}/checkout_git" - response = requests.post(checkout_url, json={"repo_url": git_url}) - else: - raise ValueError("Missing zip file or git url!") - - if response.status_code == 200: - response_data = response.json() - print(response_data) - print(f"Build successful. ID: {response_data['id']} Hostname: {response_data['hostname']} OS: {response_data['os']} CPU: {response_data['cpu']} Spec files: {len(response_data['spec_files'])} - Elapsed time: {response_data['duration']}" ) - if download_after: - download_url = f"http://{server_ip}:{server_port}/download/{response_data.get('id')}" - try: - base_name = os.path.splitext(os.path.basename(zip_file))[0] - version_string = ("-" + version) if version else "" - save_name = f"{base_name}{version_string}-{response_data['os'].lower()}-{response_data['cpu'].lower()}.zip" - download_zip(download_url, save_name=save_name) - except Exception as e: - print(f"Error downloading zip: {e}") - else: - print("Upload failed:", response.status_code, response.text) - - -def download_zip(url, save_name, save_dir="."): - """Download a ZIP file from a URL and save it with its original filename.""" - response = requests.get(url, stream=True) - response.raise_for_status() # Raise an error if request fails - - save_path = os.path.join(save_dir, save_name) - with open(save_path, "wb") as file: - for chunk in response.iter_content(chunk_size=8192): # Download in chunks - file.write(chunk) - - print(f"Saved binaries to file: {save_path}") - return save_path - - -def select_server(servers, cpu=None, os_name=None): - """Selects a build server based on CPU architecture and OS filters.""" - available = [s for s in servers if s["status"] == "ready"] - - if cpu: - available = [s for s in available if cpu.lower() in s["cpu"].lower()] - if os_name: - available = [s for s in available if os_name.lower() in s["os"].lower()] - return available[0] if available else None # Return first matching server or None - - -def process_new_job(args, server_data): - available_servers = [s for s in server_data if s["status"] == "ready"] - - if args.cpu: - available_servers = [s for s in available_servers if args.cpu.lower() in s["cpu"].lower()] - if args.os: - available_servers = [s for s in available_servers if args.os.lower() in s["os"].lower()] - - if not available_servers: - print("No available servers matching the criteria.") - return - - # Keep only unique servers - unique_servers = {} - for server in available_servers: - key = (server["cpu"], server["os"]) - if key not in unique_servers: - unique_servers[key] = server - - available_servers = list(unique_servers.values()) - print(f"Found {len(available_servers)} servers available to build") - print(tabulate(available_servers, headers="keys", tablefmt="grid")) - - zip_file = None - if args.build: - project_path = args.build - tmp_dir = tempfile.gettempdir() - zip_file = os.path.join(tmp_dir, f"{os.path.basename(project_path)}.zip") - zip_project(project_path, zip_file) - print(f"Zipped {project_path} to {zip_file}") - - # Start builds on all matching servers - with concurrent.futures.ThreadPoolExecutor() as executor: - print("Submitting builds to:") - download = True - for server in available_servers: - print(f"\t{server['hostname']} - {server['os']} - {server['cpu']}") - futures = {executor.submit( - send_build_request,server["ip"], DEFAULT_PORT, zip_file, args.checkout, download, args.version): - server for server in available_servers} - - # Collect results - for future in concurrent.futures.as_completed(futures): - server = futures[future] - try: - response = future.result() # Get the response - except Exception as e: - print(f"Build failed on {server['hostname']}: {e}") - - try: - if zip_file: - os.remove(zip_file) - except Exception as e: - print(f"Error removing zip file: {e}") - - -def delete_cache(server_data): - available_servers = [s for s in server_data if s["status"] == "ready"] - print(f"Deleting cache in from all available servers ({len(available_servers)})") - for server in available_servers: - try: - response = requests.get(f"http://{server['ip']}:{server.get('port', DEFAULT_PORT)}/delete_cache") - response.raise_for_status() - print(f"Cache cleared on {server['hostname']}") - except Exception as e: - print(f"Error deleting cache on {server['hostname']}: {e}") - - -def update_worker(server): - try: - print(f"Updating {server['hostname']} from {server.get('agent_version')} => {build_agent_version}") - - with open("build_agent.py", "rb") as file1, open("../requirements.txt", "rb") as file2: - update_files = { - "file1": open("build_agent.py", "rb"), - "file2": open("../requirements.txt", "rb") - } - response = requests.post(f"http://{server['ip']}:{server.get('port', DEFAULT_PORT)}/update", - files=update_files) - response.raise_for_status() - - response_json = response.json() - if response_json.get('updated_files') and not response_json.get('error_files'): - try: - requests.get(f"http://{server['ip']}:{server.get('port', DEFAULT_PORT)}/restart", timeout=TIMEOUT) - except requests.exceptions.ConnectionError: - pass - return server - else: - print(f"Error updating {server['hostname']}. Errors: {response_json.get('error_files')} - Updated: {response_json.get('updated_files')}") - except Exception as e: - print(f"Unhandled error updating {server['hostname']}: {e}") - return None - -def update_build_workers(server_data): - available_workers = [s for s in server_data if s["status"] == "ready"] - workers_to_update = [x for x in available_workers if Version(x.get('agent_version')) < Version(build_agent_version)] - - if not workers_to_update: - print(f"All {len(available_workers)} workers up to date") - return - - print(f"Updating workers on all available servers ({len(workers_to_update)}) to {build_agent_version}") - updated_servers = [] - with concurrent.futures.ThreadPoolExecutor() as executor: - futures = {executor.submit(update_worker, server): server for server in workers_to_update} - for future in concurrent.futures.as_completed(futures): - server = future.result() - if server: - updated_servers.append(server) - - if updated_servers: - print("Waiting for servers to restart...") - unverified_servers = {server["ip"]: server for server in updated_servers} - end_time = datetime.now() + timedelta(seconds=30) - while unverified_servers and datetime.now() < end_time: - for server_ip in list(unverified_servers.keys()): # Iterate over a copy to avoid modification issues - try: - response = requests.get(f"http://{server_ip}:{server.get('port', DEFAULT_PORT)}/status") - response.raise_for_status() - server_info = unverified_servers[server_ip] # Get full server details - agent_version = response.json().get('agent_version') - if agent_version == build_agent_version: - print(f"Agent on {server_info['hostname']} successfully upgraded to {build_agent_version}") - else: - print(f"Agent on {server_info['hostname']} failed to upgrade. Still on version {agent_version}") - unverified_servers.pop(server_ip) - except requests.exceptions.ConnectionError: - pass # Server is still restarting - if unverified_servers: - time.sleep(1) # Short delay before retrying to avoid spamming requests - if unverified_servers: - print("Some servers did not restart in time:", list(unverified_servers.keys())) - - print("Update complete") - - -def shutdown_agent(hostname): - print(f"Shutting down hostname: {hostname}") - try: - requests.get(f"http://{hostname}:{DEFAULT_PORT}/shutdown", timeout=TIMEOUT) - except (requests.exceptions.ConnectionError, TimeoutError): - pass - - -def restart_agent(hostname): - print(f"Restarting agent: {hostname}") - try: - requests.get(f"http://{hostname}:{DEFAULT_PORT}/restart", timeout=TIMEOUT) - except (requests.exceptions.ConnectionError, TimeoutError): - pass - - -def main(): - parser = argparse.ArgumentParser(description="Build agent manager for cross_py_builder") - parser.add_argument("--status", action="store_true", help="Get status of available servers") - parser.add_argument("--build", type=str, help="Path to the project to build") - parser.add_argument("--checkout", type=str, help="Url to Git repo for checkout") - parser.add_argument("-cpu", type=str, help="CPU architecture") - parser.add_argument("-os", type=str, help="Operating system") - parser.add_argument("-version", type=str, help="Version number for build") - parser.add_argument("--delete-cache", action="store_true", help="Delete cache") - parser.add_argument("--update-all", action="store_true", help="Update build agent") - parser.add_argument("--restart", type=str, help="Hostname to restart") - parser.add_argument("--restart-all", action="store_true", help="Restart all agents") - parser.add_argument("--shutdown", type=str, help="Hostname to shutdown") - parser.add_argument("--shutdown-all", action="store_true", help="Shutdown all agents") - args = parser.parse_args() - - if args.status: - server_data = get_all_servers_status() - server_data = [x for x in server_data if x['status'] != 'offline'] - print(tabulate(server_data, headers="keys", tablefmt="grid")) - return - elif args.restart: - restart_agent(args.restart) - elif args.restart_all: - print("Restarting all agents...") - for server_ip in find_server_ips(): - restart_agent(server_ip) - elif args.shutdown: - shutdown_agent(args.shutdown) - elif args.shutdown_all: - print("Shutting down all agents...") - for server_ip in find_server_ips(): - shutdown_agent(server_ip) - elif args.delete_cache: - delete_cache(get_all_servers_status()) - elif args.build or args.checkout: - process_new_job(args, get_all_servers_status()) - elif args.update_all: - update_build_workers(get_all_servers_status()) - else: - print("No path given!") - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/cross_py_builder/build_agent.py b/cross_py_builder/build_agent.py index d156f39..4155f8f 100755 --- a/cross_py_builder/build_agent.py +++ b/cross_py_builder/build_agent.py @@ -32,6 +32,20 @@ BUILD_DIR = "pybuild-data" TMP_DIR = tempfile.gettempdir() system_status = {"status": "ready", "running_job": None} build_lock = threading.Lock() +progress_lock = threading.Lock() +build_progress = {} + +def _set_step(job_id, step, **extra): + with progress_lock: + build_progress[job_id] = dict(build_progress.get(job_id, {}), + job_id=job_id, + step=step, + step_started_at=datetime.datetime.now().isoformat(), + **extra) + +def _clear_progress(job_id): + with progress_lock: + build_progress.pop(job_id, None) def is_windows(): return platform.system().lower() == "windows" @@ -167,6 +181,28 @@ def status(): }) +@app.route('/progress/', methods=['GET']) +def progress(job_id): + if not is_valid_job_id(job_id): + return jsonify({"error": f"Invalid job id: {job_id}"}), 404 + + with progress_lock: + info = dict(build_progress.get(job_id)) if job_id in build_progress else None + + if not info: + return jsonify({"status": "not_found", "job_id": job_id}), 404 + + info['elapsed'] = "" + step_started = info.get('step_started_at') + if step_started: + try: + started = datetime.datetime.fromisoformat(step_started) + info['elapsed'] = str(datetime.datetime.now() - started) + except ValueError: + pass + return jsonify(info) + + def generate_job_id(): return str(uuid.uuid4()).split('-')[-1] @@ -193,6 +229,7 @@ def checkout_project(): start_time = datetime.datetime.now() try: os.makedirs(build_root, exist_ok=True) + _set_step(job_id, "cloning_repo", repo_url=repo_url) system_status['status'] = "cloning_repo" subprocess.check_call(['git', 'clone', repo_url, repo_dir]) system_status['status'] = "ready" @@ -201,6 +238,7 @@ def checkout_project(): print(f"Error processing checkout: {e}") system_status['status'] = "ready" system_status['running_job'] = None + _clear_progress(job_id) shutil.rmtree(repo_dir, ignore_errors=True) if isinstance(e, subprocess.CalledProcessError): return jsonify({'error': 'Failed to clone repository'}), 500 @@ -233,6 +271,7 @@ def upload_project(): working_dir = os.path.join(TMP_DIR, BUILD_DIR, job_id) file = request.files['file'] + _set_step(job_id, "processing_files", source=file.filename) zip_path = os.path.join(working_dir, "source.zip") # Save ZIP file @@ -248,25 +287,29 @@ def upload_project(): except Exception as e: print(f"Uncaught error processing job: {e}") system_status['status'] = "ready" + _clear_progress(job_id) shutil.rmtree(working_dir, ignore_errors=True) return jsonify({"error": f"Uncaught error processing job: {e}"}), 500 finally: build_lock.release() def install_and_build(project_path, job_id, start_time): + _set_step(job_id, "starting", project_path=project_path) # Find the PyInstaller spec file spec_files = glob.glob(os.path.join(project_path, "*.spec")) if not spec_files: + _clear_progress(job_id) return jsonify({"error": "No .spec files found"}), 400 print(f"Starting new build job - {len(spec_files)} spec files found") system_status['status'] = "working" - system_status['running_job'] = os.path.basename(project_path) + system_status['running_job'] = job_id # Set up virtual environment venv_path = os.path.join(project_path, "venv") try: + _set_step(job_id, "creating_venv") system_status['status'] = "creating_venv" print(f"\n========== Configuring Virtual Environment ({venv_path}) ==========") python_exec = "python" if is_windows() else "python3" @@ -279,11 +322,13 @@ def install_and_build(project_path, job_id, start_time): print(f"Error setting up virtual environment: {e}") system_status['status'] = "ready" system_status['running_job'] = None + _clear_progress(job_id) shutil.rmtree(project_path, ignore_errors=True) return jsonify({"error": f"Error setting up virtual environment: {e}"}), 500 # Install requirements try: + _set_step(job_id, "installing_packages") system_status['status'] = "installing_packages" subprocess.run([py_exec, "-m", "pip", "install", "--upgrade", "pip"], check=True) subprocess.run([py_exec, "-m", "pip", "install", "pyinstaller", "pyinstaller_versionfile", "--prefer-binary"], check=True) @@ -296,13 +341,17 @@ def install_and_build(project_path, job_id, start_time): print(f"Error installing requirements: {e}") system_status['status'] = "ready" system_status['running_job'] = None + _clear_progress(job_id) shutil.rmtree(project_path, ignore_errors=True) return jsonify({"error": f"Error installing requirements: {e}"}), 500 results = {} try: + spec_total = len(spec_files) for index, spec_file in enumerate(spec_files): # Compile with PyInstaller + _set_step(job_id, "compiling", spec_index=index, spec_total=spec_total, + spec_name=os.path.basename(spec_file)) system_status['status'] = "compiling" print(f"\n========== Compiling spec file {index+1} of {len(spec_files)} - {spec_file} ==========") simple_name = os.path.splitext(os.path.basename(spec_file))[0] @@ -322,22 +371,30 @@ def install_and_build(project_path, job_id, start_time): print(line, end="") log_file.write(line) log_file.flush() + with progress_lock: + if job_id in build_progress: + build_progress[job_id]['last_log_line'] = line.rstrip() process.wait() if process.returncode != 0: raise RuntimeError( f"PyInstaller failed with exit code {process.returncode}. Last line: {str(last_line).strip()}") print(f"\n========== Compilation of spec file {spec_file} complete ==========\n") + with progress_lock: + if job_id in build_progress: + build_progress[job_id]['spec_completed'] = index + 1 except Exception as e: print(f"Error compiling project: {e}") system_status['status'] = "ready" system_status['running_job'] = None + _clear_progress(job_id) shutil.rmtree(project_path, ignore_errors=True) return jsonify({"error": f"Error compiling project: {e}"}), 500 dist_path = os.path.join(project_path, "dist") system_status['status'] = "ready" system_status['running_job'] = None + _clear_progress(job_id) return jsonify({ "id": job_id, "message": "Build completed", diff --git a/ctrl/__init__.py b/ctrl/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ctrl/app.py b/ctrl/app.py new file mode 100644 index 0000000..b7e59be --- /dev/null +++ b/ctrl/app.py @@ -0,0 +1,158 @@ +import datetime +import json +import os +import queue + +from flask import Flask, jsonify, request, send_from_directory, Response + +import ctrl.db as db +import ctrl.settings as settings +import ctrl.workers as wrk +from ctrl.scheduler import make_scheduler + +app = Flask(__name__, static_folder="static", static_url_path="") +scheduler = make_scheduler() + +VIEW_STATIC = os.path.join(os.path.dirname(__file__), "static") + +_initialized = False + + +def _ensure_ready(): + global _initialized + if _initialized: + return + settings.ensure_dirs() + db.init_db() + _initialized = True + + +@app.before_request +def _before_request(): + _ensure_ready() + + +@app.route("/") +def index(): + return send_from_directory(VIEW_STATIC, "index.html") + + +@app.get("/api/workers") +def api_workers(): + return jsonify(wrk.probe_all()) + + +@app.get("/api/capabilities") +def api_capabilities(): + combos = { + (w.get("os"), w.get("cpu")) + for w in wrk.probe_all() + if w.get("os") and w.get("cpu") + } + return jsonify(sorted({"os": o, "cpu": c} for o, c in combos if o and c)) + + +@app.post("/api/jobs") +def api_create_job(): + os_req = (request.form.get("os") or "").strip() or None + cpu_req = (request.form.get("cpu") or "").strip() or None + + if request.files and "file" in request.files: + upload = request.files["file"] + if not upload.filename: + return jsonify({"error": "No file selected"}), 400 + source_dir = os.path.join(settings.DATA_DIR, "sources") + os.makedirs(source_dir, exist_ok=True) + fname = os.path.basename(upload.filename) + dest = os.path.join(source_dir, fname) + upload.save(dest) + job_id = db.create_job("upload", fname, os_req, cpu_req) + else: + data = request.get_json(silent=True) or {} + repo_url = (data.get("repo_url") or "").strip() + if not repo_url: + return jsonify({"error": "Provide a file upload or a repo_url"}), 400 + job_id = db.create_job("git", repo_url, os_req, cpu_req) + + scheduler.start() + return jsonify({"id": job_id}), 201 + + +@app.get("/api/jobs") +def api_list_jobs(): + return jsonify(db.list_jobs()) + + +@app.get("/api/jobs/") +def api_get_job(job_id): + job = db.get_job(job_id) + if not job: + return jsonify({"error": "not found"}), 404 + return jsonify(job) + + +@app.get("/api/jobs//artifacts/") +def api_download(job_id, name): + job = db.get_job(job_id) + if not job: + return jsonify({"error": "not found"}), 404 + safe = os.path.basename(name) + out_dir = os.path.join(settings.BUILDS_DIR, job_id) + return send_from_directory(out_dir, safe, as_attachment=True) + + +@app.post("/api/jobs//cancel") +def api_cancel(job_id): + job = db.get_job(job_id) + if not job: + return jsonify({"error": "not found"}), 404 + if job["status"] in ("queued",): + db.update_job(job_id, status="cancelled", finished_at=datetime.datetime.now().isoformat()) + return jsonify({"id": job_id, "status": "cancelled"}) + return jsonify({"error": f"Cannot cancel job in state {job['status']}"}), 400 + + +@app.get("/api/jobs//stream") +def api_stream(job_id): + job = db.get_job(job_id) + if not job: + return jsonify({"error": "not found"}), 404 + + q = scheduler.subscribe(job_id) + + def gen(): + # initial snapshot + snap = db.get_job(job_id) + if snap: + yield f"event: job\ndata: {json.dumps({'status': snap['status']})}\n\n" + for line in (snap.get("log") or "").splitlines(): + yield f"event: log\ndata: {json.dumps(line)}\n\n" + try: + while True: + try: + frame = q.get(timeout=15) + yield frame + except queue.Empty: + # heartbeat to keep connection alive + yield ": keepalive\n\n" + job_now = db.get_job(job_id) + if job_now and job_now["status"] in ("done", "failed", "cancelled"): + break + finally: + scheduler.unsubscribe(job_id, q) + + return Response(gen(), mimetype="text/event-stream") + + +def main(): + _ensure_ready() + scheduler.start() + if settings.DEBUG: + app.run(host="0.0.0.0", port=settings.PORT, threaded=True) + return + from waitress import serve + serve(app, host="0.0.0.0", port=settings.PORT, threads=8) + + +if __name__ == "__main__": + main() diff --git a/ctrl/db.py b/ctrl/db.py new file mode 100644 index 0000000..f7f53b3 --- /dev/null +++ b/ctrl/db.py @@ -0,0 +1,112 @@ +import json +import sqlite3 +import threading + +import ctrl.settings as settings + + +_local = threading.local() + + +def _conn(): + conn = getattr(_local, "conn", None) + if conn is None: + conn = sqlite3.connect(settings.DB_PATH, timeout=30) + conn.row_factory = sqlite3.Row + _local.conn = conn + return conn + + +def init_db(): + settings.ensure_dirs() + conn = _conn() + conn.execute( + """ + CREATE TABLE IF NOT EXISTS jobs ( + id TEXT PRIMARY KEY, + status TEXT NOT NULL, + source_type TEXT NOT NULL, + source TEXT, + os_req TEXT, + cpu_req TEXT, + worker_host TEXT, + worker_job_id TEXT, + worker_url TEXT, + created_at TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + log TEXT DEFAULT '', + error TEXT, + artifacts TEXT DEFAULT '[]' + ) + """ + ) + conn.commit() + + +def create_job(source_type, source, os_req, cpu_req): + import uuid + import datetime + job_id = "JOB-" + uuid.uuid4().hex[:8] + conn = _conn() + conn.execute( + """ + INSERT INTO jobs (id, status, source_type, source, os_req, cpu_req, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + (job_id, "queued", source_type, source, os_req, cpu_req, + datetime.datetime.now().isoformat()), + ) + conn.commit() + return job_id + + +def _row_to_dict(row): + d = dict(row) + try: + d["artifacts"] = json.loads(d.get("artifacts") or "[]") + except (TypeError, ValueError): + d["artifacts"] = [] + return d + + +def get_job(job_id): + row = _conn().execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() + return _row_to_dict(row) if row else None + + +def list_jobs(limit=100): + rows = _conn().execute( + "SELECT * FROM jobs ORDER BY created_at DESC LIMIT ?", (limit,) + ).fetchall() + return [_row_to_dict(r) for r in rows] + + +def update_job(job_id, **fields): + if not fields: + return + cols = ", ".join(f"{k} = ?" for k in fields) + values = [] + for v in fields.values(): + if isinstance(v, (list, dict)): + v = json.dumps(v) + values.append(v) + conn = _conn() + conn.execute(f"UPDATE jobs SET {cols} WHERE id = ?", (*values, job_id)) + conn.commit() + + +def append_log(job_id, text): + conn = _conn() + conn.execute("UPDATE jobs SET log = log || ? WHERE id = ?", (text, job_id)) + conn.commit() + + +def set_queued_jobs_for_worker(worker_url): + """Reschedule any jobs currently stuck on a worker back to queued.""" + conn = _conn() + conn.execute( + "UPDATE jobs SET status = 'queued', worker_url = NULL WHERE worker_url = ? AND status IN ('dispatching', 'building')", + (worker_url,), + ) + conn.commit() diff --git a/ctrl/scheduler.py b/ctrl/scheduler.py new file mode 100644 index 0000000..db1e40f --- /dev/null +++ b/ctrl/scheduler.py @@ -0,0 +1,191 @@ +import datetime +import os +import threading + +import requests + +import ctrl.db as db +import ctrl.settings as settings +import ctrl.workers as wrk + + +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: + 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()) + 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 = self._request_git(job, worker) + else: + worker_job = self._request_upload(job, worker) + except Exception as 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_job_id = worker_job.get("id") + worker_url = worker["url"] + db.update_job(job_id, status="building", worker_job_id=worker_job_id, worker_url=worker_url) + + # download artifacts + try: + self._fetch_artifacts(job, worker, worker_job_id) + db.update_job(job_id, status="done", finished_at=datetime.datetime.now().isoformat()) + self._append(job_id, "[scheduler] Build complete, artifacts saved.\n") + except Exception as 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): + self._append(job["id"], f"[scheduler] Cloning {job['source']} on {worker['url']}\n") + resp = requests.post(f"{worker['url']}/checkout_git", + json={"repo_url": job["source"]}, + timeout=settings.WORKER_BUILD_TIMEOUT, + stream=False) + return self._worker_response(resp, "checkout") + + def _request_upload(self, job, worker): + 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", + files={"file": (os.path.basename(path), f, "application/zip")}, + timeout=settings.WORKER_BUILD_TIMEOUT, + stream=False) + worker_job = self._worker_response(resp, "upload") + if not worker_job.get("id"): + # Agent may have started building before returning; try progress fallback. + pass + return worker_job + + @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() diff --git a/ctrl/settings.py b/ctrl/settings.py new file mode 100644 index 0000000..8854140 --- /dev/null +++ b/ctrl/settings.py @@ -0,0 +1,56 @@ +import json +import os + + +def _get_bool(name, default=False): + val = os.environ.get(name) + if val is None: + return default + return val.strip().lower() in ("1", "true", "yes", "on") + + +def _get_json_list(name, default=None): + raw = os.environ.get(name) + if not raw: + return list(default or []) + try: + value = json.loads(raw) + except json.JSONDecodeError: + raise ValueError(f"{name} must be a JSON array string") + if not isinstance(value, list): + raise ValueError(f"{name} must be a JSON array") + return value + + +def _get_int(name, default): + raw = os.environ.get(name) + if raw is None: + return default + return int(raw) + + +DATA_DIR = os.environ.get("CROSS_PY_DATA", os.path.join(os.path.dirname(os.path.dirname(__file__)), "data")) +DB_PATH = os.environ.get("CROSS_PY_DB", os.path.join(DATA_DIR, "jobs.db")) +BUILDS_DIR = os.environ.get("CROSS_PY_BUILDS", os.path.join(DATA_DIR, "builds")) +PORT = _get_int("CROSS_PY_PORT", 8080) + +# Static worker list: JSON array of "host:port" strings. +WORKERS = _get_json_list("CROSS_PY_WORKERS") + +# One job at a time by default; raise to allow parallel dispatches to distinct workers. +MAX_CONCURRENT = _get_int("CROSS_PY_MAX_CONCURRENT", 1) + +# Per-request timeouts (seconds) for talking to workers. +WORKER_STATUS_TIMEOUT = _get_int("CROSS_PY_WORKER_STATUS_TIMEOUT", 5) +WORKER_BUILD_TIMEOUT = _get_int("CROSS_PY_WORKER_BUILD_TIMEOUT", 3600) + +# Interval (seconds) for the scheduler loop. +SCHEDULER_INTERVAL = 1.0 + +# Debug mode uses Flask's dev server; otherwise serve through waitress. +DEBUG = _get_bool("CROSS_PY_DEBUG", False) + + +def ensure_dirs(): + os.makedirs(DATA_DIR, exist_ok=True) + os.makedirs(BUILDS_DIR, exist_ok=True) diff --git a/ctrl/static/app.js b/ctrl/static/app.js new file mode 100644 index 0000000..d8cdebf --- /dev/null +++ b/ctrl/static/app.js @@ -0,0 +1,197 @@ +"use strict"; + +const $ = (sel) => document.querySelector(sel); + +const BADGES = { done: "done", failed: "failed", cancelled: "cancelled", queued: "queued", dispatching: "dispatching", building: "building" }; + +function esc(s) { + return String(s ?? "").replace(/[&<>"']/g, (c) => + ({ "&": "&", "<": "<", ">": ">", '"': """, "'": "'" }[c]) + ); +} + +async function api(path, opts) { + const resp = await fetch(path, opts); + if (!resp.ok) { + let msg = resp.statusText; + try { msg = (await resp.json()).error || msg; } catch (_) {} + throw new Error(msg); + } + return resp.status === 204 ? null : resp.json(); +} + +async function refreshCapabilities() { + try { + const caps = await api("/api/capabilities"); + const os = new Set(caps.map((c) => c.os).filter(Boolean)); + const cpu = new Set(caps.map((c) => c.cpu).filter(Boolean)); + fillSelect($("#os"), os); + fillSelect($("#cpu"), cpu); + } catch (e) { + $("#submit-msg").textContent = `Could not reach controller API: ${e.message}`; + $("#submit-msg").className = "msg error"; + } +} + +function fillSelect(sel, values) { + const current = sel.value; + sel.innerHTML = ``; + [...values].sort().forEach((v) => { + const opt = document.createElement("option"); + opt.value = v; + opt.textContent = v; + sel.appendChild(opt); + }); + if (current) sel.value = current; +} + +async function refreshWorkers() { + try { + const workers = await api("/api/workers"); + const online = workers.filter((w) => w.online).length; + const parts = workers.map((w) => + `${w.host || w.url}${w.online ? "" : " (down)"}` + ); + $("#worker-summary").textContent = + `${online}/${workers.length} workers online — ${parts.join(" · ")}`; + } catch (_) { + $("#worker-summary").textContent = "Worker summary unavailable"; + } +} + +async function refreshJobs() { + let jobs; + try { + jobs = await api("/api/jobs"); + } catch (e) { + $("#jobs").innerHTML = `

Error: ${esc(e.message)}

`; + return; + } + const box = $("#jobs"); + if (!jobs.length) { + box.innerHTML = `

No jobs yet.

`; + return; + } + let html = ` + `; + for (const j of jobs) { + html += ` + + + + + + `; + } + html += `
IDStatusSourceTargetCreated
${esc(j.id)}${esc(j.status)}${esc(j.source)}${esc(j.os_req || "")} ${esc(j.cpu_req || "")}${esc(shortTime(j.created_at))}
`; + box.innerHTML = html; +} + +function shortTime(iso) { + if (!iso) return ""; + const d = new Date(iso); + return isNaN(d) ? iso : d.toLocaleString(); +} + +function badgeFor(status) { + return `${esc(status)}`; +} + +function openJob(id) { + const modal = $("#modal"); + modal.hidden = false; + $("#modal-title").textContent = `Job ${id}`; + $("#log").textContent = ""; + $("#artifacts").innerHTML = ""; + let src = new EventSource(`/api/jobs/${encodeURIComponent(id)}/stream`); + src.addEventListener("job", (e) => { + try { + const data = JSON.parse(e.data); + if (data.status) $("#status-badge").innerHTML = badgeFor(data.status); + } catch (_) {} + }); + src.addEventListener("log", (e) => { + let line; + try { line = JSON.parse(e.data); } catch (_) { line = e.data; } + $("#log").textContent += line + String.fromCharCode(10); + $("#log").scrollTop = $("#log").scrollHeight; + }); + src.onerror = () => src.close(); + src.onopen = () => { + // seed meta + artifacts from REST + api(`/api/jobs/${encodeURIComponent(id)}`).then((j) => { + $("#modal-meta").textContent = + `${j.source} — ${j.os_req || "any"} / ${j.cpu_req || "any"} → ${j.worker_host || "unassigned"}`; + $("#status-badge").innerHTML = badgeFor(j.status); + const arts = j.artifacts || []; + if (arts.length) { + $("#artifacts").innerHTML = arts.map((a) => + `⬇ ${esc(a)}` + ).join(""); + } + }).catch(() => {}); + }; +} + +function closeModal() { + $("#modal").hidden = true; +} + +$("#jobs").addEventListener("click", (e) => { + const tr = e.target.closest("tr[data-id]"); + if (tr) openJob(tr.dataset.id); +}); +$("#modal-close").addEventListener("click", closeModal); +$("#modal").addEventListener("click", (e) => { + if (e.target === $("#modal")) closeModal(); +}); +document.addEventListener("keydown", (e) => { + if (e.key === "Escape") closeModal(); +}); + +$("#mode").addEventListener("change", () => { + const git = $("#mode").value === "git"; + $("#url-field").hidden = !git; + $("#upload-field").hidden = git; +}); + +$("#job-form").addEventListener("submit", async (e) => { + e.preventDefault(); + const msg = $("#submit-msg"); + msg.className = "msg"; + msg.textContent = "Submitting…"; + const git = $("#mode").value === "git"; + const body = new FormData(); + body.append("os", $("#os").value); + body.append("cpu", $("#cpu").value); + if (git) { + body.append("repo_url", $("#url").value.trim()); + } else { + const file = $("#file").files[0]; + if (!file) { + msg.textContent = "Choose a zip file to upload."; + msg.className = "msg error"; + return; + } + body.append("file", file); + } + try { + const opts = { method: "POST", body }; + if (!git) opts.headers = { "Content-Type": "multipart/form-data" }; // let browser set boundary + const res = await fetch("/api/jobs", { method: "POST", body }); + const data = await res.json().catch(() => ({})); + if (!res.ok) throw new Error(data.error || res.statusText); + msg.textContent = `Submitted job ${data.id}`; + $("#job-form").reset(); + refreshJobs(); + } catch (err) { + msg.textContent = `Error: ${err.message}`; + msg.className = "msg error"; + } +}); + +refreshCapabilities(); +refreshWorkers(); +refreshJobs(); +setInterval(refreshWorkers, 15000); +setInterval(refreshJobs, 5000); diff --git a/ctrl/static/index.html b/ctrl/static/index.html new file mode 100644 index 0000000..43c1090 --- /dev/null +++ b/ctrl/static/index.html @@ -0,0 +1,68 @@ + + + + + + Cross-Py-Builder + + + +
+

Cross-Py-Builder

+

+
+ +
+
+

New build

+
+
+ + +
+
+ + +
+ +
+
+ + +
+
+ + +
+
+ +

+
+
+ +
+

Jobs

+

Loading…

+
+
+ + + + + + diff --git a/ctrl/static/style.css b/ctrl/static/style.css new file mode 100644 index 0000000..67d905e --- /dev/null +++ b/ctrl/static/style.css @@ -0,0 +1,143 @@ +/* Cross-Py-Builder styles */ +* { box-sizing: border-box; } + +:root { + --bg: #10151c; + --panel: #1a222c; + --panel-2: #202a36; + --text: #e6edf3; + --muted: #8b98a5; + --accent: #4c8bf5; + --border: #2d3a49; + --ok: #3fb950; + --fail: #f85149; +} + +body { + margin: 0; + font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; + background: var(--bg); + color: var(--text); +} + +header { + padding: 20px 28px; + border-bottom: 1px solid var(--border); + background: var(--panel); +} + +header h1 { margin: 0 0 4px; font-size: 22px; } +header p { margin: 0; color: var(--muted); font-size: 13px; } + +main { + max-width: 900px; + margin: 0 auto; + padding: 24px; + display: grid; + gap: 20px; +} + +.card { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 10px; + padding: 18px 20px; +} + +.card h2 { margin: 0 0 14px; font-size: 16px; } + +.field { margin-bottom: 12px; } +.field label { display: block; font-size: 13px; color: var(--muted); margin-bottom: 4px; } +.field input, .field select { + width: 100%; + padding: 8px 10px; + background: var(--panel-2); + border: 1px solid var(--border); + border-radius: 6px; + color: var(--text); + font-size: 14px; +} +.fields-row { display: grid; grid-template-columns: 1fr 1fr; gap: 12px; } + +button { + padding: 8px 14px; + background: var(--accent); + border: 0; + border-radius: 6px; + color: #fff; + font-size: 14px; + cursor: pointer; +} +button:hover { filter: brightness(1.1); } + +.msg { color: var(--accent); font-size: 13px; min-height: 1em; } +.msg.error { color: var(--fail); } +.muted { color: var(--muted); } + +#jobs table { width: 100%; border-collapse: collapse; font-size: 14px; } +#jobs th, #jobs td { text-align: left; padding: 8px 6px; border-bottom: 1px solid var(--border); } +#jobs th { color: var(--muted); font-weight: 600; font-size: 12px; text-transform: uppercase; } +#jobs tr.clickable { cursor: pointer; } +#jobs tr.clickable:hover { background: var(--panel-2); } + +.badge { + display: inline-block; + padding: 2px 10px; + border-radius: 20px; + font-size: 12px; + font-weight: 600; + text-transform: capitalize; +} +.badge.done { background: #12331a; color: var(--ok); } +.badge.failed { background: #331517; color: var(--fail); } +.badge.queued, .badge.dispatching, .badge.building { background: #1c2a40; color: var(--accent); } +.badge.cancelled { background: var(--panel-2); color: var(--muted); } + +.modal { + position: fixed; inset: 0; + background: rgba(0, 0, 0, 0.65); + display: flex; align-items: center; justify-content: center; + padding: 24px; +} +.modal-body { + position: relative; + width: 100%; max-width: 760px; max-height: 90vh; + overflow: auto; + background: var(--panel); + border: 1px solid var(--border); + border-radius: 10px; + padding: 22px 26px; +} +.modal-body h3 { margin: 0 0 6px; } +.modal-body .close { + position: absolute; top: 12px; right: 14px; + background: none; border: none; color: var(--muted); font-size: 22px; line-height: 1; +} +.job-meta code { color: var(--muted); font-size: 13px; word-break: break-all; } + +#log { + background: #0b0e13; + border: 1px solid var(--border); + border-radius: 6px; + padding: 12px; + margin-top: 14px; + font-size: 12px; + line-height: 1.45; + max-height: 40vh; + overflow: auto; + white-space: pre-wrap; + color: #c9d4df; +} + +#artifacts { margin-top: 12px; } +#artifacts a { + display: inline-block; + margin: 4px 8px 0 0; + padding: 6px 12px; + background: var(--panel-2); + border: 1px solid var(--border); + border-radius: 6px; + color: var(--accent); + text-decoration: none; + font-size: 13px; +} diff --git a/ctrl/workers.py b/ctrl/workers.py new file mode 100644 index 0000000..38657a5 --- /dev/null +++ b/ctrl/workers.py @@ -0,0 +1,50 @@ +import ctrl.settings as settings +import requests + + +def parse_worker(spec): + """'host:port' -> {'url': 'http://host:port', 'host': host, 'port': port}.""" + host, _, port = spec.strip().rpartition(":") + if not host or not port.isdigit(): + raise ValueError(f"Invalid worker spec: {spec!r} (expected host:port)") + return {"url": f"http://{host}:{port}", "host": host, "port": int(port)} + + +def configured_workers(): + return [parse_worker(s) for s in settings.WORKERS] + + +def probe_worker(url, timeout=None): + """Return a normalized worker status dict, or None if unreachable.""" + timeout = timeout or settings.WORKER_STATUS_TIMEOUT + try: + resp = requests.get(f"{url}/status", timeout=timeout) + resp.raise_for_status() + info = resp.json() + except (requests.RequestException, ValueError): + return None + + info["url"] = url + info["online"] = True + return info + + +def probe_all(): + workers = [] + for w in configured_workers(): + status = probe_worker(w["url"]) + if status: + workers.append(status) + else: + workers.append({"url": w["url"], "host": w["host"], "port": w["port"], "online": False}) + return workers + + +def matches(job, worker): + if worker.get("status") != "ready": + return False + if job.get("os_req") and job["os_req"].lower() not in str(worker.get("os", "")).lower(): + return False + if job.get("cpu_req") and job["cpu_req"].lower() not in str(worker.get("cpu", "")).lower(): + return False + return True diff --git a/requirements.txt b/requirements.txt index eafa3a5..2e7acab 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ Flask~=3.1.0 requests~=2.32.3 tabulate~=0.9.0 zeroconf~=0.145.1 -packaging~=24.2 \ No newline at end of file +packaging~=24.2 +waitress~=3.0.0 diff --git a/setup.py b/setup.py index cdfa891..148ea95 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,7 @@ setup( entry_points={ "console_scripts": [ "cross-py-agent=cross_py_builder.build_agent:main", - "cross-py-builder=cross_py_builder.agent_manager:main", + "cross-py-controller=ctrl.app:main", ], }, author="Brett Williams",