diff --git a/docs/superpowers/plans/2026-07-06-openobserve-api-metrics.md b/docs/superpowers/plans/2026-07-06-openobserve-api-metrics.md new file mode 100644 index 0000000..236c8d3 --- /dev/null +++ b/docs/superpowers/plans/2026-07-06-openobserve-api-metrics.md @@ -0,0 +1,871 @@ +# 接口 QPS + 耗时可观测（OpenObserve）实现计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 给 app-server 每个接口采集 QPS + 耗时 + 错误率，经轻量 ASGI 中间件 + 后台 worker 批量直采到本地 Docker 的 OpenObserve。 + +**Architecture:** 纯 ASGI 中间件测每请求耗时/抓路由模板+状态码 → 非阻塞入有界队列（满则丢、绝不阻塞）→ 后台 asyncio worker 批量 POST 到 OpenObserve `_json` ingest 端点。请求路径零 I/O；未配置观测则整套 no-op；上报失败丢批不重试。 + +**Tech Stack:** FastAPI / Starlette ASGI 中间件、`asyncio.Queue`、`httpx.AsyncClient`（已有依赖）、pydantic-settings、OpenObserve（Docker）。 + +参考 spec：[docs/superpowers/specs/2026-07-06-openobserve-api-metrics-design.md](2026-07-06-openobserve-api-metrics-design.md) + +--- + +## 文件结构 + +| 文件 | 职责 | +|---|---| +| `app/core/config.py`（改） | 新增 `OBSERVE_*` 配置 + `observe_configured` 门槛属性 | +| `app/core/observe.py`（新） | 有界事件队列 + `record_event` + 路由模板解析 + `RequestMetricsMiddleware` | +| `app/core/observe_worker.py`（新） | 后台批量上报 worker：`_collect_batch` / `_post_batch` / `start_*` / `stop_*` | +| `app/main.py`（改） | 挂中间件（最外层）+ lifespan 启停 worker | +| `.env.example`（改） | 新增 `OBSERVE_*` 注释段 | +| `deploy/openobserve/docker-compose.yml`（新） | 本地 OpenObserve 容器 | +| `deploy/openobserve/README.md`（新） | 部署步骤 + 查询/仪表盘 SQL | +| `tests/test_observe.py`（新） | 配置门槛 / 队列 / 中间件 / worker 单测 | + +**关键接口契约（跨任务一致，勿改名）：** +- `app.core.observe.get_queue() -> asyncio.Queue[dict]` +- `app.core.observe.record_event(event: dict) -> None` +- `app.core.observe.take_dropped() -> int` +- `app.core.observe.RequestMetricsMiddleware`（ASGI class，`__init__(self, app)`） +- 事件字段：`_timestamp`(µs int) / `service` / `env` / `method` / `route` / `status` / `duration_ms`(float) +- `app.core.observe_worker.start_observe_worker() -> asyncio.Task | None` +- `app.core.observe_worker.stop_observe_worker(task) -> None` +- `settings.observe_configured -> bool` + +--- + +## Task 1: 配置项 `OBSERVE_*` + `observe_configured` + +**Files:** +- Modify: `app/core/config.py`（在 `cors_origins_list` property 之后、`is_prod` property 之前插入） +- Test: `tests/test_observe.py`（新建） + +- [ ] **Step 1: 写失败测试** + +新建 `tests/test_observe.py`： + +```python +"""接口指标可观测(observe)单测:配置门槛 / 队列 / 中间件 / worker。 + +沿用仓库约定:TestClient + monkeypatch,绝不打真网络。observe 默认关(conftest 未设 +OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。 +""" +from __future__ import annotations + +from app.core.config import settings + + +def test_observe_configured_requires_switch_and_creds(monkeypatch): + # 开关开 + endpoint(默认 localhost)+ user + password 齐全 → True + monkeypatch.setattr(settings, "OBSERVE_ENABLED", True) + monkeypatch.setattr(settings, "OBSERVE_USER", "u") + monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p") + assert settings.observe_configured is True + + # 缺密码 → False + monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "") + assert settings.observe_configured is False + + # 开关关 → False(即便凭证齐全) + monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p") + monkeypatch.setattr(settings, "OBSERVE_ENABLED", False) + assert settings.observe_configured is False +``` + +- [ ] **Step 2: 运行,确认失败** + +Run: `pytest tests/test_observe.py::test_observe_configured_requires_switch_and_creds -q` +Expected: FAIL —— `AttributeError`（`settings` 无 `OBSERVE_ENABLED` / 无 `observe_configured`） + +- [ ] **Step 3: 实现配置** + +在 `app/core/config.py` 的 `cors_origins_list` property 之后、`is_prod` property 之前插入： + +```python + # ===== 可观测(OpenObserve 接口指标)===== + # 采集每个接口的 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker)。 + # 默认关(prod 安全):未开启 → 中间件透传、worker 不启动,整套 no-op。 + # 开启需 ENABLED=true 且 ENDPOINT/USER/PASSWORD 齐全(见 observe_configured)。 + OBSERVE_ENABLED: bool = False + OBSERVE_ENDPOINT: str = "http://localhost:5080" # OpenObserve base URL + OBSERVE_ORG: str = "default" # 组织名 + OBSERVE_STREAM: str = "app_requests" # stream 名(首次上报自动建) + OBSERVE_USER: str = "" # Basic auth 邮箱 + OBSERVE_PASSWORD: str = "" # Basic auth 密码/token + OBSERVE_FLUSH_INTERVAL_SEC: float = 5.0 # worker 最长攒批间隔 + OBSERVE_BATCH_MAX: int = 200 # 单批最大事件数 + OBSERVE_QUEUE_MAX: int = 10000 # 有界队列上限,满则丢 + OBSERVE_TIMEOUT_SEC: float = 5.0 # 上报 HTTP 超时 + + @property + def observe_configured(self) -> bool: + """观测上报可用 = 总开关开 且 endpoint/账号/密码齐全(缺则整套 no-op)。""" + return bool( + self.OBSERVE_ENABLED + and self.OBSERVE_ENDPOINT + and self.OBSERVE_USER + and self.OBSERVE_PASSWORD + ) +``` + +- [ ] **Step 4: 运行,确认通过** + +Run: `pytest tests/test_observe.py::test_observe_configured_requires_switch_and_creds -q` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add app/core/config.py tests/test_observe.py +git commit -m "feat(observe): 加 OBSERVE_* 配置与 observe_configured 门槛" +``` + +--- + +## Task 2: 事件队列 + `record_event` + `take_dropped` + +**Files:** +- Create: `app/core/observe.py` +- Test: `tests/test_observe.py`（追加） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_observe.py` 顶部 import 区补 `import asyncio` 和 `from app.core import observe`，并追加： + +```python +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} # 保留的是先到的 +``` + +- [ ] **Step 2: 运行,确认失败** + +Run: `pytest tests/test_observe.py -q -k record_event` +Expected: FAIL —— `ModuleNotFoundError: app.core.observe` 或无 `record_event` + +- [ ] **Step 3: 实现 `app/core/observe.py`（先只放队列部分）** + +新建 `app/core/observe.py`： + +```python +"""接口指标埋点:有界事件队列 + 纯 ASGI 中间件。 + +每个 HTTP 请求测总耗时、抓路由模板 + 状态码,非阻塞塞进有界队列;由 observe_worker +后台批量上报到 OpenObserve。请求路径上无任何 I/O。未配置观测时中间件直接透传。 +""" +from __future__ import annotations + +import asyncio +import logging +import os +import time + +from starlette.routing import Match + +from app.core.config import settings + +logger = logging.getLogger("shagua.observe") + +# 不采集的路径(纯噪音):健康检查。 +_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 +# 队列满时的丢弃计数,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 +``` + +- [ ] **Step 4: 运行,确认通过** + +Run: `pytest tests/test_observe.py -q -k record_event` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add app/core/observe.py tests/test_observe.py +git commit -m "feat(observe): 加有界事件队列与 record_event(满则丢)" +``` + +--- + +## Task 3: `RequestMetricsMiddleware`（路由模板 + 状态码 + 耗时） + +**Files:** +- Modify: `app/core/observe.py`（追加 `_resolve_route` 和中间件 class） +- Test: `tests/test_observe.py`（追加） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_observe.py` 顶部 import 区补： + +```python +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +``` + +并追加： + +```python +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() # 未配置观测 → 零入队 +``` + +- [ ] **Step 2: 运行,确认失败** + +Run: `pytest tests/test_observe.py -q -k middleware` +Expected: FAIL —— `AttributeError: module 'app.core.observe' has no attribute 'RequestMetricsMiddleware'` + +- [ ] **Step 3: 实现中间件（追加到 `app/core/observe.py` 末尾）** + +```python +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), + }) +``` + +- [ ] **Step 4: 运行,确认通过** + +Run: `pytest tests/test_observe.py -q -k middleware` +Expected: PASS（4 个中间件用例全过） + +> 若 `test_middleware_records_route_template` 拿到的是 `/things/42` 而非模板，说明该 Starlette 版本未在 `scope["route"]` 写模板——此时 `_resolve_route` 的手动匹配兜底应已生效并返回模板；若仍不对，检查兜底分支是否被 import 顺序影响。 + +- [ ] **Step 5: 提交** + +```bash +git add app/core/observe.py tests/test_observe.py +git commit -m "feat(observe): 加 RequestMetricsMiddleware(路由模板+状态码+耗时)" +``` + +--- + +## Task 4: 后台上报 worker + +**Files:** +- Create: `app/core/observe_worker.py` +- Test: `tests/test_observe.py`（追加） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_observe.py` 顶部 import 区补： + +```python +import httpx +from app.core import observe_worker +``` + +并追加： + +```python +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 +``` + +- [ ] **Step 2: 运行,确认失败** + +Run: `pytest tests/test_observe.py -q -k "collect_batch or post_batch or start_observe"` +Expected: FAIL —— `ModuleNotFoundError: app.core.observe_worker` + +- [ ] **Step 3: 实现 `app/core/observe_worker.py`** + +新建 `app/core/observe_worker.py`： + +```python +"""接口指标后台上报 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 asyncio.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 +``` + +- [ ] **Step 4: 运行,确认通过** + +Run: `pytest tests/test_observe.py -q -k "collect_batch or post_batch or start_observe"` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add app/core/observe_worker.py tests/test_observe.py +git commit -m "feat(observe): 加后台批量上报 worker(失败丢批不重试)" +``` + +--- + +## Task 5: 接线到 `app/main.py`（挂中间件 + lifespan 启停） + +**Files:** +- Modify: `app/main.py`（import 区、lifespan、CORS 之后） +- Test: `tests/test_observe.py`（追加） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_observe.py` 追加： + +```python +def test_app_has_metrics_middleware(): + from app.main import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "RequestMetricsMiddleware" in names +``` + +- [ ] **Step 2: 运行,确认失败** + +Run: `pytest tests/test_observe.py::test_app_has_metrics_middleware -q` +Expected: FAIL —— 断言失败（中间件尚未挂载） + +- [ ] **Step 3: 实现接线** + +3a. 在 `app/main.py` import 区（`withdraw_reconcile_worker` import 块之后）加： + +```python +from app.core.observe import RequestMetricsMiddleware +from app.core.observe_worker import ( + start_observe_worker, + stop_observe_worker, +) +``` + +3b. lifespan 里加启停（现有 `daily_exchange_task = start_daily_exchange_worker()` 之后、`try:` 之前加一行；`finally` 里在 `stop_daily_exchange_worker` 之后加一行）： + +```python + daily_exchange_task = start_daily_exchange_worker() + observe_task = start_observe_worker() + try: + yield + finally: + await stop_heartbeat_monitor(heartbeat_task) + await stop_withdraw_reconcile_worker(reconcile_task) + await stop_daily_exchange_worker(daily_exchange_task) + await stop_observe_worker(observe_task) + await aclose_pricebot_client() + logger.info("shutting down") +``` + +3c. 挂中间件——在 CORS 的 `if settings.cors_origins_list:` 整块之后加（使其成为最外层，测到含 CORS 的完整耗时）： + +```python +# 接口指标埋点(最外层:测含 CORS 的完整耗时)。未配置观测时中间件自 no-op。 +app.add_middleware(RequestMetricsMiddleware) +``` + +- [ ] **Step 4: 运行,确认通过** + +Run: `pytest tests/test_observe.py::test_app_has_metrics_middleware -q` +Expected: PASS + +- [ ] **Step 5: 跑整套 observe 测试 + 全量回归,确认无破坏** + +Run: `pytest tests/test_observe.py -q && pytest -q` +Expected: 全 PASS（现有用例不受影响：conftest 未设 `OBSERVE_*` → 观测关 → worker no-op、中间件透传） + +- [ ] **Step 6: 提交** + +```bash +git add app/main.py tests/test_observe.py +git commit -m "feat(observe): main.py 挂中间件 + lifespan 启停上报 worker" +``` + +--- + +## Task 6: OpenObserve 本地部署（compose + README + .env.example） + +**Files:** +- Create: `deploy/openobserve/docker-compose.yml` +- Create: `deploy/openobserve/README.md` +- Modify: `.env.example`（追加 `OBSERVE_*` 段） + +- [ ] **Step 1: 写 docker-compose** + +新建 `deploy/openobserve/docker-compose.yml`： + +```yaml +# 本地开发用 OpenObserve(单容器 = local 模式)。用于接收 app-server 的接口指标。 +# 启动: cd deploy/openobserve && docker compose up -d +# Web UI: http://localhost:5080 (账号见下方 env) +services: + openobserve: + image: public.ecr.aws/zinclabs/openobserve:latest + container_name: openobserve + ports: + - "5080:5080" + environment: + ZO_ROOT_USER_EMAIL: "admin@shaguabijia.local" + ZO_ROOT_USER_PASSWORD: "Complexpass#123" + ZO_DATA_DIR: "/data" + volumes: + - ./data:/data + restart: unless-stopped +``` + +- [ ] **Step 2: 写 README** + +新建 `deploy/openobserve/README.md`： + +````markdown +# OpenObserve 本地部署（接口 QPS / 耗时可观测） + +app-server 通过中间件采集每个接口的 QPS + 耗时 + 错误率，批量上报到这里。 + +## 启动 + +```bash +cd deploy/openobserve +docker compose up -d +``` + +- Web UI：http://localhost:5080 +- 登录：`admin@shaguabijia.local` / `Complexpass#123`（见 `docker-compose.yml`） +- 数据落 `deploy/openobserve/data/`（已挂卷持久化；`data/` 建议 gitignore） + +## 让 app-server 上报 + +在项目根的 `.env` 打开观测（`OBSERVE_*`，账号密码与 compose 里 root 一致）： + +```dotenv +OBSERVE_ENABLED=true +OBSERVE_ENDPOINT=http://localhost:5080 +OBSERVE_ORG=default +OBSERVE_STREAM=app_requests +OBSERVE_USER=admin@shaguabijia.local +OBSERVE_PASSWORD=Complexpass#123 +``` + +重启 app-server，随便打几个接口。stream `app_requests` **首次上报自动创建**， +在 UI 的 Logs → 选 `app_requests` 就能看到逐条请求事件。 + +## 查询（Logs 页 SQL，或建 Dashboard 面板） + +各接口 QPS（1 分钟分桶，面板里再除 60 得每秒）： + +```sql +SELECT route, histogram(_timestamp, '1 minute') AS ts, count(*) AS cnt +FROM app_requests GROUP BY route, ts ORDER BY ts +``` + +各接口 P95 耗时（毫秒）： + +```sql +SELECT route, approx_percentile_cont(duration_ms, 0.95) AS p95_ms +FROM app_requests GROUP BY route ORDER BY p95_ms DESC +``` + +各接口错误率（5xx 占比）： + +```sql +SELECT route, + count(*) FILTER (WHERE status >= 500) * 100.0 / count(*) AS err_pct +FROM app_requests GROUP BY route ORDER BY err_pct DESC +``` + +## 停止 / 清数据 + +```bash +docker compose down # 停止(保留数据) +docker compose down -v && rm -rf data # 停止并清空数据 +``` + +> 生产部署（持久化规格、独立 ingest 账号、鉴权收紧）见 spec 第 9 节，本期不做。 +```` + +- [ ] **Step 3: 追加 `.env.example`** + +在 `.env.example` 末尾追加： + +```dotenv + +# ===== 可观测(OpenObserve 接口指标)===== +# 采集每个接口 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker,见 deploy/openobserve/)。 +# 默认关;开启需 ENABLED=true 且填 USER/PASSWORD(与 docker-compose 里 root 账号一致)。 +OBSERVE_ENABLED=false +OBSERVE_ENDPOINT=http://localhost:5080 +OBSERVE_ORG=default +OBSERVE_STREAM=app_requests +OBSERVE_USER=admin@shaguabijia.local +OBSERVE_PASSWORD=Complexpass#123 +# 进阶(一般不用改):攒批间隔秒 / 单批最大条数 / 有界队列上限(满则丢) / 上报超时秒 +OBSERVE_FLUSH_INTERVAL_SEC=5 +OBSERVE_BATCH_MAX=200 +OBSERVE_QUEUE_MAX=10000 +OBSERVE_TIMEOUT_SEC=5 +``` + +- [ ] **Step 4: 校验 compose 语法（不需真拉镜像）** + +Run: `docker compose -f deploy/openobserve/docker-compose.yml config` +Expected: 打印规整后的配置、无报错（若本机无 docker，可跳过，标注为手动验证项） + +- [ ] **Step 5: 提交** + +```bash +git add deploy/openobserve/docker-compose.yml deploy/openobserve/README.md .env.example +git commit -m "feat(observe): 加 OpenObserve 本地 compose + README + .env.example" +``` + +--- + +## Task 7: 端到端手动验证 + 全量 lint/test 收尾 + +**Files:** 无（验证 + 收尾） + +- [ ] **Step 1: 起 OpenObserve** + +Run: `cd deploy/openobserve && docker compose up -d` +Expected: 容器起来，浏览器打开 http://localhost:5080 能登录 + +- [ ] **Step 2: 本地开观测起 app-server** + +在根 `.env` 设 `OBSERVE_ENABLED=true` + `OBSERVE_USER/PASSWORD`（同 compose），然后： + +Run: `./run.sh`（Windows 用 `python -m uvicorn app.main:app --port 8770 --reload --reload-dir app`） +Expected: 启动日志出现 `observe worker started endpoint=http://localhost:5080 ...` + +- [ ] **Step 3: 打几个接口产生数据** + +Run: `curl http://localhost:8770/health && curl http://localhost:8770/things-does-not-exist -i`（或正常业务接口若干） +Expected: 稍等 ≤5s（flush 间隔），OpenObserve UI 的 Logs → `app_requests` 出现事件；`/health` 不应出现；不存在的路径 route 为 `__unmatched__` + +- [ ] **Step 4: 验证三条查询** + +在 OpenObserve UI 分别粘贴 README 里的 QPS / P95 / 错误率 SQL，确认能出数。 + +- [ ] **Step 5: 全量 lint + 测试** + +Run: `ruff check . && pytest -q` +Expected: ruff 无错；测试全 PASS + +- [ ] **Step 6: 关观测复跑一次,确认降级** + +把 `.env` 的 `OBSERVE_ENABLED` 改回 `false`，`ruff check .` 不涉及，直接 `pytest -q` +Expected: 全 PASS（验证 observe 关闭时零副作用） + +- [ ] **Step 7: 收尾提交（如有 .env 之外的改动）** + +```bash +git add -A +git commit -m "chore(observe): 端到端验证与收尾" --allow-empty +``` + +> `.env` 不入 git（已 gitignore）；本任务只验证，不提交 `.env`。 + +--- + +## Self-Review（写完计划后自查） + +- **Spec 覆盖**：Docker 部署→Task 6/7；事件 schema→Task 3（`record_event` 事件字段）；中间件→Task 3；worker→Task 4；配置→Task 1；main 接线→Task 5；查询/仪表盘→Task 6 README；测试→Task 1-5；决策(a)队列满丢→Task 2；(b)失败不重试→Task 4；(c)跳过 /health→Task 3。全覆盖。 +- **占位符**：无 TBD/TODO；每个代码步骤含完整代码。 +- **类型/命名一致**：`get_queue` / `record_event` / `take_dropped` / `RequestMetricsMiddleware` / `start_observe_worker` / `stop_observe_worker` / `observe_configured` / 事件字段名，跨 Task 1-5 与文件结构表一致。