# 接口 QPS + 耗时可观测（OpenObserve）设计

- **日期**：2026-07-06
- **状态**：已评审通过，待写实现计划
- **范围**：仅 app-server（8770）；admin（8771）暂不接入
- **方案**：A —— 轻量自研 ASGI 中间件 + 后台 worker 批量直采到 OpenObserve

## 1. 背景与目标

app-server 目前除 CORS 外无任何中间件，也无接口级可观测。需要按**每个接口**采集：

- **QPS**（每秒请求数，可按接口/时间分桶）
- **耗时**（P50/P95/P99 等分位）

顺带低成本拿到**错误率**（`status >= 500` 占比）。落地目标是：本地 Docker 跑一个 OpenObserve 实例接收数据，服务侧加埋点上报，在 OpenObserve 仪表盘上看各接口 QPS + 耗时。

### 非目标（YAGNI）

- 不做分布式 trace / span 关联（只要接口聚合指标）。
- 不引入 OpenTelemetry / Prometheus 客户端等重依赖。
- 不采集请求体 / query / 用户身份等，任何 PII 都不进上报。
- admin（8771）本期不接（中间件写成可复用，未来一行挂载即可）。
- 上报失败不做持久化重试 / 落盘补偿（best-effort）。

## 2. 方案选型

对比过三条路（详见评审记录）：

- **A 轻量自研中间件 + JSON 直采**（选中）：零新依赖（`httpx` 已在依赖里），完全贴合本仓库「后台 worker + JSON 事件 + `*_configured` 优雅降级」的既有习惯，恰好满足「每接口 QPS + 耗时 + 错误率」并保留原始事件下钻能力。
- B OpenTelemetry 自动埋点 + OTLP：行业标准、顺带 trace，但多 5–6 个依赖、概念多、数据量/成本高于需求，与精简代码库风格相悖。
- C Prometheus 进程内聚合 + remote_write/抓取：数据量最小，但 remote_write 编码复杂或需额外抓取进程，丢失单请求下钻，最不贴合 OpenObserve 的 log-first 强项。

**结论：A。**

## 3. 架构与数据流

```
每个 HTTP 请求
  → RequestMetricsMiddleware（最外层：测总耗时 / 抓路由模板 + 状态码）
  → record_event()  非阻塞入队（有界队列，满则丢最旧，绝不阻塞、绝不 OOM）
  → observe_worker（后台 asyncio.Task，随 lifespan 启停）批量 drain
  → httpx POST  {ENDPOINT}/api/{ORG}/{STREAM}/_json  → OpenObserve
                                                        → 仪表盘 SQL 聚合出 QPS / 分位耗时 / 错误率
```

**核心不变量**：

1. 请求路径上只做「测时 + 构建一个小 dict + `put_nowait`」，**无任何网络/磁盘 I/O**。
2. 所有上报 I/O 在后台 worker；worker 捕获全部异常，绝不让埋点影响请求。
3. 未配置观测（`observe_configured=False`）→ 中间件透传、worker 不启动，整套 no-op。
4. OpenObserve 不可用 → 队列填满后丢弃事件 + 限流告警，业务零影响。

## 4. 组件设计

### 4.1 OpenObserve 本地部署 —— `deploy/openobserve/docker-compose.yml`（新增）

```yaml
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
```

- `docker compose up -d` 启动；Web UI `http://localhost:5080`，用上面邮箱/密码登录。
- 单容器 = local 模式，数据落 `./data`（已挂卷持久化）。
- **stream 首次上报自动创建**，无需预建 `app_requests`。
- 上报鉴权：HTTP Basic auth（`email:password`），本地直接用 root 账号；生产应另建仅具 ingest 权限的用户/服务账号（本期不涉及）。

### 4.2 事件 schema（一请求一行 JSON）

```json
{
  "_timestamp": 1720000000000000,        // 微秒(µs)整数，请求完成时刻。OpenObserve 默认时间列 _timestamp 以微秒计
  "service": "app-server",               // 取 LOG_SERVICE_NAME / 固定值
  "env": "dev",                          // settings.APP_ENV
  "method": "POST",
  "route": "/api/v1/coupon/step",        // 路由模板（非实际 path）
  "status": 200,
  "duration_ms": 42.7                    // float 毫秒
}
```

- **只存路由模板**（如 `/c/{code}`、`/media` 静态归一），避免 path 参数把维度打爆。
- 未匹配路由（404 / 扫描器）归一到常量 `__unmatched__`。
- 只采 method / route / status / duration —— 无 body、无 query、无 PII。

