diff --git a/agent/fc_agent/app.py b/agent/fc_agent/app.py index 5ca989c..96a2b4b 100644 --- a/agent/fc_agent/app.py +++ b/agent/fc_agent/app.py @@ -7,10 +7,12 @@ queue. Config is env-seeded; the worker count is adjustable here on the fly. from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse +from . import logbuf from .config import Config from .gpu import read_gpu from .worker import Worker +logbuf.install() cfg = Config.from_env() worker = Worker(cfg) app = FastAPI(title="FabledCurator GPU agent") @@ -57,6 +59,11 @@ async def auto(request: Request): return JSONResponse(worker.status()) +@app.get("/logs") +def logs(): + return JSONResponse({"lines": list(logbuf.LINES)}) + + @app.get("/status") def status(): s = worker.status() @@ -71,51 +78,123 @@ def status(): _PAGE = """ -FabledCurator GPU agent + +FabledCurator · GPU agent -

FabledCurator GPU agent

-

FC: — · token —

-
- - +
+
+
FabledCurator GPU agent
+
—
+
+

Server — · token —

+ + + +
+
Control
+
+ + +
+ +
+ + + +
+
+
auto-tuning to fill the GPU · max 8
+
+ +
+
Status
+
+
—
state
+
0
active
+
0
processed
+
0
errors
+
0
waited out
+
+
+
GPU util—
+
+
VRAM—
+
+
+
queue —
+
+ +
+
Logs
+ 
waiting for activity…
+
-
- - workers - - - - auto-tuning to fill the GPU · max 8 -
-
- stopped
state - 0
active now - 0
processed - 0
errors - 0
waited out -
- -
GPU — …
-
-
""" diff --git a/agent/fc_agent/logbuf.py b/agent/fc_agent/logbuf.py new file mode 100644 index 0000000..c70ea66 --- /dev/null +++ b/agent/fc_agent/logbuf.py @@ -0,0 +1,40 @@ +"""In-memory log ring buffer so the control UI can show recent agent logs +(detector loads, job errors, autoscaler decisions, outage back-offs) without +needing `docker logs`. A bounded deque holds the last N formatted lines; a +logging.Handler appends to it; the UI polls /logs.""" +import logging +from collections import deque + +LINES: deque[str] = deque(maxlen=600) + + +class RingHandler(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + try: + LINES.append(self.format(record)) + except Exception: + pass + + +_installed = False + + +def install(level: int = logging.INFO) -> None: + """Attach the ring handler to the root logger once. fc_agent module loggers + propagate to root, so their records land here.""" + global _installed + if _installed: + return + _installed = True + h = RingHandler() + h.setFormatter( + logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s", "%H:%M:%S") + ) + root = logging.getLogger() + root.addHandler(h) + if root.level == logging.NOTSET or root.level > level: + root.setLevel(level) + # Keep the buffer signal-rich: drop the per-request access spam + ultralytics + # banner noise to WARNING. + logging.getLogger("uvicorn.access").setLevel(logging.WARNING) + logging.getLogger("ultralytics").setLevel(logging.WARNING) diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index e8b21f4..49937b6 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -9,6 +9,7 @@ That's the dial the operator turns to trade desktop responsiveness for speed. Stop (or shrinking the pool) RELEASES a slot's still-leased jobs immediately so orphaned work is re-picked at once rather than waiting out the lease. """ +import logging import threading import numpy as np @@ -58,6 +59,8 @@ UTIL_HI = 96 # GPU util% considered saturated TPUT_MARGIN = 0.10 # a step up must beat the baseline by this to "help" REPROBE_TICKS = 5 # ticks to hold after settling before re-probing up +log = logging.getLogger("fc_agent.worker") + class _Slot: """One worker loop. `inflight` = jobs leased but not yet processed, so a @@ -230,6 +233,7 @@ class Worker: dt = max(1e-3, now - prev_t) tput = (self.processed - prev_p) / dt prev_p, prev_t = self.processed, now + t0 = self._target g = gpumod.read_gpu() or {} mt = g.get("mem_total_mb") or 0 @@ -260,6 +264,12 @@ class Worker: else: base_tput = None # settled → re-probe next cycle + if self._target != t0: + log.info( + "autoscale: %d→%d workers (%.2f jobs/s · util %d%% · vram %d%%)", + t0, self._target, tput, util, round(vram * 100), + ) + def _ensure_embedder(self, model_name: str): if self._embedder is not None: return self._embedder @@ -400,15 +410,21 @@ class Worker: # effort; no-ops if the server is still down, then the server's # orphan-recovery reclaims it) and signal the loop to wait. self._bump(transient=1) + log.info("curator unreachable — released job %s, backing off", + job.get("job_id")) self.client.release([job["job_id"]]) return False # A job-specific HTTP fault (404 image gone, 400) → fail it so it # doesn't re-lease forever. self._bump(errors=1) + log.warning("job %s (image %s) failed: %s", + job.get("job_id"), job.get("image_id"), str(exc)[:200]) self.client.fail(job["job_id"], str(exc)[:500]) return True except Exception as exc: # noqa: BLE001 — a genuine job fault: report it self._bump(errors=1) + log.warning("job %s (image %s) failed: %s", + job.get("job_id"), job.get("image_id"), str(exc)[:200]) self.client.fail(job["job_id"], str(exc)[:500]) return True finally: