fix: add request wedge diagnostics
This commit is contained in:
Frank Song
committed by
nesquena-hermes
nesquena-hermes
parent
773857d159
commit
7e2709e281
160
api/request_diagnostics.py
Normal file
160
api/request_diagnostics.py
Normal file
@@ -0,0 +1,160 @@
|
||||
"""Slow request diagnostics for latency-sensitive browser API paths."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
|
||||
DEFAULT_SLOW_REQUEST_SECONDS = 5.0
|
||||
MAX_STACK_FRAMES_PER_THREAD = 40
|
||||
|
||||
|
||||
def _slow_request_seconds() -> float:
|
||||
raw = os.getenv("HERMES_WEBUI_SLOW_REQUEST_SECONDS", "").strip()
|
||||
if not raw:
|
||||
return DEFAULT_SLOW_REQUEST_SECONDS
|
||||
try:
|
||||
value = float(raw)
|
||||
except ValueError:
|
||||
return DEFAULT_SLOW_REQUEST_SECONDS
|
||||
return max(0.0, value)
|
||||
|
||||
|
||||
class RequestDiagnostics:
|
||||
"""Track request stages and emit a watchdog record if a request wedges."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
method: str,
|
||||
path: str,
|
||||
*,
|
||||
logger: logging.Logger | None = None,
|
||||
timeout_seconds: float | None = None,
|
||||
auto_start: bool = True,
|
||||
) -> None:
|
||||
self.request_id = uuid.uuid4().hex[:10]
|
||||
self.method = str(method or "-")
|
||||
self.path = str(path or "-").split("?", 1)[0]
|
||||
self.logger = logger or logging.getLogger(__name__)
|
||||
self.timeout_seconds = _slow_request_seconds() if timeout_seconds is None else max(0.0, float(timeout_seconds))
|
||||
self.started_monotonic = time.monotonic()
|
||||
self.started_wall = time.time()
|
||||
self._lock = threading.Lock()
|
||||
self._stages: list[dict[str, Any]] = []
|
||||
self._current_stage = "start"
|
||||
self._current_stage_started = self.started_monotonic
|
||||
self._finished = False
|
||||
self._watchdog_logged = False
|
||||
self._timer: threading.Timer | None = None
|
||||
if auto_start and self.timeout_seconds > 0:
|
||||
self._timer = threading.Timer(self.timeout_seconds, self._on_timeout)
|
||||
self._timer.daemon = True
|
||||
self._timer.start()
|
||||
|
||||
@classmethod
|
||||
def maybe_start(
|
||||
cls,
|
||||
method: str,
|
||||
path: str,
|
||||
*,
|
||||
logger: logging.Logger | None = None,
|
||||
) -> "RequestDiagnostics | None":
|
||||
clean_path = str(path or "").split("?", 1)[0]
|
||||
if (method.upper(), clean_path) not in {
|
||||
("GET", "/api/sessions"),
|
||||
("POST", "/api/chat/start"),
|
||||
}:
|
||||
return None
|
||||
return cls(method, clean_path, logger=logger)
|
||||
|
||||
def stage(self, name: str) -> None:
|
||||
now = time.monotonic()
|
||||
clean = str(name or "unknown").strip() or "unknown"
|
||||
with self._lock:
|
||||
if self._finished:
|
||||
return
|
||||
self._stages.append(
|
||||
{
|
||||
"name": self._current_stage,
|
||||
"ms": round((now - self._current_stage_started) * 1000, 1),
|
||||
}
|
||||
)
|
||||
self._current_stage = clean
|
||||
self._current_stage_started = now
|
||||
|
||||
def finish(self) -> None:
|
||||
timer = None
|
||||
record = None
|
||||
with self._lock:
|
||||
if self._finished:
|
||||
return
|
||||
self._finished = True
|
||||
timer = self._timer
|
||||
record = self._build_record_locked(include_stacks=False)
|
||||
if timer is not None:
|
||||
timer.cancel()
|
||||
if record and self.timeout_seconds > 0 and record["elapsed_ms"] >= self.timeout_seconds * 1000:
|
||||
self.logger.warning(
|
||||
"Slow WebUI request completed: %s",
|
||||
json.dumps(record, sort_keys=True),
|
||||
)
|
||||
|
||||
def _on_timeout(self) -> None:
|
||||
with self._lock:
|
||||
if self._finished or self._watchdog_logged:
|
||||
return
|
||||
self._watchdog_logged = True
|
||||
record = self._build_record_locked(include_stacks=True)
|
||||
self.logger.warning(
|
||||
"Slow WebUI request still running: %s",
|
||||
json.dumps(record, sort_keys=True),
|
||||
)
|
||||
|
||||
def _build_record_locked(self, *, include_stacks: bool) -> dict[str, Any]:
|
||||
now = time.monotonic()
|
||||
stages = list(self._stages)
|
||||
stages.append(
|
||||
{
|
||||
"name": self._current_stage,
|
||||
"ms": round((now - self._current_stage_started) * 1000, 1),
|
||||
}
|
||||
)
|
||||
record: dict[str, Any] = {
|
||||
"request_id": self.request_id,
|
||||
"method": self.method,
|
||||
"path": self.path,
|
||||
"started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.started_wall)),
|
||||
"elapsed_ms": round((now - self.started_monotonic) * 1000, 1),
|
||||
"current_stage": self._current_stage,
|
||||
"stages": stages,
|
||||
}
|
||||
if include_stacks:
|
||||
record["thread_stacks"] = _thread_stack_snapshot()
|
||||
return record
|
||||
|
||||
|
||||
def _thread_stack_snapshot() -> list[dict[str, Any]]:
|
||||
frames = sys._current_frames()
|
||||
threads = {thread.ident: thread for thread in threading.enumerate()}
|
||||
snapshot: list[dict[str, Any]] = []
|
||||
for ident, frame in frames.items():
|
||||
thread = threads.get(ident)
|
||||
stack = traceback.format_stack(frame, limit=MAX_STACK_FRAMES_PER_THREAD)
|
||||
snapshot.append(
|
||||
{
|
||||
"thread_id": ident,
|
||||
"thread_name": thread.name if thread else "",
|
||||
"daemon": bool(thread.daemon) if thread else None,
|
||||
"stack": [line.rstrip() for line in stack],
|
||||
}
|
||||
)
|
||||
snapshot.sort(key=lambda item: str(item.get("thread_name") or ""))
|
||||
return snapshot
|
||||
Reference in New Issue
Block a user