### 4.3 埋点中间件 —— `app/core/observe.py`（新增）

**纯 ASGI 中间件**（比 `BaseHTTPMiddleware` 开销低；能可靠读到路由与最终状态码；scope 按引用透传，内层 router 的 `scope["route"]` 外层可见）。

职责：

1. 非 `http` 请求、或 `not settings.observe_configured` → 直接透传，不测。
2. `perf_counter()` 记起点；包一层 `send` 抓 `http.response.start` 的 `status`（默认兜底 500，覆盖下游抛异常未产出 response 的情况）。
3. `finally` 里算 `duration_ms`，从 `scope` 取路由模板（见下），构建事件，调 `record_event()`。
4. 跳过路径集合 `_SKIP_PATHS = {"/health"}`（纯噪音）。

**路由模板解析（跨 Starlette 版本稳健）**：

```python
route = scope.get("route")
template = getattr(route, "path", None)
if template is None:            # 未匹配 / 老版本未写 scope["route"]
    template = "__unmatched__"
```

（若实测某 Starlette 版本不写 `scope["route"]`，回退用 `request.app.router.routes` 逐个 `route.matches(scope)==Match.FULL` 找模板；实现时以实际版本为准，优先 `scope["route"]`。）

**入队（`record_event`）**：模块级 `asyncio.Queue(maxsize=OBSERVE_QUEUE_MAX)`。用 `put_nowait`，`QueueFull` 则丢弃并累加一个 `_dropped` 计数（每累计 N 条限流打一条 WARNING）。**永不 `await put()`、永不阻塞请求**。

> 决策(a)：队列满 → **丢弃**（不阻塞请求）。

### 4.4 上报 worker —— `app/core/observe_worker.py`（新增）

对齐现有 `heartbeat_monitor_worker.py` / `daily_exchange_worker.py` / `withdraw_reconcile_worker.py` 的 `start_*` / `stop_*` 形态。

- `start_observe_worker() -> asyncio.Task | None`
  - `not settings.observe_configured` → 返回 `None`（no-op）。
  - 否则建专用 `httpx.AsyncClient`（`base_url=ENDPOINT`，`auth=(USER, PASSWORD)`，`timeout=OBSERVE_TIMEOUT_SEC`），起 `_run_loop` task。
- `_run_loop()`：循环
  1. `_collect_batch()`：`await asyncio.wait_for(queue.get(), timeout=FLUSH_INTERVAL)` 拿到首条（超时且空 → 返回空，continue）；再 `get_nowait()` 连抽到 `BATCH_MAX` 条或抽空。
  2. `POST /api/{ORG}/{STREAM}/_json`，body 为事件数组。
  3. **catch 所有异常**：失败限流打 WARNING，**直接丢弃该批，不重试**。
- `stop_observe_worker(task)`：best-effort 收尾 flush（短超时）→ `task.cancel()` → `await`（吞 `CancelledError`）→ 关 client。

> 决策(b)：上报失败 → **直接丢弃，不重试**（best-effort 遥测）。

### 4.5 配置 —— `app/core/config.py`（改）

新增一段 `# ===== 可观测（OpenObserve 接口指标）=====`，默认全关（prod 安全）：

| 配置 | 默认 | 说明 |
|---|---|---|
| `OBSERVE_ENABLED` | `False` | 总开关；默认关，opt-in |
| `OBSERVE_ENDPOINT` | `http://localhost:5080` | OpenObserve base URL |
| `OBSERVE_ORG` | `default` | 组织名 |
| `OBSERVE_STREAM` | `app_requests` | stream 名 |
| `OBSERVE_USER` | `""` | Basic auth 邮箱 |
| `OBSERVE_PASSWORD` | `""` | Basic auth 密码/token |
| `OBSERVE_FLUSH_INTERVAL_SEC` | `5.0` | worker 最长攒批间隔 |
| `OBSERVE_BATCH_MAX` | `200` | 单批最大事件数 |
| `OBSERVE_QUEUE_MAX` | `10000` | 有界队列上限，满则丢 |
| `OBSERVE_TIMEOUT_SEC` | `5.0` | 上报 HTTP 超时 |

```python
@property
def observe_configured(self) -> bool:
    return bool(self.OBSERVE_ENABLED and self.OBSERVE_ENDPOINT
                and self.OBSERVE_USER and self.OBSERVE_PASSWORD)
```

