Upload folder using huggingface_hub

README.md

@@ -1,13 +1,25 @@
 ---
-title: Embedding Fleet Dashboard
-emoji: 😻
-colorFrom: red
-colorTo: blue
+title: Embedding Fleet
+emoji: 🛰️
+colorFrom: blue
+colorTo: gray
 sdk: gradio
-sdk_version: 6.20.0
-python_version: '3.13'
+sdk_version: "5.49.1"
 app_file: app.py
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Embedding Fleet — Jobs control plane
+
+Run-level dashboard for a fan-out embedding run launched with
+[`launch-embedding-fleet.py`](https://huggingface.co/datasets/uv-scripts/embeddings/blob/main/launch-embedding-fleet.py)
+from [uv-scripts/embeddings](https://huggingface.co/datasets/uv-scripts/embeddings):
+documents-processed progress, tokens, live ~$ cost vs the hard timeout ceiling, ETA,
+GPU utilization, replica health, and a per-worker table.
+
+Data sources: the run bucket (manifest + worker heartbeats, polled) and the Jobs API
+(stages, durations, one metrics sample per running job per tick).
+
+Config (Space variables): `FLEET_BUCKET` (default `davanstrien/embedding-runs`),
+`FLEET_NAMESPACE`, `FLEET_POLL_SECS`. Secret: `HF_TOKEN` with read access to jobs +
+the bucket. Deep-link a run with `?run=<run-id>`.

app.py

@@ -0,0 +1,153 @@
+"""Embedding-fleet control plane — run-level view of a fan-out embedding run.
+
+Renders one run of launch-embedding-fleet.py (uv-scripts/embeddings): documents-processed
+progress, tokens, live ~$ cost vs the hard ceiling, ETA, GPU utilization, replica health,
+and a per-worker table. Polls the run bucket + Jobs API every few seconds.
+"""
+
+import os
+
+import gradio as gr
+
+from control_plane import RunView, list_runs, load_run
+
+BUCKET = os.environ.get("FLEET_BUCKET", "davanstrien/embedding-runs")
+NAMESPACE = os.environ.get("FLEET_NAMESPACE") or BUCKET.split("/")[0]
+POLL_SECS = float(os.environ.get("FLEET_POLL_SECS", "5"))
+
+CSS = """
+:root {
+  --ink: #1a1a1a; --ink-2: #555; --ink-3: #999;
+  --surface: #fffef9; --card: #ffffff; --line: #e4e2da;
+  --accent: #3d6ea5; --ok: #2e7d43; --bad: #b3382c;
+}
+@media (prefers-color-scheme: dark) {
+  :root { --ink: #ececec; --ink-2: #b0b0b0; --ink-3: #7a7a7a;
+          --surface: #131313; --card: #1c1c1c; --line: #333;
+          --accent: #7ba7d4; --ok: #6fbf85; --bad: #e07a6e; }
+}
+.gradio-container { max-width: 1080px !important; }
+#fleet-html h2 { font-weight: 600; margin: 0 0 2px; }
+.fleet-head { color: var(--ink-2); font-size: 0.92rem; margin-bottom: 14px; }
+.fleet-head a { color: var(--accent); text-decoration: none; }
+.tiles { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
+         gap: 10px; margin: 12px 0 6px; }
+.tile { background: var(--card); border: 1px solid var(--line); border-radius: 6px;
+        padding: 10px 14px; }
+.tile .k { font-size: 0.72rem; text-transform: uppercase; letter-spacing: 0.05em;
+           color: var(--ink-3); }
+.tile .v { font-size: 1.45rem; font-variant-numeric: tabular-nums; color: var(--ink); }
+.tile .s { font-size: 0.78rem; color: var(--ink-2); }
+.tile .v.ok { color: var(--ok); } .tile .v.bad { color: var(--bad); }
+.bar-wrap { margin: 10px 0 2px; }
+.bar-label { display: flex; justify-content: space-between; font-size: 0.85rem;
+             color: var(--ink-2); margin-bottom: 4px; font-variant-numeric: tabular-nums; }
+.bar { height: 10px; background: var(--line); border-radius: 5px; overflow: hidden; }
+.bar > div { height: 100%; background: var(--accent); border-radius: 5px 0 0 5px;
+             transition: width 0.6s ease; }
+"""
+
+
+def fmt_int(n):
+    return f"{n:,}" if n is not None else "—"
+
+
+def fmt_secs(s):
+    if s is None:
+        return "—"
+    if s < 90:
+        return f"{s:.0f}s"
+    if s < 5400:
+        return f"{s / 60:.0f} min"
+    return f"{s / 3600:.1f} h"
+
+
+def render_run(view: RunView) -> str:
+    m = view.manifest
+    pct = 100.0 * view.rows_done / view.rows_total if view.rows_total else 0.0
+    in_url = f"https://huggingface.co/datasets/{m['input_dataset']}"
+    out_url = f"https://huggingface.co/datasets/{m['output_dataset']}"
+    ceiling = f"ceiling ≤ ${view.cost_ceiling_usd:,.2f}" if view.cost_ceiling_usd else "est."
+    gpu = f"{view.gpu_util:.0f}%" if view.gpu_util is not None else "—"
+    health_cls = "bad" if view.errored else "ok"
+    eta = "done" if view.eta_secs == 0 else fmt_secs(view.eta_secs)
+    return f"""
+<h2>{m["output_dataset"].split("/")[-1]}</h2>
+<div class="fleet-head">
+  <a href="{in_url}">{m["input_dataset"]}</a> → <a href="{out_url}">{m["output_dataset"]}</a>
+  · <code>{m["model"].split("/")[-1]}</code>
+  · {view.num_shards} × {m["flavor"]} · run <code>{view.run_id}</code>
+</div>
+<div class="bar-wrap">
+  <div class="bar-label">
+    <span>{fmt_int(view.rows_done)} of {fmt_int(view.rows_total)} documents</span>
+    <span>{pct:.1f}%</span>
+  </div>
+  <div class="bar"><div style="width:{min(pct, 100):.2f}%"></div></div>
+</div>
+<div class="tiles">
+  <div class="tile"><div class="k">Tokens (est.)</div>
+    <div class="v">{fmt_int(view.tokens_done_est)}</div></div>
+  <div class="tile"><div class="k">Cost</div>
+    <div class="v">~${view.cost_usd:,.2f}</div><div class="s">{ceiling}</div></div>
+  <div class="tile"><div class="k">ETA</div>
+    <div class="v">{eta}</div></div>
+  <div class="tile"><div class="k">GPU util</div>
+    <div class="v">{gpu}</div><div class="s">mean of running</div></div>
+  <div class="tile"><div class="k">Replicas healthy</div>
+    <div class="v {health_cls}">{view.healthy}/{view.num_shards}</div>
+    <div class="s">{view.done} done · {view.errored} error</div></div>
+</div>
+"""
+
+
+def worker_table(view: RunView):
+    rows = []
+    for w in view.workers:
+        rows.append([
+            w.rank,
+            w.stage + (f" ({w.state})" if w.state and w.state != "running" else ""),
+            f"{w.rows_done:,}" + (f" / {w.rows_total:,}" if w.rows_total else ""),
+            f"{w.rows_per_sec:,.0f}" if w.rows_per_sec else "—",
+            f"{w.gpu_util:.0f}%" if w.gpu_util is not None else "—",
+            f"~${w.cost_usd:.3f}" if w.cost_usd is not None else "—",
+            w.job_id or "—",
+        ])
+    return rows
+
+
+def refresh(run_id):
+    if not run_id:
+        return "<p>No runs found in the bucket yet.</p>", []
+    view = load_run(BUCKET, run_id, namespace=NAMESPACE)
+    if view is None:
+        return f"<p>Run <code>{run_id}</code> has no manifest.</p>", []
+    return render_run(view), worker_table(view)
+
+
+def init(request: gr.Request):
+    runs = list_runs(BUCKET)
+    wanted = dict(request.query_params).get("run")
+    selected = wanted if wanted in runs else (runs[0] if runs else None)
+    html, table = refresh(selected)
+    return gr.Dropdown(choices=runs, value=selected), html, table
+
+
+with gr.Blocks(css=CSS, title="Embedding Fleet") as demo:
+    with gr.Row():
+        run_dd = gr.Dropdown(label="Run", choices=[], scale=3)
+        reload_btn = gr.Button("Reload runs", scale=1)
+    html = gr.HTML(elem_id="fleet-html")
+    table = gr.Dataframe(
+        headers=["rank", "stage", "rows", "rows/s", "gpu", "~$", "job"],
+        interactive=False, label="Workers",
+    )
+    timer = gr.Timer(POLL_SECS)
+
+    demo.load(init, inputs=None, outputs=[run_dd, html, table])
+    timer.tick(refresh, inputs=run_dd, outputs=[html, table])
+    run_dd.change(refresh, inputs=run_dd, outputs=[html, table])
+    reload_btn.click(lambda: gr.Dropdown(choices=list_runs(BUCKET)), outputs=run_dd)
+
+if __name__ == "__main__":
+    demo.launch()

control_plane.py

@@ -0,0 +1,244 @@
+"""Data layer for the embedding-fleet control plane.
+
+One poll tick = bucket reads (run manifest + worker heartbeats) + Jobs API reads
+(stage, durations, one metrics sample per running job). All aggregation to the
+run level happens here; app.py only renders.
+
+Cost figures are client-side estimates (flavor unit price x running time), NOT
+billing — always presented as "~$".
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import time
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from huggingface_hub import (
+    download_bucket_files,
+    fetch_job_metrics,
+    list_bucket_tree,
+    list_jobs,
+    list_jobs_hardware,
+)
+
+TERMINAL_OK = {"COMPLETED"}
+TERMINAL_BAD = {"ERROR", "CANCELED", "DELETED"}
+
+
+def _stage_name(job) -> str:
+    stage = job.status.stage if job.status else None
+    return getattr(stage, "value", None) or str(stage or "UNKNOWN")
+
+
+@dataclass
+class WorkerRow:
+    rank: int
+    job_id: str | None = None
+    stage: str = "UNKNOWN"
+    rows_done: int = 0
+    rows_total: int | None = None
+    rows_per_sec: float = 0.0
+    tokens_done_est: int | None = None
+    gpu_util: float | None = None
+    cost_usd: float | None = None
+    state: str | None = None  # worker-reported: running/done/error
+
+
+@dataclass
+class RunView:
+    run_id: str
+    manifest: dict
+    workers: list[WorkerRow] = field(default_factory=list)
+
+    # run-level aggregates
+    rows_done: int = 0
+    rows_total: int = 0
+    tokens_done_est: int = 0
+    cost_usd: float = 0.0
+    cost_ceiling_usd: float | None = None
+    eta_secs: float | None = None
+    gpu_util: float | None = None
+    healthy: int = 0
+    errored: int = 0
+    done: int = 0
+    num_shards: int = 0
+
+
+_PRICING: dict | None = None
+
+
+def pricing() -> dict:
+    global _PRICING
+    if _PRICING is None:
+        _PRICING = {hw.name: hw for hw in list_jobs_hardware()}
+    return _PRICING
+
+
+def parse_timeout_secs(timeout) -> float | None:
+    """'20m' / '1h' / '90s' / plain seconds -> seconds."""
+    if timeout is None:
+        return None
+    s = str(timeout).strip().lower()
+    try:
+        mult = {"s": 1, "m": 60, "h": 3600, "d": 86400}.get(s[-1])
+        return float(s[:-1]) * mult if mult else float(s)
+    except (ValueError, IndexError):
+        return None
+
+
+def list_runs(bucket: str) -> list[str]:
+    """Run ids under runs/, newest first (ids are timestamp-prefixed)."""
+    try:
+        ids = [Path(e.path.rstrip("/")).name
+               for e in list_bucket_tree(bucket, prefix="runs/", recursive=False)
+               if e.__class__.__name__ == "BucketFolder"]
+        # Timestamp-prefixed ids first (newest first), ad-hoc ids after.
+        return sorted(set(ids), key=lambda r: (r[:8].isdigit(), r), reverse=True) if ids else []
+    except Exception:
+        return []
+
+
+def _read_bucket_json(bucket: str, paths: list[str]) -> dict[str, dict]:
+    """Fetch small JSON files from the bucket; missing files are skipped."""
+    out: dict[str, dict] = {}
+    if not paths:
+        return out
+    with tempfile.TemporaryDirectory() as td:
+        pairs = [(p, Path(td) / p.replace("/", "__")) for p in paths]
+        try:
+            download_bucket_files(bucket, pairs, raise_on_missing_files=False)
+        except Exception:
+            return out
+        for src, dst in pairs:
+            if dst.exists():
+                try:
+                    out[src] = json.loads(dst.read_text())
+                except (json.JSONDecodeError, OSError):
+                    pass
+    return out
+
+
+def _sample_gpu_util(job_id: str, timeout: float = 3.0) -> float | None:
+    """One metrics sample -> mean GPU utilization. Never blocks past `timeout`."""
+
+    def _one():
+        gen = iter(fetch_job_metrics(job_id=job_id))
+        try:
+            raw = next(gen)
+        finally:
+            getattr(gen, "close", lambda: None)()
+        gpus = raw.get("gpus") or {}
+        utils = [float(g.get("utilization") or 0) for g in gpus.values()]
+        return sum(utils) / len(utils) if utils else None
+
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        fut = pool.submit(_one)
+        try:
+            return fut.result(timeout=timeout)
+        except Exception:
+            return None
+
+
+def _accrued_cost(job, hw_pricing: dict) -> float | None:
+    flavor = getattr(job.flavor, "value", None) or (str(job.flavor) if job.flavor else None)
+    hw = hw_pricing.get(flavor)
+    if not hw:
+        return None
+    secs = job.durations.running_secs if job.durations else None
+    if not secs and job.started_at and _stage_name(job) == "RUNNING":
+        secs = time.time() - job.started_at.timestamp()
+    if not secs:
+        return None
+    return secs / 60.0 * hw.unit_cost_usd
+
+
+def load_run(bucket: str, run_id: str, namespace: str | None = None) -> RunView | None:
+    """One full poll tick: manifest + heartbeats + job stages + metrics samples -> RunView."""
+    manifest = _read_bucket_json(bucket, [f"runs/{run_id}/run.json"]).get(f"runs/{run_id}/run.json")
+    if not manifest:
+        return None
+    n = manifest["num_shards"]
+    view = RunView(run_id=run_id, manifest=manifest, num_shards=n,
+                   rows_total=manifest.get("rows_total") or 0)
+
+    status_paths = [f"runs/{run_id}/status/{i:05d}.json" for i in range(n)]
+    statuses = _read_bucket_json(bucket, status_paths)
+
+    # Jobs by label (server-side filter); fall back to manifest job_ids via list comprehension.
+    jobs_by_id = {}
+    try:
+        for j in list_jobs(labels={"embedding-fleet-run": run_id}, namespace=namespace):
+            jobs_by_id[j.id] = j
+    except Exception:
+        pass
+    manifest_job_ids = manifest.get("job_ids") or []
+
+    hw_pricing = pricing()
+    workers: list[WorkerRow] = []
+    running_job_ids: list[str] = []
+    for rank in range(n):
+        row = WorkerRow(rank=rank)
+        st = statuses.get(f"runs/{run_id}/status/{rank:05d}.json")
+        if st:
+            row.state = st.get("state")
+            row.rows_done = st.get("rows_done") or 0
+            row.rows_total = st.get("rows_total")
+            row.rows_per_sec = st.get("rows_per_sec") or 0.0
+            row.tokens_done_est = st.get("tokens_done_est")
+            row.job_id = st.get("job_id")
+        if row.job_id is None and rank < len(manifest_job_ids):
+            row.job_id = manifest_job_ids[rank]
+        job = jobs_by_id.get(row.job_id)
+        if job is None and str(rank) in {j.labels.get("rank") for j in jobs_by_id.values() if j.labels}:
+            job = next(j for j in jobs_by_id.values() if (j.labels or {}).get("rank") == str(rank))
+        if job is not None:
+            row.stage = _stage_name(job)
+            row.cost_usd = _accrued_cost(job, hw_pricing)
+            if row.stage == "RUNNING":
+                running_job_ids.append(row.job_id)
+        workers.append(row)
+
+    # One GPU sample per running job, in parallel, bounded.
+    if running_job_ids:
+        with ThreadPoolExecutor(max_workers=min(8, len(running_job_ids))) as pool:
+            samples = dict(zip(running_job_ids,
+                               pool.map(_sample_gpu_util, running_job_ids)))
+        for row in workers:
+            if row.job_id in samples:
+                row.gpu_util = samples[row.job_id]
+
+    # Consolidator cost (labeled role=consolidate) counts toward the run.
+    consolidator_cost = sum(
+        _accrued_cost(j, hw_pricing) or 0.0
+        for j in jobs_by_id.values()
+        if (j.labels or {}).get("role") == "consolidate"
+    )
+
+    # ---- aggregate ----
+    view.workers = workers
+    view.rows_done = sum(w.rows_done for w in workers)
+    view.tokens_done_est = sum(w.tokens_done_est or 0 for w in workers)
+    view.cost_usd = sum(w.cost_usd or 0.0 for w in workers) + consolidator_cost
+    view.done = sum(1 for w in workers if w.state == "done" or w.stage in TERMINAL_OK)
+    view.errored = sum(1 for w in workers if w.state == "error" or w.stage in TERMINAL_BAD)
+    view.healthy = n - view.errored
+    gpu_vals = [w.gpu_util for w in workers if w.gpu_util is not None]
+    view.gpu_util = sum(gpu_vals) / len(gpu_vals) if gpu_vals else None
+
+    timeout_secs = parse_timeout_secs(manifest.get("timeout"))
+    hw = hw_pricing.get(manifest.get("flavor"))
+    if timeout_secs and hw:
+        view.cost_ceiling_usd = n * timeout_secs / 60.0 * hw.unit_cost_usd
+
+    active_rps = sum(w.rows_per_sec for w in workers
+                     if w.state == "running" and w.stage not in TERMINAL_BAD)
+    remaining = max((view.rows_total or 0) - view.rows_done, 0)
+    if active_rps > 0 and remaining > 0:
+        view.eta_secs = remaining / active_rps
+    elif remaining == 0 and view.rows_total:
+        view.eta_secs = 0.0
+    return view

requirements.txt

@@ -0,0 +1,2 @@
+gradio>=5.0
+huggingface-hub>=1.12