f408a555f8
Diagnose-Sichtbarkeit (Plan A):
- Job.step_started_at + last_heartbeat_at, Heartbeat-Task pingt alle 3 s
während laufender Mac-Calls
- Mac-Worker hält per X-Job-Id Header einen Log-Ringbuffer pro Job
(200 Zeilen, 1 h TTL); GET /api/jobs/{id}/log liefert ihn aus
- LXC-Endpoint GET /api/jobs/{id}/diag bündelt Job-Status, Stage-Sekunden,
Heartbeat-Alter und Mac-Worker-Log
- Frontend: Live-Timing im Status-Pill ("läuft seit Xs · letzter Ping vor Ys"),
klappbares Diagnose-Panel mit Mac-Log für FAILED + langlaufende Jobs
Robustheit (Plan B):
- MacClient: einmaliger Retry mit 3 s Backoff bei RemoteProtocolError /
ConnectError / ReadError für summarize/protocol/export
- Mac-Worker /api/preload heizt Ollama vor (keep_alive 30 m, num_predict 1);
Pipeline ruft Preload vor summarize, damit Modell-Reload nicht mitten
in der Generierung passiert (Hauptursache "Server disconnected")
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
"""Per-Job Log-Ringbuffer.
|
|
|
|
Während ein HTTP-Request läuft, setzt eine Middleware die ContextVar `current_job`
|
|
auf die `X-Job-Id` aus dem Header. Ein Logging-Handler greift diesen Wert ab und
|
|
schreibt jede Logzeile zusätzlich in einen Ringbuffer für genau diesen Job.
|
|
|
|
Die LXC-Frontend kann den Buffer per `GET /api/jobs/{job_id}/log` abrufen und in
|
|
der Diagnose-Anzeige rendern — der User sieht so ohne SSH-Zugang zum Mac, was
|
|
während eines hängenden Calls passiert.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from collections import deque
|
|
from contextvars import ContextVar
|
|
from threading import Lock
|
|
|
|
# Per-Coroutine: welcher Job-ID gehört dem aktuellen Async-Kontext.
|
|
current_job: ContextVar[int | None] = ContextVar("current_job", default=None)
|
|
|
|
_RING_MAX = 200 # Zeilen pro Job
|
|
_TTL_SECONDS = 3600 # alte Buffers nach 1h löschen
|
|
_buffers: dict[int, dict] = {}
|
|
_lock = Lock()
|
|
|
|
|
|
def append(job_id: int, line: str) -> None:
|
|
now = time.time()
|
|
with _lock:
|
|
buf = _buffers.get(job_id)
|
|
if buf is None:
|
|
buf = _buffers[job_id] = {"lines": deque(maxlen=_RING_MAX), "last": now}
|
|
ts = time.strftime("%H:%M:%S", time.localtime(now))
|
|
buf["lines"].append(f"{ts} {line}")
|
|
buf["last"] = now
|
|
# Lazy GC: alte Buffer rauswerfen.
|
|
for jid in list(_buffers):
|
|
if now - _buffers[jid]["last"] > _TTL_SECONDS:
|
|
_buffers.pop(jid, None)
|
|
|
|
|
|
def get(job_id: int, last_n: int = 200) -> list[str]:
|
|
with _lock:
|
|
buf = _buffers.get(job_id)
|
|
if buf is None:
|
|
return []
|
|
return list(buf["lines"])[-last_n:]
|
|
|
|
|
|
class JobLogHandler(logging.Handler):
|
|
"""Spiegelt jede Log-Zeile in den Ringbuffer des aktuellen Jobs."""
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
jid = current_job.get()
|
|
if jid is None:
|
|
return
|
|
try:
|
|
msg = self.format(record)
|
|
except Exception: # noqa: BLE001
|
|
msg = record.getMessage()
|
|
append(jid, msg)
|
|
|
|
|
|
def install() -> None:
|
|
"""An Root-Logger anhängen, damit alle Module abgegriffen werden."""
|
|
h = JobLogHandler()
|
|
h.setFormatter(logging.Formatter("[%(levelname)s] %(name)s: %(message)s"))
|
|
logging.getLogger().addHandler(h)
|