feat(observe): 加后台批量上报 worker(失败丢批不重试)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/core/observe_worker.py

@@ -0,0 +1,120 @@
+"""接口指标后台上报 worker:批量 drain 事件队列 → POST 到 OpenObserve。
+
+对齐 heartbeat_monitor_worker 等的 start_*/stop_* 形态。best-effort 遥测:catch 全部
+异常,上报失败直接丢批不重试。未配置观测 → start 返回 None(不启动),整套 no-op。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+import httpx
+
+from app.core.config import settings
+from app.core.observe import get_queue, take_dropped
+
+logger = logging.getLogger("shagua.observe")
+
+# 上报用的 httpx client,start 时建、stop 时关。
+_client: httpx.AsyncClient | None = None
+
+
+async def _collect_batch() -> list[dict]:
+    """等到 ≥1 条(或到 flush 间隔)后,连抽到 BATCH_MAX 条或抽空。超时且空 → 返回 []。"""
+    queue = get_queue()
+    batch: list[dict] = []
+    try:
+        first = await asyncio.wait_for(
+            queue.get(), timeout=settings.OBSERVE_FLUSH_INTERVAL_SEC
+        )
+    except TimeoutError:
+        return batch
+    batch.append(first)
+    while len(batch) < settings.OBSERVE_BATCH_MAX:
+        try:
+            batch.append(queue.get_nowait())
+        except asyncio.QueueEmpty:
+            break
+    return batch
+
+
+async def _post_batch(client: httpx.AsyncClient, batch: list[dict]) -> None:
+    """POST 一批事件到 OpenObserve 的 _json ingest 端点。非 2xx 仅告警。"""
+    url = f"/api/{settings.OBSERVE_ORG}/{settings.OBSERVE_STREAM}/_json"
+    resp = await client.post(url, json=batch)
+    if resp.status_code >= 300:
+        logger.warning(
+            "observe ingest failed status=%s body=%s",
+            resp.status_code,
+            resp.text[:200],
+        )
+
+
+async def _run_loop(client: httpx.AsyncClient) -> None:
+    try:
+        while True:
+            batch = await _collect_batch()
+            dropped = take_dropped()
+            if dropped:
+                logger.warning("observe dropped %d events (queue full)", dropped)
+            if not batch:
+                continue
+            try:
+                await _post_batch(client, batch)
+            except Exception:  # noqa: BLE001 - best-effort 遥测,失败丢批不重试、不退出
+                logger.warning(
+                    "observe post batch failed, dropped %d events",
+                    len(batch),
+                    exc_info=True,
+                )
+    except asyncio.CancelledError:
+        logger.info("observe worker stopped")
+        raise
+
+
+def start_observe_worker() -> asyncio.Task | None:
+    """启动上报 worker。未配置观测 → 返回 None(no-op)。"""
+    global _client
+    if not settings.observe_configured:
+        return None
+    _client = httpx.AsyncClient(
+        base_url=settings.OBSERVE_ENDPOINT,
+        auth=(settings.OBSERVE_USER, settings.OBSERVE_PASSWORD),
+        timeout=settings.OBSERVE_TIMEOUT_SEC,
+    )
+    logger.info(
+        "observe worker started endpoint=%s org=%s stream=%s",
+        settings.OBSERVE_ENDPOINT,
+        settings.OBSERVE_ORG,
+        settings.OBSERVE_STREAM,
+    )
+    return asyncio.create_task(_run_loop(_client), name="observe-worker")
+
+
+async def stop_observe_worker(task: asyncio.Task | None) -> None:
+    """收尾:cancel worker → best-effort 发最后一批 → 关 client。"""
+    global _client
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+    if _client is not None:
+        # worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停)。
+        try:
+            queue = get_queue()
+            final: list[dict] = []
+            while len(final) < settings.OBSERVE_BATCH_MAX:
+                try:
+                    final.append(queue.get_nowait())
+                except asyncio.QueueEmpty:
+                    break
+            if final:
+                await asyncio.wait_for(
+                    _post_batch(_client, final), timeout=settings.OBSERVE_TIMEOUT_SEC
+                )
+        except Exception:  # noqa: BLE001 - 关停期尽力而为,失败忽略
+            pass
+        await _client.aclose()
+        _client = None

tests/test_observe.py

@@ -7,11 +7,12 @@ from __future__ import annotations
 
 import asyncio
 
+import httpx
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
-from app.core import observe
+from app.core import observe, observe_worker
 from app.core.config import settings
 
 
@@ -116,3 +117,46 @@ def test_middleware_noop_when_disabled(monkeypatch):
     client = TestClient(_make_probe_app())
     client.get("/things/1")
     assert q.empty()   # 未配置观测 → 零入队
+
+
+async def test_collect_batch_drains_up_to_batch_max(monkeypatch):
+    q = asyncio.Queue(maxsize=100)
+    monkeypatch.setattr(observe, "_queue", q)
+    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.1)
+    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
+    for i in range(3):
+        q.put_nowait({"n": i})
+    batch = await observe_worker._collect_batch()
+    assert [e["n"] for e in batch] == [0, 1, 2]
+
+
+async def test_collect_batch_timeout_returns_empty(monkeypatch):
+    q = asyncio.Queue(maxsize=100)
+    monkeypatch.setattr(observe, "_queue", q)
+    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.05)
+    batch = await observe_worker._collect_batch()
+    assert batch == []
+
+
+async def test_post_batch_hits_json_ingest_url(monkeypatch):
+    monkeypatch.setattr(settings, "OBSERVE_ORG", "default")
+    monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests")
+    captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["url"] = str(request.url)
+        captured["json"] = request.content
+        return httpx.Response(200, json={"code": 200})
+
+    client = httpx.AsyncClient(
+        base_url="http://oo", transport=httpx.MockTransport(handler)
+    )
+    await observe_worker._post_batch(client, [{"route": "/x", "status": 200}])
+    await client.aclose()
+    assert captured["url"] == "http://oo/api/default/app_requests/_json"
+    assert b"/x" in captured["json"]
+
+
+def test_start_observe_worker_noop_when_not_configured(monkeypatch):
+    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
+    assert observe_worker.start_observe_worker() is None