Stream job logs from current offset

This commit is contained in:
2026-06-30 15:51:22 +08:00
parent e766e4ed26
commit 4b3d750df9
5 changed files with 73 additions and 10 deletions

View File

@@ -63,7 +63,10 @@ are parsed into lightweight training curves.
Job APIs and the SSE log stream also expose structured progress parsed from Job APIs and the SSE log stream also expose structured progress parsed from
YOLO, MMSeg/MMEngine, SegModel-style epoch logs, and generic tqdm percentages, YOLO, MMSeg/MMEngine, SegModel-style epoch logs, and generic tqdm percentages,
so the queue and live log panel can show stage, epoch/iteration, and percent so the queue and live log panel can show stage, epoch/iteration, and percent
without changing the original training scripts. without changing the original training scripts. Starting any web job or
dataset YOLO shortcut automatically opens its live log; the SSE stream resumes
from the current log size after the initial tail so existing lines are not
duplicated in the panel.
The coverage panel calls `GET /api/coverage` and verifies that the user-facing The coverage panel calls `GET /api/coverage` and verifies that the user-facing
scripts from the existing `Seg/` workspace are mapped to web jobs. MMSeg scripts from the existing `Seg/` workspace are mapped to web jobs. MMSeg

View File

@@ -59,6 +59,11 @@ def evaluate_project() -> dict:
and "setResults(resultsNext)" in frontend_text and "setResults(resultsNext)" in frontend_text
and "slice(0, 240)" not in frontend_text, and "slice(0, 240)" not in frontend_text,
"job_progress_ui": "JobProgressBar" in frontend_text and "progressTrack" in frontend_text, "job_progress_ui": "JobProgressBar" in frontend_text and "progressTrack" in frontend_text,
"live_log_stream_ui": "EventSource" in frontend_text
and "eventSourceRef" in frontend_text
and "log_size" in frontend_text
and "events?offset=" in frontend_text,
"live_log_offset_api": "log_size" in backend_text and "offset: int = Query(0, ge=0)" in backend_text,
"runtime_readiness_ui": "runtimeReadiness" in frontend_text and "环境就绪" in frontend_text, "runtime_readiness_ui": "runtimeReadiness" in frontend_text and "环境就绪" in frontend_text,
"capability_matrix_ui": "capabilities" in frontend_text and "全功能矩阵" in frontend_text, "capability_matrix_ui": "capabilities" in frontend_text and "全功能矩阵" in frontend_text,
"dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text, "dataset_api": "/api/datasets" in backend_text and "api_upload_dataset_files" in backend_text,

View File

