feat(observe): 加 RequestMetricsMiddleware(路由模板+状态码+耗时)

app/core/observe.py

@@ -6,9 +6,20 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import time
+
+from starlette.routing import Match
 
 from app.core.config import settings
 
+# 不采集的路径(纯噪音):健康检查。
+_SKIP_PATHS = frozenset({"/health"})
+# 未匹配路由(404/扫描器)归一到此,防维度爆炸。
+_UNMATCHED = "__unmatched__"
+# service 字段:与 logging.py 同源(LOG_SERVICE_NAME),默认 app-server。
+_SERVICE = os.getenv("LOG_SERVICE_NAME", "app-server")
+
 # 有界事件队列(懒创建,见 get_queue):首次取用时在运行中的 loop 里建,避免 import 期
 # 无 loop 的边角问题;put_nowait/get_nowait 不需运行中的 loop → 可在无 loop 下测试。
 _queue: asyncio.Queue[dict] | None = None
@@ -38,3 +49,62 @@ def record_event(event: dict) -> None:
         get_queue().put_nowait(event)
     except asyncio.QueueFull:
         _dropped += 1
+
+
+def _resolve_route(scope) -> str:
+    """从 scope 取路由模板(如 /things/{tid})。优先 scope['route'](现代 Starlette
+    路由后写入);取不到则手动匹配一次(老版本兜底);仍无 → __unmatched__(404/扫描器)。"""
+    route = scope.get("route")
+    path = getattr(route, "path", None)
+    if path:
+        return path
+    app_ = scope.get("app")
+    router = getattr(app_, "router", None)
+    for candidate in getattr(router, "routes", []):
+        try:
+            match, _ = candidate.matches(scope)
+        except Exception:  # noqa: BLE001 - 匹配兜底,任一路由异常不影响整体
+            continue
+        if match == Match.FULL and getattr(candidate, "path", None):
+            return candidate.path
+    return _UNMATCHED
+
+
+class RequestMetricsMiddleware:
+    """纯 ASGI 中间件:测每个 http 请求耗时,记 method/route/status/duration。
+
+    放在最外层(main.py 里 CORS 之后 add),测到含 CORS 的完整耗时。未配置观测 → 透传。
+    """
+
+    def __init__(self, app) -> None:
+        self.app = app
+
+    async def __call__(self, scope, receive, send) -> None:
+        if scope["type"] != "http" or not settings.observe_configured:
+            await self.app(scope, receive, send)
+            return
+        if scope.get("path") in _SKIP_PATHS:
+            await self.app(scope, receive, send)
+            return
+
+        start = time.perf_counter()
+        status_holder = {"status": 500}  # 下游异常未产出 response 时兜底 500
+
+        async def send_wrapper(message) -> None:
+            if message["type"] == "http.response.start":
+                status_holder["status"] = message["status"]
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_wrapper)
+        finally:
+            duration_ms = (time.perf_counter() - start) * 1000.0
+            record_event({
+                "_timestamp": int(time.time() * 1_000_000),  # µs,OpenObserve 时间列
+                "service": _SERVICE,
+                "env": settings.APP_ENV,
+                "method": scope.get("method", ""),
+                "route": _resolve_route(scope),
+                "status": status_holder["status"],
+                "duration_ms": round(duration_ms, 3),
+            })

tests/test_observe.py

@@ -7,6 +7,10 @@ from __future__ import annotations
 
 import asyncio
 
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
 from app.core import observe
 from app.core.config import settings
 
@@ -49,3 +53,66 @@ def test_record_event_drops_when_full(monkeypatch):
     assert observe.take_dropped() == 1
     assert observe.take_dropped() == 0   # 取出后清零
     assert q.get_nowait() == {"n": 1}    # 保留的是先到的
+
+
+def _make_probe_app() -> FastAPI:
+    """独立最小 app:只挂中间件 + 两个无鉴权路由,不碰真业务 DB/auth。"""
+    app = FastAPI()
+    app.add_middleware(observe.RequestMetricsMiddleware)
+
+    @app.get("/things/{tid}")
+    def get_thing(tid: str):
+        return {"tid": tid}
+
+    @app.get("/health")
+    def health():
+        return {"ok": True}
+
+    return app
+
+
+@pytest.fixture
+def observe_on(monkeypatch):
+    """开启观测 + 换一个干净小队列,返回该队列供断言。"""
+    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
+    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
+    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
+    q = asyncio.Queue(maxsize=100)
+    monkeypatch.setattr(observe, "_queue", q)
+    return q
+
+
+def test_middleware_records_route_template(observe_on):
+    client = TestClient(_make_probe_app())
+    r = client.get("/things/42")
+    assert r.status_code == 200
+    evt = observe_on.get_nowait()
+    assert evt["route"] == "/things/{tid}"   # 模板,不是 /things/42
+    assert evt["method"] == "GET"
+    assert evt["status"] == 200
+    assert evt["duration_ms"] >= 0
+    assert evt["service"] and "env" in evt and isinstance(evt["_timestamp"], int)
+
+
+def test_middleware_skips_health(observe_on):
+    client = TestClient(_make_probe_app())
+    client.get("/health")
+    assert observe_on.empty()
+
+
+def test_middleware_unmatched_route_is_normalized(observe_on):
+    client = TestClient(_make_probe_app())
+    r = client.get("/definitely-not-a-route")
+    assert r.status_code == 404
+    evt = observe_on.get_nowait()
+    assert evt["route"] == "__unmatched__"
+    assert evt["status"] == 404
+
+
+def test_middleware_noop_when_disabled(monkeypatch):
+    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
+    q = asyncio.Queue(maxsize=100)
+    monkeypatch.setattr(observe, "_queue", q)
+    client = TestClient(_make_probe_app())
+    client.get("/things/1")
+    assert q.empty()   # 未配置观测 → 零入队