"""有界、线程安全的请求遥测；请求线程只写内存，绝不等待监控数据库。""" import hashlib import logging import math import threading from collections import OrderedDict, deque from contextvars import ContextVar from copy import deepcopy from datetime import datetime, timedelta, timezone from time import monotonic, perf_counter from uuid import uuid4 from starlette.types import ASGIApp, Message, Receive, Scope, Send REQUEST_ID: ContextVar[str] = ContextVar("request_id", default="-") SHANGHAI = timezone(timedelta(hours=8)) # 固定上界直方图，可相加后求近似 P95；最后一档使用实际最大耗时。 BOUNDS = ( 5, 10, 25, 50, 100, 200, 300, 500, 750, 1000, 1500, 2000, 3000, 5000, 10000, 30000, 60000, 120000, 300000, ) def now_local() -> datetime: return datetime.now(SHANGHAI).replace(tzinfo=None) def minute_of(value: datetime) -> datetime: return value.replace(second=0, microsecond=0) def empty_metric() -> dict: return { "count": 0, "errors_5xx": 0, "errors_4xx": 0, "max_ms": 0.0, "histogram": [0] * (len(BOUNDS) + 1), "last_error_at": None, } def merge_metrics(items) -> dict: result = empty_metric() for item in items: for key in ("count", "errors_5xx", "errors_4xx"): result[key] += item[key] result["max_ms"] = max(result["max_ms"], item["max_ms"]) for index, count in enumerate(item["histogram"]): result["histogram"][index] += count last = item.get("last_error_at") if last and (not result["last_error_at"] or last > result["last_error_at"]): result["last_error_at"] = last result["error_rate"] = ( round(result["errors_5xx"] / result["count"] * 100, 2) if result["count"] else 0 ) result["p95_ms"] = percentile(result) return result def percentile(metric: dict) -> float | None: if not metric["count"]: return None target, count = math.ceil(metric["count"] * 0.95), 0 for index, value in enumerate(metric["histogram"]): count += value if count >= target: return min(BOUNDS[index], metric["max_ms"]) if index < len(BOUNDS) else metric["max_ms"] return None def fingerprint(code: str, component: str = "") -> str: return hashlib.sha256(f"{code}:{component}".encode()).hexdigest() class Recorder: """最多保留十分钟、每分钟 200 个路由、2000 条待写事件。""" def __init__(self): self.lock = threading.Lock() self.minutes = OrderedDict() self.events = deque(maxlen=2000) self.dropped_events = 0 self.started = monotonic() self.instance_id = uuid4().hex[:16] self.slow_ms = 2000 def record(self, method, route, status, elapsed, request_id, *, at=None): at = at or now_local() key = minute_of(at) with self.lock: endpoints = self.minutes.setdefault(key, {}) endpoint = (method, route) if endpoint not in endpoints and len(endpoints) >= 200: endpoint = ("OTHER", "/__other__") metric = endpoints.setdefault(endpoint, empty_metric()) metric["count"] += 1 metric["errors_5xx"] += int(status >= 500) metric["errors_4xx"] += int(400 <= status < 500) metric["max_ms"] = max(metric["max_ms"], round(elapsed, 2)) bucket = next((i for i, bound in enumerate(BOUNDS) if elapsed <= bound), len(BOUNDS)) metric["histogram"][bucket] += 1 if status >= 500: metric["last_error_at"] = at.isoformat() while len(self.minutes) > 10: self.minutes.popitem(last=False) if status >= 500 or elapsed >= self.slow_ms: self._event( { "kind": "REQUEST_ERROR" if status >= 500 else "SLOW_REQUEST", "code": "HTTP_5XX" if status >= 500 else "REQUEST_SLOW", "component": "http", "message": "接口处理异常，请按请求编号检索服务端日志。" if status >= 500 else "接口响应较慢，请核对同一时段资源和任务。", "request_id": request_id, "method": method, "route": route, "status_code": status, "duration_ms": round(elapsed, 2), "first_seen": at, "last_seen": at, "occurrences": 1, "level": "error" if status >= 500 else "warning", } ) def _event(self, event): if len(self.events) == self.events.maxlen: self.dropped_events += 1 self.events.append(event) def runtime_error(self, component): with self.lock: at = now_local() self._event( { "kind": "RUNTIME_ERROR", "code": "BACKGROUND_ERROR", "component": component[:128], "message": "后台运行出现错误，请检索服务端日志。", "first_seen": at, "last_seen": at, "occurrences": 1, "level": "error", } ) def metrics(self, start=None, end=None): with self.lock: return deepcopy( { key: value for key, value in self.minutes.items() if (start is None or key >= start) and (end is None or key < end) } ) def take_events(self): with self.lock: values = list(self.events) self.events.clear() return values def restore_events(self, values): with self.lock: for value in reversed(values): if len(self.events) < self.events.maxlen: self.events.appendleft(value) else: self.dropped_events += 1 class MonitorMiddleware: """纯 ASGI 中间件：包含异常、流式响应与取消路径，避免读取请求正文。""" def __init__(self, app: ASGIApp, recorder: Recorder): self.app, self.recorder = app, recorder async def __call__(self, scope: Scope, receive: Receive, send: Send): if scope["type"] != "http": await self.app(scope, receive, send) return request_id = uuid4().hex scope.setdefault("state", {})["request_id"] = request_id token = REQUEST_ID.set(request_id) started, status = perf_counter(), 500 async def with_id(message: Message): nonlocal status if message["type"] == "http.response.start": status = message["status"] message["headers"] = [ (k, v) for k, v in message.get("headers", []) if k.lower() != b"x-request-id" ] message["headers"].append((b"x-request-id", request_id.encode())) if scope.get("path", "").startswith("/api/v1/system/monitor"): message["headers"].append((b"cache-control", b"no-store")) await send(message) try: await self.app(scope, receive, with_id) except Exception: status = 500 raise finally: try: route = getattr(scope.get("route"), "path", "/__unmatched__") excluded = route.startswith( ("/api/v1/system/monitor", "/api/health", "/api/docs", "/api/openapi") ) if not excluded: method = scope.get("method", "OTHER") if method not in {"GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}: method = "OTHER" self.recorder.record( method, route, status, (perf_counter() - started) * 1000, request_id ) except Exception: logging.getLogger(__name__).warning("请求遥测写入失败", exc_info=True) finally: REQUEST_ID.reset(token)