259 lines
13 KiB
Python
259 lines
13 KiB
Python
"""Opt-in, bounded HTTP timing. Never consumes/rebuffers a streaming body."""
|
|
|
|
import logging
|
|
import math
|
|
import re
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
|
|
from app.gateway.perf_logging import is_verbose_path, should_log_request, verbose_request_logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def route_name(path):
|
|
path = re.sub(r"(/api/threads/)(?!search(?:/|$)|count(?:/|$))[^/]+", r"\1{thread_id}", path)
|
|
path = re.sub(r"(/runs/)[^/]+(?=/|$)", r"\1{run_id}", path) if not path.endswith(("/runs/stream", "/runs/wait")) else path
|
|
return re.sub(r"[0-9a-f]{8}-[0-9a-f-]{27,}", "{id}", path, flags=re.I)[:240]
|
|
|
|
|
|
def percentile(values, percent):
|
|
values = sorted(v for v in values if v is not None)
|
|
return round(values[max(0, math.ceil(len(values) * percent / 100) - 1)], 2) if values else None
|
|
|
|
|
|
def _database_pool_snapshot():
|
|
try:
|
|
from deerflow.persistence.engine import get_engine
|
|
|
|
engine = get_engine()
|
|
if engine is None:
|
|
return None
|
|
pool = engine.sync_engine.pool
|
|
return {
|
|
"backend": engine.dialect.name,
|
|
"checked_out": pool.checkedout(),
|
|
"size": pool.size(),
|
|
"overflow": pool.overflow(),
|
|
"max_overflow": getattr(pool, "_max_overflow", None),
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
class RequestRecorder:
|
|
def __init__(self):
|
|
self.enabled = False
|
|
self.completed = deque(maxlen=2000)
|
|
self.pending = deque(maxlen=4000)
|
|
self.active = {}
|
|
|
|
def finish(self, record):
|
|
self.completed.append(record)
|
|
self.pending.append(record)
|
|
self.active.pop(record["id"], None)
|
|
|
|
def summary(self):
|
|
now = time.time()
|
|
rows = [r for r in self.completed if r["started"] >= now - 900]
|
|
grouped = {}
|
|
for r in rows:
|
|
# Preserve where the request actually ran. A gateway-only
|
|
# aggregate hides a saturated Worker behind a healthy main node.
|
|
key = (r.get("node") or "main", r["method"], r["path"], r["stream"])
|
|
grouped.setdefault(key, []).append(r)
|
|
aggregates = []
|
|
for (node, method, path, stream), records in grouped.items():
|
|
aggregates.append({"node": node, "method": method, "path": path, "stream": stream, "count": len(records),
|
|
"errors": sum(r["status"] >= 400 for r in records),
|
|
"p95_headers_ms": percentile([r["headers_ms"] for r in records], 95),
|
|
"p95_ms": percentile([r["total_ms"] for r in records], 95),
|
|
"max_ms": max(r["total_ms"] for r in records)})
|
|
return {"window_seconds": 900, "count": len(rows), "errors": sum(r["status"] >= 400 for r in rows),
|
|
"p95_headers_ms": percentile([r["headers_ms"] for r in rows], 95),
|
|
"routes": sorted(aggregates, key=lambda r: r["max_ms"], reverse=True)[:40],
|
|
"slow_requests": sorted(rows, key=lambda r: r["total_ms"], reverse=True)[:30],
|
|
"active": [{**r, "elapsed_ms": round((now - r["started"]) * 1000, 1)} for r in list(self.active.values())[:100]]}
|
|
|
|
def worker_status(self):
|
|
"""Bounded status payload for the main process' 5-second health probe."""
|
|
summary = self.summary()
|
|
return {
|
|
"count": summary["count"],
|
|
"errors": summary["errors"],
|
|
"p95_headers_ms": summary["p95_headers_ms"],
|
|
"routes": summary["routes"][:8],
|
|
"active_count": len(self.active),
|
|
}
|
|
|
|
|
|
class RequestTelemetryMiddleware:
|
|
def __init__(self, app):
|
|
self.app = app
|
|
self._active_requests = 0
|
|
self._active_by_path = {}
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
if scope["type"] != "http":
|
|
return await self.app(scope, receive, send)
|
|
recorder = getattr(scope["app"].state, "request_recorder", None)
|
|
path = scope.get("path", "")
|
|
if not path.startswith("/api/") or path.startswith(("/api/system-settings/worker-pool", "/api/health")):
|
|
return await self.app(scope, receive, send)
|
|
|
|
record_request = recorder is not None and recorder.enabled and len(recorder.active) < 2000
|
|
begin = time.perf_counter()
|
|
request_id = uuid.uuid4().hex[:12]
|
|
route_key = route_name(path)
|
|
self._active_requests += 1
|
|
self._active_by_path[route_key] = self._active_by_path.get(route_key, 0) + 1
|
|
api_inflight_start = self._active_requests
|
|
route_inflight_start = self._active_by_path[route_key]
|
|
record = {"id": request_id, "started": time.time(), "method": scope["method"], "path": route_name(path),
|
|
"status": 499, "headers_ms": None, "first_body_ms": None, "stream": False} if record_request else None
|
|
if record is not None:
|
|
recorder.active[record["id"]] = record
|
|
state = scope.setdefault("state", {})
|
|
state["diagnostic_request_id"] = request_id
|
|
state["perf_api_inflight_start"] = api_inflight_start
|
|
state["perf_route_inflight_start"] = route_inflight_start
|
|
request_context_token = None
|
|
try:
|
|
from deerflow.persistence.engine import bind_performance_request
|
|
|
|
request_context_token = bind_performance_request(request_id)
|
|
except Exception:
|
|
logger.debug("PERF request ID could not be attached to the SQL context", exc_info=True)
|
|
|
|
async def measured_send(message):
|
|
if message["type"] == "http.response.start":
|
|
headers_ms = round((time.perf_counter() - begin) * 1000, 2)
|
|
headers = list(message.get("headers", []))
|
|
is_stream = any(k.lower() == b"content-type" and b"text/event-stream" in v for k, v in headers)
|
|
state["perf_headers_ms"] = headers_ms
|
|
state["perf_stream"] = is_stream
|
|
state["perf_status"] = message["status"]
|
|
if record is not None:
|
|
record["status"] = message["status"]
|
|
record["headers_ms"] = headers_ms
|
|
record["stream"] = is_stream
|
|
# Record existing operation phases without query text or headers.
|
|
phases = next((v.decode("latin-1")[:1000] for k, v in headers if k.lower() == b"server-timing"), "")
|
|
if record is not None:
|
|
record["phases"] = re.sub(r"[^a-zA-Z0-9_.,;= -]", "", phases)
|
|
headers.append((b"x-deerflow-request-id", request_id.encode()))
|
|
message = {**message, "headers": headers}
|
|
elif message["type"] == "http.response.body" and message.get("body") and state.get("perf_first_body_ms") is None:
|
|
state["perf_first_body_ms"] = round((time.perf_counter() - begin) * 1000, 2)
|
|
if record is not None:
|
|
record["first_body_ms"] = state["perf_first_body_ms"]
|
|
await send(message)
|
|
|
|
try:
|
|
await self.app(scope, receive, measured_send)
|
|
except BaseException as exc:
|
|
cancelled = type(exc).__name__.lower() in {"cancelled", "cancellederror"} or "cancelled" in type(exc).__name__.lower()
|
|
if record is not None and record["headers_ms"] is None:
|
|
record["status"] = 499 if cancelled else 500
|
|
state["perf_exception_type"] = type(exc).__name__
|
|
state["perf_cancelled"] = cancelled
|
|
if record is None or record["headers_ms"] is None:
|
|
state["perf_status"] = 499 if cancelled else 500
|
|
raise
|
|
finally:
|
|
total_ms = round((time.perf_counter() - begin) * 1000, 2)
|
|
route = scope.get("route")
|
|
route_path = getattr(route, "path", None) or route_name(path)
|
|
route_path = route_path[:240]
|
|
headers_ms = record.get("headers_ms") if record is not None else state.get("perf_headers_ms")
|
|
is_stream = bool(state.get("perf_stream", record.get("stream") if record is not None else False))
|
|
status = record.get("status", 500) if record is not None else state.get("perf_status", 500)
|
|
exception_type = state.get("perf_exception_type")
|
|
node = state.get("worker_node", "main")
|
|
trace = state.get("perf_trace")
|
|
threshold = should_log_request(
|
|
elapsed_ms=total_ms,
|
|
headers_ms=headers_ms,
|
|
first_body_ms=state.get("perf_first_body_ms"),
|
|
is_stream=is_stream,
|
|
)
|
|
verbose = verbose_request_logging() and is_verbose_path(path)
|
|
cancelled = state.get("perf_cancelled", False)
|
|
concurrency_detail = api_inflight_start >= 8
|
|
detailed = threshold or verbose or status >= 500 or (exception_type is not None and not cancelled) or concurrency_detail
|
|
if detailed:
|
|
endpoint_ms = trace.get("endpoint_ms") if isinstance(trace, dict) else None
|
|
pool_snapshot = _database_pool_snapshot()
|
|
level = logging.ERROR if not cancelled and (status >= 500 or exception_type is not None) else logging.WARNING if threshold else logging.INFO
|
|
logger.log(
|
|
level,
|
|
"PERF http_request id=%s method=%s path=%s status=%s total_ms=%.2f headers_ms=%s first_body_ms=%s stream=%s "
|
|
"api_inflight_start=%s route_inflight_start=%s api_inflight_end=%s route_inflight_end=%s db_pool=%s node=%s "
|
|
"worker_access_ms=%s worker_upstream_headers_ms=%s worker_fallback=%s event_loop_lag_ms=%s "
|
|
"exception_type=%s endpoint=%s endpoint_ms=%s outside_handler_ms=%s stages=%s",
|
|
request_id,
|
|
scope["method"],
|
|
route_path,
|
|
status,
|
|
total_ms,
|
|
headers_ms,
|
|
state.get("perf_first_body_ms"),
|
|
is_stream,
|
|
api_inflight_start,
|
|
route_inflight_start,
|
|
self._active_requests,
|
|
self._active_by_path.get(route_key, 0),
|
|
pool_snapshot,
|
|
node,
|
|
state.get("worker_access_ms"),
|
|
state.get("worker_upstream_headers_ms"),
|
|
state.get("worker_pool_fallback_reason"),
|
|
getattr(scope["app"].state, "perf_event_loop_lag_ms", None),
|
|
exception_type,
|
|
trace.get("endpoint") if isinstance(trace, dict) else None,
|
|
endpoint_ms,
|
|
round(max(0.0, headers_ms - endpoint_ms), 2) if headers_ms is not None and endpoint_ms is not None else None,
|
|
{"stages": trace.get("stages"), "details": trace.get("details")} if isinstance(trace, dict) else None,
|
|
)
|
|
else:
|
|
logger.info(
|
|
"PERF http_request id=%s method=%s path=%s status=%s total_ms=%.2f headers_ms=%s api_inflight_start=%s route_inflight_start=%s api_inflight_end=%s route_inflight_end=%s node=%s fallback=%s exception_type=%s",
|
|
request_id,
|
|
scope["method"],
|
|
route_path,
|
|
status,
|
|
total_ms,
|
|
headers_ms,
|
|
api_inflight_start,
|
|
route_inflight_start,
|
|
self._active_requests,
|
|
self._active_by_path.get(route_key, 0),
|
|
node,
|
|
state.get("worker_pool_fallback_reason"),
|
|
exception_type,
|
|
)
|
|
state["perf_status"] = status
|
|
if record is not None:
|
|
record["total_ms"] = total_ms
|
|
record["node"] = node
|
|
record["access_ms"] = state.get("worker_access_ms")
|
|
record["upstream_headers_ms"] = state.get("worker_upstream_headers_ms")
|
|
record["fallback_reason"] = state.get("worker_pool_fallback_reason")
|
|
if record is not None:
|
|
record["path"] = route_path
|
|
recorder.finish(record)
|
|
if request_context_token is not None:
|
|
try:
|
|
from deerflow.persistence.engine import reset_performance_request
|
|
|
|
reset_performance_request(request_context_token)
|
|
except Exception:
|
|
logger.debug("PERF request ID could not be cleared from the SQL context", exc_info=True)
|
|
self._active_requests = max(0, self._active_requests - 1)
|
|
remaining_for_path = self._active_by_path.get(route_key, 0) - 1
|
|
if remaining_for_path > 0:
|
|
self._active_by_path[route_key] = remaining_for_path
|
|
else:
|
|
self._active_by_path.pop(route_key, None)
|