Studio: quiet noisy logs, log real progress, and speed up Windows/macOS dataset prep (#7087)
* Studio: exclude /api/export/status from request access logs The frontend polls /api/export/status every 5s to detect export start, so it fires continuously even when idle. Each poll emitted an info request_completed access line, making up most of the server access logs. Add it to _EXCLUDED_PATHS alongside /api/train/status. The endpoint is unchanged; export state is still logged by the export modules and streamed over SSE, so no signal is lost. * Studio: collapse hub download-progress polls in the access log download-status and gguf-download-progress (plus the dataset equivalents) are polled about twice a second for the whole download, so each emitted an info request_completed line. Add them to _QUIET_POLL_PATHS so they collapse to one heartbeat line per 10s instead of one per poll. * Studio: log hub download progress at 10% steps The access log carried no real progress, only poll pings. Emit one hub_download_progress line per 10% step from the shared snapshot progress reader, so an active download shows actual percentage without a line per poll. Throttled per job and resynced if the same download restarts. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: drop successful chat thread/project CRUD from the access log A single chat turn fans out about twenty requests under /api/chat/threads and /api/chat/projects (list, fetch, per-message forks, and the message writes) that only reflect the UI re-rendering. Suppress their 2xx access line so the log keeps the signal (generation, tool calls, code execution, engine stats) and errors. Non-2xx on these paths still log. * Studio: silence transformers torch_dtype deprecation warning transformers logs "`torch_dtype` is deprecated! Use `dtype` instead!" once at model-config load via logger.warning_once (logging, not warnings), so a warnings filter cannot catch it. Attach a small logging.Filter in setup_logging, which runs before any model config is parsed, to drop that record on the transformers loggers that emit it. * Studio: quiet inference load-progress polls and log throttled load progress The frontend polls /api/inference/load-progress about twice a second for the whole model load, so each emitted a request_completed line. Add it to _QUIET_POLL_PATHS (heartbeat) and emit one inference_load_progress line per 10% step from the load-progress route, so a load shows real percentage instead of a line per poll. * Studio: fully suppress download/load progress poll access lines The download-status, download-progress, gguf-download-progress, active-downloads and transport-status polls (model and dataset), plus inference load-progress, fire ~2x/s for the whole download or load. Their progress is now reported by the hub_download_progress / inference_load_progress events (and the viewer's progress line), so the per-poll access line adds nothing. Drop it on 2xx and keep it on errors, instead of the prior 10s heartbeat. Chat CRUD suppression is folded into the same _is_quiet_success helper. * Studio: suppress training-tab model/dataset download-progress polls The training tab polls /api/models/download-progress and /api/datasets/download-progress about twice a second for the whole prep phase. These are separate routes from the /api/hub equivalents and only scan the cache, so their 2xx access line adds nothing (on Windows they always read 0 since the bytes live in snapshots/, not blobs/). Suppress the 2xx line and keep errors, alongside /api/models/gguf-download-progress. * Studio: drop transient pre-auth 401 on chat thread/project polls On first load the SPA fires chat thread/project GETs before the initial token refresh, so they 401 until /api/auth/refresh runs and the retries succeed. That pre-auth 401 is a bootstrap artifact, not an error; suppress it alongside the already-quiet 2xx line. Genuine 4xx/5xx on these paths, the download/load poll 401s, and all /api/auth/* still log. * Studio: quiet tab-switch list polls and per-poll scan/reconnect logs Switching between the Train, Export, and Chat tabs refetches list endpoints on a timer, and each hit re-logs internal detail. Heartbeat /api/train/runs, /api/models/checkpoints, /api/models/local and /api/rag/knowledge-bases (10s window, first hit and errors still log), and downgrade two per-poll INFO lines to debug: the checkpoints scan summary ("Found N training runs") and the per-reconnect SSE resume line. The meaningful "replayed N missed steps" line, logged only when steps were actually replayed, stays at info. * Studio: enable tokenizer parallelism for dataset prep on Windows/macOS TOKENIZERS_PARALLELISM was forced off everywhere to stop datasets' forked map() workers from deadlocking, but that fork only happens on Linux. On spawn platforms (Windows/macOS) dataset.map() runs in-process (dataset_map_num_proc returns None), so disabling tokenizer parallelism leaves the fast tokenizer single-threaded and dataset prep runs serially on one core. Keep it off on Linux (fork safety) and on for spawn platforms, where there is no fork to deadlock. Measured ~7x faster tokenization (12.5s -> 1.7s for 20k rows on a 32-core Windows box). * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: log throttled training status to the server log Training step/loss/epoch only went to the UI via SSE, so the server log showed inference engine_stats and train/runs heartbeats but nothing about the actual run. Emit one throttled training_progress line (step/total, percent, loss, epoch, eta) from the CUDA event pump: the first step, then at most every 30s, plus the final step, resyncing when a new run restarts the counter. Per-step UI streaming is unchanged. * Studio: quiet llama.cpp update-status polls and log throttled update progress The prebuilt llama.cpp update polls /api/llama/update-status about twice a second for the whole download and install. Suppress its 2xx access line (errors still log) and emit one throttled llama_update_progress line per 10% step from the status route, so the update shows progress without a line per poll. The existing "llama update: installing" and "llama update: success" events still bracket it. * Studio: quiet the export log-tail poll The Export tab polls /api/export/logs about once a second to stream the export subprocess output into the UI panel. Suppress its 2xx access line; the real progress is already logged as event-driven "Export subprocess status: <phase>" lines plus the subprocess start and checkpoint-loaded events, and errors still log. * studio: keep errors and mutations visible in access-log suppression Make the quiet-success access-log suppression GET-only so chat thread/project mutations (POST/PUT/DELETE) still log; only their list-poll 2xx and the transient pre-auth 401 are dropped. Suppress /api/export/status 2xx only (move it out of the all-status exclude set) so a 401/403/500 on it stays visible. Legacy /api/models and /api/datasets download-progress polls emit no hub_download_progress events, so heartbeat them via the 10s quiet-poll window instead of suppressing outright, keeping download visibility (notably on Linux). The event-emitting /api/hub download polls stay fully suppressed. Update and extend the middleware tests to cover GET-only suppression, the export-status error path, and the legacy download heartbeat. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * studio: tighten access-log and training-progress comments Comment-only pass: collapse the multi-line explanations in the logging middleware and the throttled training-progress logger to fewer lines while keeping the rationale. No behavior change. * studio: log structured export_progress phases Emit a structured export_progress event per phase (consolidated in the server log like training and download progress) instead of a plain status string, and add a phase milestone at the start of the heavy export step so the merge/save/convert is visible in the server log, not only in the forwarded stdout panel. * Studio: reset training-progress log throttle on each new run start_training rebuilds the per-run progress state but left _last_progress_log_ts/_last_progress_log_step at their prior values. A run started within 30s of a previous one whose last logged step matched the new run's first step would hit the step == prev short-circuit and drop the promised first training_progress line, then stay suppressed until the old 30s window expired. Reset both fields when a new job is accepted. * Studio: keep post-bootstrap chat 401s visible in the access log The chat thread/project 401 suppression dropped every GET 401 on those prefixes, so a genuine expired-session 401 vanished alongside the transient pre-auth race. Gate the 401 drop on a per-middleware bootstrap latch that flips once /api/auth/refresh first succeeds: before that the 401s are the pre-refresh race and are suppressed; after it any chat 401 is a real failure and logs. Add a test for the post-refresh case. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: limit chat access-log suppression to the exact list polls The chat thread/project suppression matched by startswith, so it also dropped the 2xx access line for detail and message reads (/threads/{id}, /threads/{id}/messages, /threads/{id}/messages/{id}, /projects/{id}) that are not the high-frequency list polls, losing their access and latency logging. Match the two list paths exactly instead, so only the intended list polls (and their pre-auth 401 race) are suppressed while detail and message reads keep their access line. Add a regression test. * Studio: reset inference load-progress throttle for each load The load-progress throttle (_last_load_progress_step) is a module global that persisted across loads, so a cached or small load whose first sampled /api/inference/load-progress response already reported fraction=1.0 hit step == prev (10) from a prior completed load and emitted no inference_load_progress line, while that endpoint's access log is suppressed, leaving the new load with no progress signal. Arm the throttle at load initiation in _load_model_impl so each load's first step always logs. Add a regression test. * Studio: tighten logging comments Collapse a few verbose comments (tokenizer-parallelism note, torch_dtype filter, legacy download-poll heartbeat, chat list-path suppression) to fewer lines without changing intent or code. --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Daniel Han
GitHub
parent
162cf38586
commit
a6aa4fff10
14 changed files with 407 additions and 24 deletions
|
|
@ -377,9 +377,10 @@ class ExportOrchestrator:
|
|||
|
||||
if rtype == "status":
|
||||
message = resp.get("message", "")
|
||||
logger.info("Export subprocess status: %s", message)
|
||||
# Surface status in the live log panel for high-level progress.
|
||||
# One structured export_progress line per phase (consolidated in the
|
||||
# server log, like training/download progress); also shown live.
|
||||
if message:
|
||||
logger.info("export_progress", phase = message)
|
||||
self._append_log(
|
||||
{
|
||||
"stream": "status",
|
||||
|
|
|
|||
|
|
@ -398,6 +398,19 @@ def _handle_export(backend, cmd: dict, resp_queue: Any) -> None:
|
|||
# orchestrator spawns a fresh subprocess per checkpoint load, resetting it.
|
||||
_log_forward_gate.set()
|
||||
|
||||
# Phase milestone so the heavy export step shows in the server log; the
|
||||
# merge/save/convert itself only forwards stdout to the live panel.
|
||||
_phase = {
|
||||
"merged": f"Exporting merged model ({cmd.get('format_type', '16-bit (FP16)')})...",
|
||||
"gguf": f"Exporting GGUF ({cmd.get('quantization_method', 'Q4_K_M')})...",
|
||||
"lora": "Exporting LoRA adapter...",
|
||||
"base": "Exporting base model...",
|
||||
}.get(export_type, f"Exporting ({export_type})...")
|
||||
_send_response(
|
||||
resp_queue,
|
||||
{"type": "status", "message": _phase, "ts": time.time()},
|
||||
)
|
||||
|
||||
output_path: Any = None
|
||||
try:
|
||||
if export_type == "merged":
|
||||
|
|
|
|||
|
|
@ -11,8 +11,10 @@ import os
|
|||
import sys
|
||||
import types
|
||||
|
||||
# Prevent tokenizer parallelism deadlocks when datasets forks.
|
||||
os.environ["TOKENIZERS_PARALLELISM"] = "false"
|
||||
# Off on Linux so datasets' forked map() workers can't deadlock. On spawn platforms
|
||||
# (Windows/macOS) map() runs in-process, so keep the fast tokenizer's Rust threads on
|
||||
# (the only parallelism single-process tokenize gets; off makes prep run serially).
|
||||
os.environ["TOKENIZERS_PARALLELISM"] = "true" if sys.platform in ("win32", "darwin") else "false"
|
||||
|
||||
# Make compiled cache modules importable by any subprocess. On spawn platforms
|
||||
# (Windows/macOS) spawned dataset.map() workers re-import top-level modules, and
|
||||
|
|
|
|||
|
|
@ -774,6 +774,10 @@ class TrainingBackend:
|
|||
self._should_stop = False
|
||||
self._cancel_requested = False # True only for stop(save=False)
|
||||
|
||||
# Throttled training-status logging to the server log (not one line/step).
|
||||
self._last_progress_log_ts: float = 0.0
|
||||
self._last_progress_log_step: int = -1
|
||||
|
||||
# Training metrics (consumed by routes for SSE and /metrics)
|
||||
self.loss_history: list = []
|
||||
self.lr_history: list = []
|
||||
|
|
@ -956,6 +960,10 @@ class TrainingBackend:
|
|||
self._progress = TrainingProgress(
|
||||
is_training = True, status_message = "Initializing training..."
|
||||
)
|
||||
# Reset the progress-log throttle so the new run always logs its first step,
|
||||
# even if it starts within 30s of a prior run whose last logged step matches.
|
||||
self._last_progress_log_ts = 0.0
|
||||
self._last_progress_log_step = -1
|
||||
self.loss_history.clear()
|
||||
self.lr_history.clear()
|
||||
self.step_history.clear()
|
||||
|
|
@ -1831,6 +1839,37 @@ class TrainingBackend:
|
|||
elif db_action == "finalize":
|
||||
self._finalize_run_in_db(**db_action_kwargs)
|
||||
|
||||
if etype == "progress":
|
||||
self._log_training_progress()
|
||||
|
||||
def _log_training_progress(self) -> None:
|
||||
"""One throttled training-status line to the server log (the per-step stream
|
||||
still goes to the UI via SSE): first step, then at most every 30s, plus the
|
||||
final step; resyncs on a new run. Runs on the pump thread."""
|
||||
p = self._progress
|
||||
step = int(p.step or 0)
|
||||
if step <= 0:
|
||||
return
|
||||
total = int(p.total_steps or 0)
|
||||
is_final = total > 0 and step >= total
|
||||
prev = self._last_progress_log_step
|
||||
if step == prev:
|
||||
return
|
||||
now = time.monotonic()
|
||||
if prev >= 0 and step > prev and not is_final and (now - self._last_progress_log_ts) < 30.0:
|
||||
return
|
||||
self._last_progress_log_ts = now
|
||||
self._last_progress_log_step = step
|
||||
logger.info(
|
||||
"training_progress",
|
||||
step = step,
|
||||
total_steps = total or None,
|
||||
percent = int(step * 100 / total) if total > 0 else None,
|
||||
loss = round(p.loss, 4) if p.loss is not None else None,
|
||||
epoch = round(p.epoch, 2) if p.epoch is not None else None,
|
||||
eta_s = int(p.eta_seconds) if p.eta_seconds else None,
|
||||
)
|
||||
|
||||
def _ensure_db_run_created(self) -> None:
|
||||
"""Create the DB row if it doesn't exist yet. An in-progress flag lets only one
|
||||
caller create at a time, and ``_db_run_created`` is published only after
|
||||
|
|
|
|||
|
|
@ -2190,7 +2190,11 @@ def run_training_process(*, event_queue: Any, stop_queue: Any, config: dict) ->
|
|||
stop_queue: mp.Queue for stop commands from the parent.
|
||||
config: Training config dict with all parameters.
|
||||
"""
|
||||
os.environ["TOKENIZERS_PARALLELISM"] = "false"
|
||||
# Off on Linux (forked datasets map() workers deadlock otherwise); on spawn
|
||||
# platforms map() is in-process, so keep tokenizer threads on for faster prep.
|
||||
os.environ["TOKENIZERS_PARALLELISM"] = (
|
||||
"true" if sys.platform in ("win32", "darwin") else "false"
|
||||
)
|
||||
os.environ["PYTHONWARNINGS"] = "ignore" # before imports
|
||||
|
||||
# HTTP-fallback respawn: disable Xet before any huggingface_hub import (the
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ summing stale blobs against the wrong total)."""
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
|
|
@ -34,6 +35,28 @@ logger = get_logger(__name__)
|
|||
# (repo_id, hf_token) -> (expected_total_bytes, expected_blob_hashes)
|
||||
SnapshotMetadataResolver = Callable[[str, Optional[str]], "tuple[int, frozenset[str]]"]
|
||||
|
||||
# One progress log per 10% step per job, so an active download reports progress
|
||||
# without emitting a line on every poll.
|
||||
_progress_step_lock = threading.Lock()
|
||||
_last_progress_step: dict[str, int] = {}
|
||||
|
||||
|
||||
def _log_progress_step(job_key: str, repo_id: str, variant: Optional[str], progress: float) -> None:
|
||||
step = int(progress * 10)
|
||||
with _progress_step_lock:
|
||||
last = _last_progress_step.get(job_key, -1)
|
||||
if step == last:
|
||||
return
|
||||
_last_progress_step[job_key] = step
|
||||
if step < last:
|
||||
return # download restarted; resync without logging
|
||||
logger.info(
|
||||
"hub_download_progress",
|
||||
repo_id = repo_id,
|
||||
variant = variant or "",
|
||||
percent = step * 10,
|
||||
)
|
||||
|
||||
|
||||
def _empty_progress(expected_bytes: int) -> dict:
|
||||
return {
|
||||
|
|
@ -215,6 +238,8 @@ def compute_snapshot_progress(
|
|||
else 0
|
||||
)
|
||||
)
|
||||
if force_active:
|
||||
_log_progress_step(job_key, repo_id, variant, progress)
|
||||
return {
|
||||
"downloaded_bytes": display_downloaded_bytes,
|
||||
"completed_bytes": display_completed_bytes,
|
||||
|
|
|
|||
|
|
@ -17,6 +17,15 @@ import structlog
|
|||
from loggers.handlers import filter_sensitive_data
|
||||
|
||||
|
||||
class _DropTorchDtypeDeprecation(logging.Filter):
|
||||
"""Drop transformers' once-per-run "`torch_dtype` is deprecated" warning_once.
|
||||
It is emitted via logging (not warnings), so a warnings filter cannot catch it."""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
msg = record.getMessage()
|
||||
return not ("torch_dtype" in msg and "deprecated" in msg)
|
||||
|
||||
|
||||
class LogConfig:
|
||||
"""Structured logging configuration for the application."""
|
||||
|
||||
|
|
@ -72,4 +81,13 @@ class LogConfig:
|
|||
cache_logger_on_first_use = True,
|
||||
)
|
||||
|
||||
# Drop transformers' cosmetic "`torch_dtype` is deprecated" warning_once (see filter).
|
||||
_dtype_filter = _DropTorchDtypeDeprecation()
|
||||
for _name in (
|
||||
"transformers.configuration_utils",
|
||||
"transformers.modeling_utils",
|
||||
"transformers.pipelines.base",
|
||||
):
|
||||
logging.getLogger(_name).addFilter(_dtype_filter)
|
||||
|
||||
return structlog.get_logger(service_name)
|
||||
|
|
|
|||
|
|
@ -28,19 +28,26 @@ def _env_int(name: str, default: int) -> int:
|
|||
return default
|
||||
|
||||
|
||||
# Drop duplicate successful-GET access logs repeated within the window: the SPA
|
||||
# fans one cache invalidation into many identical list fetches; only the first
|
||||
# informs. Loading polls, mutations, and errors are unaffected. 0 = log all.
|
||||
# Collapse identical GET/2xx logs within the window (the SPA fans one invalidation
|
||||
# into many list fetches). Mutations and errors always log. 0 = off.
|
||||
_ACCESS_LOG_DEDUP_MS = _env_int("UNSLOTH_STUDIO_ACCESS_LOG_DEDUP_MS", 300)
|
||||
# Pure-liveness/UI polls whose access line carries no signal beyond "client still
|
||||
# polling" (state changes are logged by their own modules). Collapsed to a longer
|
||||
# heartbeat instead of one line per poll; first hit and any error still log. 0 = off.
|
||||
# Liveness/UI polls whose line means only "still polling"; collapse to a longer
|
||||
# heartbeat. First hit and errors still log. 0 = off.
|
||||
_QUIET_POLL_DEDUP_MS = _env_int("UNSLOTH_STUDIO_ACCESS_LOG_POLL_DEDUP_MS", 10000)
|
||||
_QUIET_POLL_PATHS = {
|
||||
"/api/health",
|
||||
"/api/auth/status",
|
||||
"/api/inference/status",
|
||||
"/api/inference/monitor",
|
||||
# List polls the tabs refetch on a timer and on every tab switch.
|
||||
"/api/train/runs",
|
||||
"/api/models/checkpoints",
|
||||
"/api/models/local",
|
||||
"/api/rag/knowledge-bases",
|
||||
# Legacy download polls emit no progress events (unlike /api/hub/*), so heartbeat them.
|
||||
"/api/models/download-progress",
|
||||
"/api/models/gguf-download-progress",
|
||||
"/api/datasets/download-progress",
|
||||
}
|
||||
_DEDUP_MAP_MAX = 4096
|
||||
_NATIVE_PATH_LEASE_RE = re.compile(
|
||||
|
|
@ -62,6 +69,46 @@ _EXCLUDED_SUFFIXES = (
|
|||
".woff2",
|
||||
".ttf",
|
||||
)
|
||||
# GET polls whose 2xx line carries no signal (their progress/phase events and the UI
|
||||
# do), so drop it entirely; non-2xx still logs. Only /api/hub download polls emit
|
||||
# events; the legacy /api/models and /api/datasets ones heartbeat via _QUIET_POLL_PATHS.
|
||||
_QUIET_SUCCESS_PATHS = {
|
||||
"/api/inference/load-progress",
|
||||
"/api/llama/update-status",
|
||||
"/api/export/logs",
|
||||
"/api/export/status",
|
||||
"/api/hub/download-status",
|
||||
"/api/hub/download-progress",
|
||||
"/api/hub/gguf-download-progress",
|
||||
"/api/hub/active-downloads",
|
||||
"/api/hub/transport-status",
|
||||
"/api/hub/datasets/download-status",
|
||||
"/api/hub/datasets/download-progress",
|
||||
"/api/hub/datasets/active-downloads",
|
||||
"/api/hub/datasets/transport-status",
|
||||
}
|
||||
# The token-refresh route. Its first 2xx means the client has obtained a valid
|
||||
# session, so from then on chat 401s are real failures and must stay visible.
|
||||
_AUTH_REFRESH_PATH = "/api/auth/refresh"
|
||||
# High-frequency chat list polls; their 2xx is covered by generation/tool-call/stats
|
||||
# events. Exact paths only, so detail/message reads (/threads/{id}, .../messages,
|
||||
# /projects/{id}) keep their logs. The pre-auth 401 race also fires on these polls.
|
||||
_CHAT_LIST_PATHS = {
|
||||
"/api/chat/threads",
|
||||
"/api/chat/projects",
|
||||
}
|
||||
|
||||
|
||||
def _is_quiet_success(method: str, path: str, status_code: int, pre_auth: bool) -> bool:
|
||||
"""GET-only. Suppress a 2xx poll line that carries no signal, plus a chat list
|
||||
poll's transient pre-auth 401 (only in the bootstrap window before the first
|
||||
successful token refresh). Mutations, real (post-refresh) auth failures, and
|
||||
all other errors always log."""
|
||||
if method != "GET":
|
||||
return False
|
||||
if 200 <= status_code < 300:
|
||||
return path in _QUIET_SUCCESS_PATHS or path in _CHAT_LIST_PATHS
|
||||
return pre_auth and status_code == 401 and path in _CHAT_LIST_PATHS
|
||||
|
||||
|
||||
class LoggingMiddleware:
|
||||
|
|
@ -71,14 +118,16 @@ class LoggingMiddleware:
|
|||
self.app = app
|
||||
# (method, path, query, status_code) -> monotonic ts of the last EMITTED log.
|
||||
self._last_log: dict[tuple[str, str, bytes, int], float] = {}
|
||||
# Flips True after the first successful /api/auth/refresh; before that, chat
|
||||
# list-poll 401s are the transient bootstrap race and are suppressed.
|
||||
self._auth_refreshed = False
|
||||
|
||||
def _is_redundant_repeat(
|
||||
self, method: str, path: str, query: bytes, status_code: int, now: float
|
||||
) -> bool:
|
||||
"""True if an identical GET/2xx log fired < window ago. The query string
|
||||
is part of the identity, so distinct query-driven GETs are not collapsed.
|
||||
Mutations and non-2xx are never deduped. Quiet-poll paths use a longer
|
||||
heartbeat window. Stamps only on emit, so steady polls still log."""
|
||||
"""True if an identical GET/2xx log fired < window ago (query string is part
|
||||
of the identity). Non-GET/non-2xx never dedup; quiet-poll paths use the longer
|
||||
heartbeat. Stamps only on emit, so steady polls still log."""
|
||||
if method != "GET" or not (200 <= status_code < 300):
|
||||
return False
|
||||
window_ms = _QUIET_POLL_DEDUP_MS if path in _QUIET_POLL_PATHS else _ACCESS_LOG_DEDUP_MS
|
||||
|
|
@ -129,8 +178,16 @@ class LoggingMiddleware:
|
|||
raise
|
||||
else:
|
||||
end_time = time.perf_counter()
|
||||
if not excluded and not self._is_redundant_repeat(
|
||||
scope["method"], path, scope.get("query_string", b""), status_code, end_time
|
||||
if 200 <= status_code < 300 and path == _AUTH_REFRESH_PATH:
|
||||
self._auth_refreshed = True
|
||||
if (
|
||||
not excluded
|
||||
and not _is_quiet_success(
|
||||
scope["method"], path, status_code, not self._auth_refreshed
|
||||
)
|
||||
and not self._is_redundant_repeat(
|
||||
scope["method"], path, scope.get("query_string", b""), status_code, end_time
|
||||
)
|
||||
):
|
||||
logger.info(
|
||||
"request_completed",
|
||||
|
|
|
|||
|
|
@ -3917,6 +3917,10 @@ async def load_model(
|
|||
async def _load_model_impl(request: LoadRequest, fastapi_request: Request, current_subject: str):
|
||||
from core.inference.llama_cpp import LlamaServerNotFoundError
|
||||
|
||||
# A new load starts here; arm the progress throttle so this load's first
|
||||
# sampled step logs even if it reports 100% immediately (cached/small load).
|
||||
_reset_load_progress_step()
|
||||
|
||||
native_grant_backed = False
|
||||
model_log_label = request.model_path
|
||||
try:
|
||||
|
|
@ -5495,6 +5499,33 @@ async def get_status(current_subject: str = Depends(get_current_subject)):
|
|||
raise HTTPException(status_code = 500, detail = "Failed to get status")
|
||||
|
||||
|
||||
_load_progress_lock = threading.Lock()
|
||||
_last_load_progress_step = -1
|
||||
|
||||
|
||||
def _log_load_progress_step(fraction, phase):
|
||||
"""One inference_load_progress line per 10% step, so a model load shows
|
||||
progress without a line per poll. Reset per load by _reset_load_progress_step."""
|
||||
global _last_load_progress_step
|
||||
step = int(max(0.0, min(float(fraction), 1.0)) * 10)
|
||||
with _load_progress_lock:
|
||||
prev = _last_load_progress_step
|
||||
if step == prev:
|
||||
return
|
||||
_last_load_progress_step = step
|
||||
if step < prev:
|
||||
return # load regressed/restarted mid-poll; resync without logging
|
||||
logger.info("inference_load_progress", phase = phase or "", percent = step * 10)
|
||||
|
||||
|
||||
def _reset_load_progress_step():
|
||||
"""Arm the throttle for a new load so its first sampled step always logs,
|
||||
even a cached load that already reports fraction=1.0 on the first poll."""
|
||||
global _last_load_progress_step
|
||||
with _load_progress_lock:
|
||||
_last_load_progress_step = -1
|
||||
|
||||
|
||||
@router.get("/load-progress", response_model = LoadProgressResponse)
|
||||
async def get_load_progress(current_subject: str = Depends(get_current_subject)):
|
||||
"""
|
||||
|
|
@ -5513,7 +5544,9 @@ async def get_load_progress(current_subject: str = Depends(get_current_subject))
|
|||
progress = llama_backend.load_progress()
|
||||
if progress is None:
|
||||
return LoadProgressResponse()
|
||||
return LoadProgressResponse(**progress)
|
||||
resp = LoadProgressResponse(**progress)
|
||||
_log_load_progress_step(resp.fraction, resp.phase)
|
||||
return resp
|
||||
except Exception as e:
|
||||
logger.warning(f"Error sampling load progress: {e}")
|
||||
return LoadProgressResponse()
|
||||
|
|
|
|||
|
|
@ -14,14 +14,17 @@ never blocks on a missing marker / offline GitHub.
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from auth.authentication import get_current_subject
|
||||
from loggers import get_logger
|
||||
from utils.llama_cpp_update import get_update_status, start_update
|
||||
|
||||
logger = get_logger(__name__)
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
|
|
@ -69,6 +72,27 @@ class LlamaUpdateActionResponse(BaseModel):
|
|||
job: LlamaUpdateJob = Field(default_factory = LlamaUpdateJob)
|
||||
|
||||
|
||||
_llama_update_lock = threading.Lock()
|
||||
_last_llama_update_step = -1
|
||||
|
||||
|
||||
def _log_llama_update_progress(job: LlamaUpdateJob) -> None:
|
||||
"""One llama_update_progress line per 10% step so a prebuilt update reports
|
||||
progress without a line per poll. Resyncs when a new update starts."""
|
||||
global _last_llama_update_step
|
||||
if job.state != "running" or job.progress is None:
|
||||
return
|
||||
step = int(max(0.0, min(float(job.progress), 1.0)) * 10)
|
||||
with _llama_update_lock:
|
||||
prev = _last_llama_update_step
|
||||
if step == prev:
|
||||
return
|
||||
_last_llama_update_step = step
|
||||
if step < prev:
|
||||
return # new update; resync without logging
|
||||
logger.info("llama_update_progress", to_tag = job.to_tag or "", percent = step * 10)
|
||||
|
||||
|
||||
@router.get("/update-status", response_model = LlamaUpdateStatusResponse)
|
||||
async def llama_update_status(
|
||||
force_refresh: bool = Query(
|
||||
|
|
@ -78,7 +102,9 @@ async def llama_update_status(
|
|||
) -> LlamaUpdateStatusResponse:
|
||||
# Off the event loop: detection may probe the host and read GitHub.
|
||||
status = await asyncio.to_thread(get_update_status, force_refresh = force_refresh)
|
||||
return LlamaUpdateStatusResponse(**status)
|
||||
resp = LlamaUpdateStatusResponse(**status)
|
||||
_log_llama_update_progress(resp.job)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/update", response_model = LlamaUpdateActionResponse)
|
||||
|
|
|
|||
|
|
@ -733,7 +733,9 @@ async def stream_training_progress(
|
|||
if last_event_id is not None:
|
||||
try:
|
||||
resume_from_step = int(last_event_id)
|
||||
logger.info(f"SSE reconnect: resuming from step {resume_from_step}")
|
||||
# Fires on every reconnect (each tab switch); the meaningful signal is
|
||||
# the "replayed N missed steps" line below, logged only when N > 0.
|
||||
logger.debug(f"SSE reconnect: resuming from step {resume_from_step}")
|
||||
except ValueError:
|
||||
logger.warning(f"Invalid Last-Event-ID: {last_event_id}")
|
||||
|
||||
|
|
|
|||
48
studio/backend/tests/test_load_progress_throttle.py
Normal file
48
studio/backend/tests/test_load_progress_throttle.py
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
||||
|
||||
"""The /api/inference/load-progress throttle: one line per 10% step, reset per load."""
|
||||
|
||||
import pytest
|
||||
|
||||
import routes.inference as ri
|
||||
|
||||
|
||||
class _Capture:
|
||||
def __init__(self):
|
||||
self.events = []
|
||||
|
||||
def info(self, event, **kw):
|
||||
self.events.append((event, kw))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cap(monkeypatch):
|
||||
capture = _Capture()
|
||||
monkeypatch.setattr(ri, "logger", capture)
|
||||
ri._reset_load_progress_step()
|
||||
return capture
|
||||
|
||||
|
||||
def _percents(cap):
|
||||
return [kw["percent"] for _event, kw in cap.events]
|
||||
|
||||
|
||||
def test_new_load_first_step_logs_after_reset(cap):
|
||||
# Load A reaches 100%.
|
||||
ri._log_load_progress_step(1.0, "ready")
|
||||
assert _percents(cap) == [100]
|
||||
# Same value keeps deduping (steady poll on a finished load stays quiet).
|
||||
ri._log_load_progress_step(1.0, "ready")
|
||||
assert _percents(cap) == [100]
|
||||
# A new load arms the throttle, so a cached load B that reports 100% on its
|
||||
# first poll still emits its progress line instead of hitting step == prev.
|
||||
ri._reset_load_progress_step()
|
||||
ri._log_load_progress_step(1.0, "ready")
|
||||
assert _percents(cap) == [100, 100]
|
||||
|
||||
|
||||
def test_steady_poll_dedups_within_a_load(cap):
|
||||
for _ in range(3):
|
||||
ri._log_load_progress_step(0.3, "mmap")
|
||||
assert _percents(cap) == [30] # one line per 10% step, not one per poll
|
||||
|
|
@ -135,7 +135,7 @@ def test_duplicate_get_within_window_deduped(logs, monkeypatch):
|
|||
|
||||
mw = LoggingMiddleware(app)
|
||||
for _ in range(3):
|
||||
_run(mw(_http_scope("/api/chat/projects"), _noop_receive, send))
|
||||
_run(mw(_http_scope("/api/models/browse-folders"), _noop_receive, send))
|
||||
|
||||
# Only the first of the identical GET/200 burst is logged.
|
||||
assert len(logs.events) == 1
|
||||
|
|
@ -183,11 +183,11 @@ def test_quiet_poll_paths_use_longer_heartbeat_window(logs, monkeypatch):
|
|||
for _ in range(3):
|
||||
_run(mw(_http_scope("/api/inference/monitor"), _noop_receive, send)) # quiet
|
||||
for _ in range(3):
|
||||
_run(mw(_http_scope("/api/chat/projects"), _noop_receive, send)) # normal
|
||||
_run(mw(_http_scope("/api/models/browse-folders"), _noop_receive, send)) # normal
|
||||
|
||||
paths = [e[2]["path"] for e in logs.events]
|
||||
assert paths.count("/api/inference/monitor") == 1 # collapsed to one heartbeat
|
||||
assert paths.count("/api/chat/projects") == 3 # base dedup off -> all logged
|
||||
assert paths.count("/api/models/browse-folders") == 3 # base dedup off -> all logged
|
||||
|
||||
|
||||
def test_distinct_query_strings_are_not_deduped(logs, monkeypatch):
|
||||
|
|
@ -242,3 +242,118 @@ def test_fastapi_static_asset_success_skips_log(tmp_path, logs):
|
|||
assert response.status_code == 200
|
||||
assert response.text == "body { color: black; }"
|
||||
assert len(logs.events) == log_count
|
||||
|
||||
|
||||
def _status_app(status):
|
||||
async def app(scope, receive, send):
|
||||
await send({"type": "http.response.start", "status": status, "headers": []})
|
||||
await send({"type": "http.response.body", "body": b""})
|
||||
|
||||
return app
|
||||
|
||||
|
||||
async def _drop(message):
|
||||
pass
|
||||
|
||||
|
||||
def _paths_logged(logs):
|
||||
return [e[2]["path"] for e in logs.events]
|
||||
|
||||
|
||||
def test_quiet_success_get_2xx_suppressed(logs):
|
||||
# A GET/2xx poll on a quiet-success path logs nothing; the signal is in events.
|
||||
for path in ("/api/chat/threads", "/api/export/status", "/api/hub/download-status"):
|
||||
_run(LoggingMiddleware(_status_app(200))(_http_scope(path), _noop_receive, _drop))
|
||||
assert logs.events == []
|
||||
|
||||
|
||||
def test_chat_detail_and_message_reads_still_log(logs):
|
||||
# Only the exact list polls are suppressed; detail/message reads carry latency
|
||||
# signal and keep their access line.
|
||||
for path in (
|
||||
"/api/chat/threads/abc123",
|
||||
"/api/chat/threads/abc123/messages",
|
||||
"/api/chat/threads/abc123/messages/m1",
|
||||
"/api/chat/projects/p1",
|
||||
):
|
||||
_run(LoggingMiddleware(_status_app(200))(_http_scope(path), _noop_receive, _drop))
|
||||
assert _paths_logged(logs) == [
|
||||
"/api/chat/threads/abc123",
|
||||
"/api/chat/threads/abc123/messages",
|
||||
"/api/chat/threads/abc123/messages/m1",
|
||||
"/api/chat/projects/p1",
|
||||
]
|
||||
|
||||
|
||||
def test_quiet_success_is_get_only(logs):
|
||||
# Mutations on the same paths still log (suppression is GET-only).
|
||||
for method in ("POST", "PUT", "DELETE"):
|
||||
_run(
|
||||
LoggingMiddleware(_status_app(200))(
|
||||
_http_scope("/api/chat/threads", method = method), _noop_receive, _drop
|
||||
)
|
||||
)
|
||||
assert len(logs.events) == 3
|
||||
|
||||
|
||||
def test_chat_pre_auth_401_suppressed_other_errors_logged(logs):
|
||||
# The transient bootstrap 401 on a chat list GET is dropped, but a 500 (or any
|
||||
# other status) still logs so real failures stay visible.
|
||||
_run(
|
||||
LoggingMiddleware(_status_app(401))(_http_scope("/api/chat/projects"), _noop_receive, _drop)
|
||||
)
|
||||
assert logs.events == []
|
||||
_run(
|
||||
LoggingMiddleware(_status_app(500))(_http_scope("/api/chat/projects"), _noop_receive, _drop)
|
||||
)
|
||||
assert _paths_logged(logs) == ["/api/chat/projects"]
|
||||
|
||||
|
||||
def test_chat_401_logged_after_first_auth_refresh(logs):
|
||||
# A chat 401 before any successful token refresh is the bootstrap race and is
|
||||
# dropped, but once /api/auth/refresh has succeeded on this instance later chat
|
||||
# 401s are real failures and stay visible.
|
||||
responses: dict[tuple[str, str], int] = {}
|
||||
|
||||
async def app(scope, receive, send):
|
||||
status = responses.get((scope["method"], scope["path"]), 200)
|
||||
await send({"type": "http.response.start", "status": status, "headers": []})
|
||||
await send({"type": "http.response.body", "body": b""})
|
||||
|
||||
mw = LoggingMiddleware(app)
|
||||
|
||||
responses[("GET", "/api/chat/threads")] = 401
|
||||
_run(mw(_http_scope("/api/chat/threads"), _noop_receive, _drop))
|
||||
assert logs.events == [] # bootstrap race: suppressed
|
||||
|
||||
# A successful refresh (POST, always logged) closes the bootstrap window.
|
||||
responses[("POST", "/api/auth/refresh")] = 200
|
||||
_run(mw(_http_scope("/api/auth/refresh", method = "POST"), _noop_receive, _drop))
|
||||
assert _paths_logged(logs) == ["/api/auth/refresh"]
|
||||
|
||||
# Now the same chat 401 is a real failure and logs.
|
||||
_run(mw(_http_scope("/api/chat/threads"), _noop_receive, _drop))
|
||||
assert _paths_logged(logs) == ["/api/auth/refresh", "/api/chat/threads"]
|
||||
|
||||
|
||||
def test_export_status_error_still_logs(logs):
|
||||
# 2xx suppressed, but an HTTP-level error on export status remains visible.
|
||||
_run(
|
||||
LoggingMiddleware(_status_app(200))(_http_scope("/api/export/status"), _noop_receive, _drop)
|
||||
)
|
||||
assert logs.events == []
|
||||
_run(
|
||||
LoggingMiddleware(_status_app(500))(_http_scope("/api/export/status"), _noop_receive, _drop)
|
||||
)
|
||||
assert _paths_logged(logs) == ["/api/export/status"]
|
||||
|
||||
|
||||
def test_legacy_download_progress_heartbeats_not_suppressed(logs, monkeypatch):
|
||||
# Legacy /api/models download polls emit no progress events, so they heartbeat
|
||||
# (first hit logs, the burst collapses) rather than vanish entirely.
|
||||
monkeypatch.setattr(hmod, "_ACCESS_LOG_DEDUP_MS", 0)
|
||||
monkeypatch.setattr(hmod, "_QUIET_POLL_DEDUP_MS", 1000)
|
||||
mw = LoggingMiddleware(_status_app(200))
|
||||
for _ in range(3):
|
||||
_run(mw(_http_scope("/api/models/download-progress"), _noop_receive, _drop))
|
||||
assert _paths_logged(logs) == ["/api/models/download-progress"]
|
||||
|
|
|
|||
|
|
@ -248,7 +248,7 @@ def scan_checkpoints(
|
|||
# Sort by modification time (newest first)
|
||||
models.sort(key = lambda x: Path(x[1][0][1]).stat().st_mtime, reverse = True)
|
||||
|
||||
logger.info(f"Found {len(models)} training runs in {outputs_dir}")
|
||||
logger.debug(f"Found {len(models)} training runs in {outputs_dir}")
|
||||
return models
|
||||
|
||||
except Exception as e:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue