126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""Low-overhead performance traces for diagnosing slow Gateway requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import faulthandler
|
|
import logging
|
|
import os
|
|
import signal
|
|
import sys
|
|
import time
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_VERBOSE_PATHS = {
|
|
"/api/v1/auth/options",
|
|
"/api/agents",
|
|
"/api/llmwiki/knowledge-bases",
|
|
}
|
|
_STACK_DUMP_REGISTERED = False
|
|
|
|
|
|
def slow_request_threshold_ms() -> float:
|
|
"""Return the warning threshold, defaulting to half a second."""
|
|
|
|
try:
|
|
return max(0.0, min(float(os.getenv("DEERFLOW_SLOW_REQUEST_LOG_MS", "500")), 60_000.0))
|
|
except (TypeError, ValueError):
|
|
return 500.0
|
|
|
|
|
|
def verbose_request_logging() -> bool:
|
|
return os.getenv("DEERFLOW_PERF_DIAGNOSTICS", "").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def is_verbose_path(path: str) -> bool:
|
|
return path in _VERBOSE_PATHS or (
|
|
path.startswith("/api/llmwiki/agents/")
|
|
and path.endswith("/knowledge-bases")
|
|
)
|
|
|
|
|
|
def should_log_request(
|
|
*,
|
|
elapsed_ms: float,
|
|
headers_ms: float | None,
|
|
first_body_ms: float | None,
|
|
is_stream: bool,
|
|
) -> bool:
|
|
threshold = slow_request_threshold_ms()
|
|
if is_stream:
|
|
return bool(
|
|
(headers_ms is not None and headers_ms >= threshold)
|
|
or (first_body_ms is not None and first_body_ms >= threshold)
|
|
)
|
|
return elapsed_ms >= threshold
|
|
|
|
|
|
def trace_request(request: Any, endpoint: str) -> dict[str, Any]:
|
|
"""Start an endpoint trace attached to the ASGI request state."""
|
|
|
|
trace: dict[str, Any] = {
|
|
"endpoint": endpoint,
|
|
"started": time.perf_counter(),
|
|
"stages": {},
|
|
"details": {},
|
|
}
|
|
try:
|
|
request.state.perf_trace = trace
|
|
except AttributeError:
|
|
# Direct router calls in isolated tests may use a minimal request stub.
|
|
pass
|
|
return trace
|
|
|
|
|
|
def trace_stage(trace: dict[str, Any], name: str, started: float) -> float:
|
|
"""Record one endpoint stage and return its completion timestamp."""
|
|
|
|
ended = time.perf_counter()
|
|
trace["stages"][name] = round((ended - started) * 1000, 2)
|
|
return ended
|
|
|
|
|
|
def finish_trace(trace: dict[str, Any]) -> None:
|
|
trace["endpoint_ms"] = round((time.perf_counter() - trace["started"]) * 1000, 2)
|
|
|
|
|
|
def register_stack_dump_signal() -> None:
|
|
"""Allow SIGUSR1 to dump all Python thread stacks to container stderr."""
|
|
|
|
global _STACK_DUMP_REGISTERED
|
|
if _STACK_DUMP_REGISTERED:
|
|
return
|
|
if os.getenv("DEERFLOW_ENABLE_STACK_DUMP", "1").strip().lower() in {"0", "false", "no", "off"}:
|
|
logger.info("PERF Python stack dump signal disabled by configuration")
|
|
return
|
|
stack_signal = getattr(signal, "SIGUSR1", None)
|
|
register = getattr(faulthandler, "register", None)
|
|
if stack_signal is None or register is None:
|
|
logger.info("PERF Python stack dump signal is unavailable on this platform")
|
|
return
|
|
try:
|
|
register(stack_signal, file=sys.stderr, all_threads=True, chain=False)
|
|
except (OSError, RuntimeError, ValueError):
|
|
logger.exception("PERF unable to register Python stack dump signal")
|
|
return
|
|
_STACK_DUMP_REGISTERED = True
|
|
logger.info("PERF Python stack dump ready signal=SIGUSR1 all_threads=true")
|
|
|
|
|
|
async def event_loop_watchdog(app: Any) -> None:
|
|
"""Log substantial event-loop stalls without requiring the diagnostics UI."""
|
|
|
|
loop = asyncio.get_running_loop()
|
|
last_warning = 0.0
|
|
while True:
|
|
expected = loop.time() + 1.0
|
|
await asyncio.sleep(1.0)
|
|
lag_ms = round(max(0.0, loop.time() - expected) * 1000, 2)
|
|
app.state.perf_event_loop_lag_ms = lag_ms
|
|
now = time.monotonic()
|
|
if lag_ms >= 200 and now - last_warning >= 5:
|
|
logger.warning("PERF event_loop_stall lag_ms=%.2f threshold_ms=200", lag_ms)
|
|
last_warning = now
|