`.env.example` 同步补一段带注释的 `OBSERVE_*`（沿用该文件重注释风格），`OBSERVE_ENABLED=false`。

### 4.6 接线 —— `app/main.py`（改）

- import `RequestMetricsMiddleware`、`start_observe_worker` / `stop_observe_worker`。
- `app.add_middleware(RequestMetricsMiddleware)`：放在 CORS `add_middleware` **之后** → 成为最外层，测到含 CORS 的完整耗时。无条件挂载（内部自 no-op）。
- `lifespan`：启动 `observe_task = start_observe_worker()`；`finally` 里 `await stop_observe_worker(observe_task)`，与现有 worker 并列。

### 4.7 OpenObserve 查询 / 仪表盘 —— `deploy/openobserve/README.md`（新增）

含：compose 启停、登录、stream 自动创建说明、`.env` 接线，以及可直接粘的示例 SQL：

- **各接口 QPS**（1 分钟分桶）：
  ```sql
  SELECT route, histogram(_timestamp, '1 minute') AS ts, count(*) AS cnt
  FROM app_requests GROUP BY route, ts ORDER BY ts
  ```
  （面板按 `cnt/60` 展示每秒；或用 OpenObserve 图表的 rate 能力。）
- **各接口 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
  ```
- **各接口错误率**：
  ```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
  ```

## 5. 关键设计决策汇总

- **(a) 队列满 → 丢弃**（不阻塞请求）：遥测让路于业务可用性。
- **(b) 上报失败 → 不重试**：best-effort；避免 poison batch 堆积与队列无限增长。
- **(c) 跳过 `/health`**：健康检查是纯噪音，硬编码在 `_SKIP_PATHS`。
- **只存路由模板 + `__unmatched__`**：防维度爆炸。
- **默认 OFF、opt-in**：prod 安全默认；开启后仍全异步 + 有界。
- **纯 ASGI 中间件 + `perf_counter`**：请求路径开销微秒级，无 I/O。

## 6. 安全 / 性能保证

- 请求路径新增开销 ≈ 一次 `perf_counter` 差 + 一个小 dict + 一次 `put_nowait`（微秒级），无锁竞争的显著热点。
- 失败隔离：入队丢弃 + worker 全异常捕获；OpenObserve 宕机不影响任何请求。
- 有界内存：队列 `maxsize` 封顶，最坏丢事件不涨内存。
- 无 PII：仅 method / route / status / duration。

## 7. 测试策略 —— `tests/test_observe.py`（新增）

沿用仓库约定（`TestClient` + `monkeypatch`，绝不打真网络；`conftest` 在 import 前设 env）：

1. 埋点入队字段正确：模板路由、`status`、`duration_ms > 0`。
2. 参数化路由 → 取到**模板**而非实际 path。
3. 未匹配路径（404）→ `route == "__unmatched__"`。
4. `OBSERVE_ENABLED=false` → 零入队、零 HTTP（现有测试不受影响）。
5. 队列满 → `record_event` 不抛异常（走丢弃分支）。
6. worker 批量 POST 的 URL / payload 正确（monkeypatch httpx client / `_post`，不打网络）。
7. `/health` 被跳过 → 不入队。

> `settings` 是 `lru_cache` 单例；需要开启观测的用例通过 monkeypatch `settings` 属性或直接调 `record_event` / 中间件并 patch `observe_configured` 实现，避免全局 env 改动波及他用例。

## 8. 文件清单

| 文件 | 动作 |
|---|---|
| `deploy/openobserve/docker-compose.yml` | 新增（OpenObserve 容器）|
| `deploy/openobserve/README.md` | 新增（部署步骤 + 查询/仪表盘）|
| `app/core/observe.py` | 新增（中间件 + 有界队列 + `record_event` + 路由解析）|
| `app/core/observe_worker.py` | 新增（后台批量上报 worker）|
| `app/core/config.py` | 改（`OBSERVE_*` + `observe_configured`）|
| `app/main.py` | 改（挂中间件 + lifespan 启停 worker）|
| `.env.example` | 改（新增 `OBSERVE_*` 注释段）|
| `tests/test_observe.py` | 新增 |

## 9. 未来工作（本期不做）

- admin（8771）接入同一套中间件（`service` 字段区分）。
- 生产部署 OpenObserve（持久化、独立 ingest 账号、资源规格、鉴权收紧）。
- 上报字段扩展（如按 user/设备维度、上游 pricebot 透传耗时拆分）。
