From 6c624c3cea155660ddbd5f40a2465cd28a4db929 Mon Sep 17 00:00:00 2001
From: guke <guke@autohome.com.cn>
Date: Mon, 6 Jul 2026 11:04:00 +0800
Subject: [PATCH] =?UTF-8?q?feat(observe):=20=E5=8A=A0=E6=9C=89=E7=95=8C?=
 =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E9=98=9F=E5=88=97=E4=B8=8E=20record=5Fevent(?=
 =?UTF-8?q?=E6=BB=A1=E5=88=99=E4=B8=A2)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/observe.py   | 40 ++++++++++++++++++++++++++++++++++++++++
 tests/test_observe.py | 21 +++++++++++++++++++++
 2 files changed, 61 insertions(+)
 create mode 100644 app/core/observe.py

diff --git a/app/core/observe.py b/app/core/observe.py
new file mode 100644
index 0000000..1f4cb75
--- /dev/null
+++ b/app/core/observe.py
@@ -0,0 +1,40 @@
+"""接口指标埋点:有界事件队列 + 纯 ASGI 中间件。
+
+每个 HTTP 请求测总耗时、抓路由模板 + 状态码,非阻塞塞进有界队列;由 observe_worker
+后台批量上报到 OpenObserve。请求路径上无任何 I/O。未配置观测时中间件直接透传。
+"""
+from __future__ import annotations
+
+import asyncio
+
+from app.core.config import settings
+
+# 有界事件队列(懒创建,见 get_queue):首次取用时在运行中的 loop 里建,避免 import 期
+# 无 loop 的边角问题;put_nowait/get_nowait 不需运行中的 loop → 可在无 loop 下测试。
+_queue: "asyncio.Queue[dict] | None" = None
+# 队列满时的丢弃计数,worker 定期取出打日志。
+_dropped = 0
+
+
+def get_queue() -> "asyncio.Queue[dict]":
+    """返回全局有界事件队列(懒创建)。测试可 monkeypatch 模块级 _queue 换成小队列。"""
+    global _queue
+    if _queue is None:
+        _queue = asyncio.Queue(maxsize=settings.OBSERVE_QUEUE_MAX)
+    return _queue
+
+
+def take_dropped() -> int:
+    """取出并清零累计丢弃数(供 worker 打点)。"""
+    global _dropped
+    n, _dropped = _dropped, 0
+    return n
+
+
+def record_event(event: dict) -> None:
+    """非阻塞入队;队列满则丢弃当前事件并计数。永不抛异常、永不阻塞请求。"""
+    global _dropped
+    try:
+        get_queue().put_nowait(event)
+    except asyncio.QueueFull:
+        _dropped += 1
diff --git a/tests/test_observe.py b/tests/test_observe.py
index 485931e..0d49dc3 100644
--- a/tests/test_observe.py
+++ b/tests/test_observe.py
@@ -5,7 +5,10 @@ OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。
 """
 from __future__ import annotations
 
+import asyncio
+
 from app.core.config import settings
+from app.core import observe
 
 
 def test_observe_configured_requires_switch_and_creds(monkeypatch):
@@ -28,3 +31,21 @@ def test_observe_configured_requires_switch_and_creds(monkeypatch):
     monkeypatch.setattr(settings, "OBSERVE_USER", "u")
     monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
     assert settings.observe_configured is False
+
+
+def test_record_event_enqueues(monkeypatch):
+    q = asyncio.Queue(maxsize=10)
+    monkeypatch.setattr(observe, "_queue", q)
+    observe.record_event({"route": "/x"})
+    assert q.get_nowait() == {"route": "/x"}
+
+
+def test_record_event_drops_when_full(monkeypatch):
+    q = asyncio.Queue(maxsize=1)
+    monkeypatch.setattr(observe, "_queue", q)
+    monkeypatch.setattr(observe, "_dropped", 0)
+    observe.record_event({"n": 1})   # 占满
+    observe.record_event({"n": 2})   # 满 → 丢弃当前,不抛异常
+    assert observe.take_dropped() == 1
+    assert observe.take_dropped() == 0   # 取出后清零
+    assert q.get_nowait() == {"n": 1}    # 保留的是先到的