"""请求日志中间件 — 记录最近 MAX_ENTRIES 条请求到内存 实现为纯 ASGI 中间件(而非 BaseHTTPMiddleware),原因: - BaseHTTPMiddleware 中先 `await request.body()` 再 `call_next(request)` 会消费请求体, 导致下游 handler 再次 receive 时拿不到 body,POST 请求死锁/挂起; - 单 worker 部署下,一个请求挂起会卡死整个 worker,所有请求(含 /api/v1/logs)超时。 纯 ASGI 中间件通过包装 receive/send 通道缓存请求体与响应状态,既记录日志又不破坏下游请求。 """ import time import uuid from collections import deque import logging logger = logging.getLogger(__name__) MAX_ENTRIES = 500 _request_log: deque = deque(maxlen=MAX_ENTRIES) def get_request_log() -> list: """返回请求日志副本(最近在前)""" return list(_request_log)[::-1] def clear_request_log() -> int: """清空日志,返回清空条数""" n = len(_request_log) _request_log.clear() return n def _decode_body(raw: bytes) -> str: """将请求体字节解码为可读文本(截断超长内容)""" if not raw: return "" try: text = raw.decode("utf-8") if len(text) > 2000: text = text[:2000] + "...(truncated)" return text except Exception: return f"" class RequestLogMiddleware: """纯 ASGI 请求日志中间件。 通过包装 receive 通道缓存请求体(不消费、不阻塞下游), 通过包装 send 通道捕获响应状态码与响应体,避免破坏原请求。 """ def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return start_time = time.perf_counter() method = scope.get("method", "GET") path = scope.get("path", "") client = scope.get("client") client_ip = client[0] if client else "-" # 读取或生成 X-Request-ID(Java AiGateway 在每次调用时都会注入 UUID) request_id = None for name, value in scope.get("headers", []): if name == b"x-request-id": request_id = value.decode() break request_id = request_id or str(uuid.uuid4()) # 带 trace_id 的 adapter:JsonFormatter 会把它写为顶层字段 req_logger = logging.LoggerAdapter(logger, {"trace_id": request_id}) # 缓存请求体(仅对带 body 的方法),不影响下游 receive body_bytes = b"" body_cached = False async def receive_wrapper(): nonlocal body_bytes, body_cached message = await receive() if message["type"] == "http.request": body_bytes += message.get("body", b"") if not message.get("more_body", False): body_cached = True return message # 捕获响应状态码与响应体 status_code = None response_chunks = [] sent_response_start = False async def send_wrapper(message): nonlocal status_code, response_chunks, sent_response_start if message["type"] == "http.response.start": status_code = message.get("status", 500) sent_response_start = True elif message["type"] == "http.response.body": body = message.get("body", b"") if body: response_chunks.append(body) total = sum(len(c) for c in response_chunks) if total > 1000: response_chunks[:] = [b"".join(response_chunks)[:1000]] return await send(message) try: await self.app(scope, receive_wrapper, send_wrapper) except Exception as exc: req_logger.error("REQUEST EXCEPTION: %s %s %.2fs error=%r", method, path, time.perf_counter() - start_time, exc) if not sent_response_start: from starlette.responses import JSONResponse resp = JSONResponse(status_code=500, content={"detail": str(exc)}) await resp(scope, receive, send) status_code = status_code or 500 elapsed = time.perf_counter() - start_time final_status = status_code if status_code is not None else 500 entry = { "ts": time.strftime("%H:%M:%S"), "method": method, "path": path, "status": final_status, "duration_ms": round(elapsed * 1000), "client_ip": client_ip, "trace_id": request_id, "body": _decode_body(body_bytes) if body_cached else None, "response": _decode_body(b"".join(response_chunks)[:1000]) if response_chunks else "", } _request_log.appendleft(entry) if elapsed > 5: req_logger.warning("SLOW_REQUEST: %s %s %.2fs", method, path, elapsed) else: req_logger.debug("REQUEST: %s %s %.2fs", method, path, elapsed)