@@ -39,6 +39,8 @@ app.add_middleware(
def _job_with_progress(job: dict, include_log_tail: bool = False) -> dict: def _job_with_progress(job: dict, include_log_tail: bool = False) -> dict:
enriched = dict(job) enriched = dict(job)
max_bytes = 65536 if include_log_tail else 32768 max_bytes = 65536 if include_log_tail else 32768
log_path = Path(enriched["log_path"])
enriched["log_size"] = log_path.stat().st_size if log_path.exists() else 0
enriched["progress"] = progress_from_log_path(enriched["log_path"], enriched["status"], max_bytes=max_bytes) enriched["progress"] = progress_from_log_path(enriched["log_path"], enriched["status"], max_bytes=max_bytes)
if include_log_tail: if include_log_tail:
enriched["log_tail"] = db.log_tail(enriched["log_path"], max_bytes=max_bytes) enriched["log_tail"] = db.log_tail(enriched["log_path"], max_bytes=max_bytes)
@@ -183,9 +185,9 @@ def api_cancel_job(job_id: str) -> dict:
@app.get("/api/jobs/{job_id}/events") @app.get("/api/jobs/{job_id}/events")
async def api_job_events(job_id: str): async def api_job_events(job_id: str, offset: int = Query(0, ge=0)):
async def stream(): async def stream():
last_size = 0 last_size = offset
while True: while True:
job = db.get_job(job_id) job = db.get_job(job_id)
if not job: if not job:
@@ -196,6 +198,8 @@ async def api_job_events(job_id: str):
chunk = "" chunk = ""
if path.exists(): if path.exists():
size = path.stat().st_size size = path.stat().st_size
if last_size > size:
last_size = size
if size > last_size: if size > last_size:
with path.open("rb") as handle: with path.open("rb") as handle:
handle.seek(last_size) handle.seek(last_size)

View File

@@ -1,6 +1,38 @@
from fastapi.testclient import TestClient
from app.jobs import default_conda_env_for_job from app.jobs import default_conda_env_for_job
from app.main import _job_with_progress, app
def test_mmseg_jobs_use_mmseg_conda_env_by_default(): def test_mmseg_jobs_use_mmseg_conda_env_by_default():
assert default_conda_env_for_job("mmseg.train") == "seg_mmcv" assert default_conda_env_for_job("mmseg.train") == "seg_mmcv"
assert default_conda_env_for_job("segmodel.train") == "seg_smp" assert default_conda_env_for_job("segmodel.train") == "seg_smp"
def test_job_progress_reports_log_size(tmp_path):
log_path = tmp_path / "job.log"
log_path.write_text("line one\nline two\n", encoding="utf-8")
job = {"id": "job1", "status": "running", "log_path": str(log_path)}
enriched = _job_with_progress(job, include_log_tail=True)
assert enriched["log_size"] == log_path.stat().st_size
assert enriched["log_tail"] == "line one\nline two\n"
def test_job_events_respect_log_offset(tmp_path, monkeypatch):
from app import main
log_path = tmp_path / "job.log"
old_chunk = "old chunk\n"
log_path.write_text(old_chunk + "new chunk\n", encoding="utf-8")
def fake_get_job(job_id):
return {"id": job_id, "status": "success", "log_path": str(log_path)}
monkeypatch.setattr(main.db, "get_job", fake_get_job)
response = TestClient(app).get(f"/api/jobs/job1/events?offset={len(old_chunk.encode('utf-8'))}")
assert response.status_code == 200
assert "new chunk" in response.text
assert "old chunk" not in response.text

View File

@@ -1,4 +1,4 @@
import React, { useEffect, useMemo, useState } from "react"; import React, { useEffect, useMemo, useRef, useState } from "react";
import { createRoot } from "react-dom/client"; import { createRoot } from "react-dom/client";
import { import {
Activity, Activity,
@@ -45,6 +45,7 @@ type Job = {
started_at?: string; started_at?: string;
finished_at?: string; finished_at?: string;
log_tail?: string; log_tail?: string;
log_size?: number;
params: Record<string, unknown>; params: Record<string, unknown>;
progress?: JobProgress; progress?: JobProgress;
}; };
@@ -430,6 +431,11 @@ function App() {
const [uploadFiles, setUploadFiles] = useState<FileList | null>(null); const [uploadFiles, setUploadFiles] = useState<FileList | null>(null);
const [agentValidation, setAgentValidation] = useState<ValidationAgentPayload | null>(null); const [agentValidation, setAgentValidation] = useState<ValidationAgentPayload | null>(null);
const [agentBusy, setAgentBusy] = useState(false); const [agentBusy, setAgentBusy] = useState(false);
const eventSourceRef = useRef<EventSource | null>(null);
useEffect(() => () => {
eventSourceRef.current?.close();
}, []);
const runningCount = jobs.filter((job) => job.status === "running").length; const runningCount = jobs.filter((job) => job.status === "running").length;
const successCount = jobs.filter((job) => job.status === "success").length; const successCount = jobs.filter((job) => job.status === "success").length;
@@ -515,10 +521,11 @@ function App() {
async function createJob() { async function createJob() {
setBusy(true); setBusy(true);
try { try {
await api<Job>("/api/jobs", { const job = await api<Job>("/api/jobs", {
method: "POST", method: "POST",
body: JSON.stringify({ type: taskType, params: JSON.parse(params) }) body: JSON.stringify({ type: taskType, params: JSON.parse(params) })
}); });
await inspectJob(job);
await refresh(); await refresh();
} finally { } finally {
setBusy(false); setBusy(false);
@@ -633,13 +640,14 @@ function App() {
try { try {
const generated = await createSelectedYoloYaml(); const generated = await createSelectedYoloYaml();
if (!generated) return; if (!generated) return;
await api<Job>("/api/jobs", { const job = await api<Job>("/api/jobs", {
method: "POST", method: "POST",
body: JSON.stringify({ body: JSON.stringify({
type: "yolo.train_custom", type: "yolo.train_custom",
params: { model: "YOLO11n-seg", data: generated.path, epochs: 10, imgsz: 640, batch: 1, workers: 0, device: "cpu", project: "var/custom_yolo_runs", name: selectedDataset.name, exist_ok: true } params: { model: "YOLO11n-seg", data: generated.path, epochs: 10, imgsz: 640, batch: 1, workers: 0, device: "cpu", project: "var/custom_yolo_runs", name: selectedDataset.name, exist_ok: true }
}) })
}); });
await inspectJob(job);
window.location.hash = "jobs"; window.location.hash = "jobs";
await refresh(); await refresh();
} finally { } finally {
@@ -651,13 +659,14 @@ function App() {
if (!selectedDataset?.absolute_layout) return; if (!selectedDataset?.absolute_layout) return;
setBusy(true); setBusy(true);
try { try {
await api<Job>("/api/jobs", { const job = await api<Job>("/api/jobs", {
method: "POST", method: "POST",
body: JSON.stringify({ body: JSON.stringify({
type: "yolo.predict_custom", type: "yolo.predict_custom",
params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, imgsz: 640, conf: 0.25, device: "cpu", project: "var/custom_yolo_runs", name: `${selectedDataset.name}_predict`, exist_ok: true } params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, imgsz: 640, conf: 0.25, device: "cpu", project: "var/custom_yolo_runs", name: `${selectedDataset.name}_predict`, exist_ok: true }
}) })
}); });
await inspectJob(job);
window.location.hash = "jobs"; window.location.hash = "jobs";
await refresh(); await refresh();
} finally { } finally {
@@ -669,13 +678,14 @@ function App() {
if (!selectedDataset?.absolute_layout) return; if (!selectedDataset?.absolute_layout) return;
setBusy(true); setBusy(true);
try { try {
await api<Job>("/api/jobs", { const job = await api<Job>("/api/jobs", {
method: "POST", method: "POST",
body: JSON.stringify({ body: JSON.stringify({
type: "yolo.heatmap_custom", type: "yolo.heatmap_custom",
params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, model_key: "YOLO11n-seg", cam_method: "GradCAM", target_layers: "model.model.model[9]", limit: 3, project: "var/custom_yolo_runs", name: `${selectedDataset.name}_heatmap` } params: { weights: customYoloWeightPath(selectedDataset), source: selectedDataset.absolute_layout.images, model_key: "YOLO11n-seg", cam_method: "GradCAM", target_layers: "model.model.model[9]", limit: 3, project: "var/custom_yolo_runs", name: `${selectedDataset.name}_heatmap` }
}) })
}); });
await inspectJob(job);
window.location.hash = "jobs"; window.location.hash = "jobs";
await refresh(); await refresh();
} finally { } finally {
@@ -684,15 +694,24 @@ function App() {
} }
async function inspectJob(job: Job) { async function inspectJob(job: Job) {
eventSourceRef.current?.close();
const detail = await api<Job>(`/api/jobs/${job.id}`); const detail = await api<Job>(`/api/jobs/${job.id}`);
setSelectedJob(detail); setSelectedJob(detail);
setLog(detail.log_tail ?? ""); setLog(detail.log_tail ?? "");
const source = new EventSource(`${API_BASE}/api/jobs/${job.id}/events`); const source = new EventSource(`${API_BASE}/api/jobs/${job.id}/events?offset=${detail.log_size ?? 0}`);
eventSourceRef.current = source;
source.onmessage = (event) => { source.onmessage = (event) => {
const payload = JSON.parse(event.data); const payload = JSON.parse(event.data);
if (payload.chunk) setLog((prev) => `${prev}${payload.chunk}`); if (payload.chunk) setLog((prev) => `${prev}${payload.chunk}`);
setSelectedJob(payload.job); setSelectedJob(payload.job);
if (["success", "failed", "cancelled"].includes(payload.job.status)) source.close(); if (["success", "failed", "cancelled"].includes(payload.job.status)) {
source.close();
if (eventSourceRef.current === source) eventSourceRef.current = null;
}
};
source.onerror = () => {
source.close();
if (eventSourceRef.current === source) eventSourceRef.current = null;
}; };
} }