"""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)