docs: OpenObserve 接口 QPS/耗时可观测设计 spec
方案 A:轻量自研 ASGI 中间件 + 后台 worker 批量直采到本地 Docker OpenObserve。 仅 app-server(8770);默认关、opt-in;队列满丢弃、上报失败不重试、跳过 /health。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
guke
@@ -0,0 +1,236 @@
|
||||
# 接口 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 透传耗时拆分）。
|
||||
Reference in New Issue
Block a user