shaguabijia-app-server/docs/superpowers/plans/2026-07-06-openobserve-api-metrics.md

接口 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

---

文件结构

文件	职责
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：

"""接口指标可观测(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 之前插入：

    # ===== 可观测(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: 提交

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，并追加：

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（先只放队列部分）

注意:本步只放队列相关代码。中间件用到的 os/time/Match 及 _SKIP_PATHS/_UNMATCHED/_SERVICE 常量放到 Task 3 一并加入——否则本步提交时 ruff 会报 F401 未用导入。

新建 app/core/observe.py：

"""接口指标埋点:有界事件队列 + 纯 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

• Step 4: 运行,确认通过

Run: pytest tests/test_observe.py -q -k record_event Expected: PASS

• Step 5: 提交

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 区补：

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

并追加：

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 3a: 给 app/core/observe.py 补中间件用的导入与常量

把顶部 import 段从

from __future__ import annotations

import asyncio

from app.core.config import settings

改成

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")

（_queue / _dropped / get_queue / take_dropped / record_event 保持不动。）

• Step 3b: 实现中间件（追加到 app/core/observe.py 末尾）

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: 提交

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 区补：

import httpx
from app.core import observe_worker

并追加：

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：

"""接口指标后台上报 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: 提交

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 追加：

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 块之后）加：

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 之后加一行）：

    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 的完整耗时）：

# 接口指标埋点(最外层:测含 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: 提交

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：

# 本地开发用 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：

# 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 末尾追加：

# ===== 可观测(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: 提交

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(仅本改动涉及文件)+ 全量测试

说明:仓库基线有 ~558 个既有 ruff 错误、且未强制 ruff 通过。不要去清历史欠债(范围蔓延)。只要求本次新增/改动的文件零 ruff 错误。

Run: ruff check app/core/observe.py app/core/observe_worker.py tests/test_observe.py && python -m pytest -q Expected: 上述三个新文件 ruff 无错;测试里 tests/test_observe.py 全 PASS,且全量失败数不超过基线的 4 个(test_compare_proxy ×2 / test_coupon_proxy ×1 / test_invite ×1,均与本功能无关)。 额外确认我对既有文件的改动没有引入新的 ruff 错误:ruff check app/core/config.py app/main.py(数量应与基线一致,不因本改动增加)。

• Step 6: 关观测复跑一次,确认降级

把 .env 的 OBSERVE_ENABLED 改回 false，ruff check . 不涉及，直接 pytest -q Expected: 全 PASS（验证 observe 关闭时零副作用）

• Step 7: 收尾提交（如有 .env 之外的改动）

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 与文件结构表一致。