Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/inactivity-reset
# Conflicts: # docs/superpowers/specs/2026-07-16-inactivity-reset-design.md
This commit is contained in:
guke
@@ -115,6 +115,7 @@
|
||||
| 40a | `GET /api/v1/platform/flags` | 无 | [详情](./platform/platform-flags.md)（客户端运营 feature flag,比价/领券期广告开关等,拉取后缓存） |
|
||||
| 40b | `GET /api/v1/platform/ad-config` | 无 | [详情](./platform/platform-ad-config.md)（客户端拉广告配置:穿山甲 app_id+各位ID+各场景开关;不含验签密钥） |
|
||||
| 40c | `GET /api/v1/platform/app-version` | 无 | [详情](./platform/platform-app-version.md)（最新 App 版本,OTA 检查更新;与本机 versionCode 比） |
|
||||
| 40d | `GET /api/v1/platform/huawei-review` | 无 | [详情](./platform/platform-huawei-review.md)（华为审核开关:快速设置权限步能否被用户关闭;仅华为 ROM 客户端拉） |
|
||||
| **微信支付回调**（前缀 `/api/v1/wxpay`） |||
|
||||
| W1 | `POST /api/v1/wxpay/transfer-auth-notify` | 无 | 免确认收款授权结果通知（一期 stub:仅应答 200 不验签不改账,授权状态靠主动查询兜底)(无单独文档) |
|
||||
| **CPS 群发短链落地**（**无前缀**,挂域名根;公网不鉴权） |||
|
||||
@@ -153,6 +154,7 @@
|
||||
| A12 | `GET /admin/api/ad-revenue-report` | admin | [详情](./admin/ad/admin-ad-revenue-report.md)(广告收益报表:分页/场景/`app_env` 筛 + **DAU/ARPU** #120;真实收益侧接穿山甲日表 #92) |
|
||||
| A13 | `GET / PATCH /admin/api/ad-config` | operator/finance | 广告配置(穿山甲 ID/验签密钥/各场景开关;C 端只读版见 40b)(无单独文档,见 `app/admin/routers/ad_config.py`) |
|
||||
| A14 | `GET /admin/api/config`、`PATCH /config/{key}` | operator/finance | 运营可配置项([app_config](../database/app_config.md):奖励常量/提现地板价等;#117 修系统配置下发)(无单独文档,见 `app/admin/routers/config.py`) |
|
||||
| A16 | `GET / PATCH /admin/api/huawei-review` | operator/tech | 华为审核开关(快速设置权限步能否被用户关闭,落 `app_config.huawei_review`;C 端只读版见 40d)(无单独文档,见 `app/admin/routers/huawei_review.py`) |
|
||||
| **A·管理员与角色**(super_admin):`GET`/`POST` `/admins`、`PATCH`/`DELETE` `/admins/{id}`(#126 删除+`pages_override`)、`GET`/`POST` `/roles`、`GET /roles/catalog`、`PATCH`/`DELETE` `/roles/{id}`(#117/#126 自定义角色) ||| [列表](./admin/admins/admin-admins-list.md) / [建](./admin/admins/admin-admin-create.md) / [改+删](./admin/admins/admin-admin-update.md) / [角色](./admin/admin-roles.md) |
|
||||
| A15 | `GET /admin/api/audit-logs` | admin | [详情](./admin/admin-audit-logs.md) |
|
||||
| **A·CPS 运营台**:群/活动 CRUD、`POST /referral-links`、`POST /orders/reconcile`(美团+京东 #90)、`GET /orders`、`/stats`、群 `timeseries`/`daily`/`wx-users`/`day-users`(#79) ||| [详情](./admin/admin-cps.md) |
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
# GET /api/v1/platform/huawei-review — 华为审核开关
|
||||
|
||||
> 所属:Platform 组（前缀 `/api/v1/platform`） | 鉴权:无 | [← 返回 API 索引](../README.md)
|
||||
|
||||
华为应用市场审核要求:新手引导的**「快速设置」权限步必须可被用户关闭**（引导视频页不在要求内）。本端点把运营后台配的开关下发给客户端,决定该步左上角是否出现退出按钮。
|
||||
|
||||
**不鉴权**:引导页在登录之前就展示,此时客户端必然没有 token。
|
||||
|
||||
值来自 `app_config` 表的 `huawei_review` 行（admin 页 `GET / PATCH /admin/api/huawei-review` 可改），空库回退 `default`。
|
||||
|
||||
## 入参
|
||||
|
||||
无。
|
||||
|
||||
## 出参
|
||||
|
||||
响应 `200`:`HuaweiReviewOut`
|
||||
|
||||
| 字段 | 类型 | 说明 |
|
||||
|---|---|---|
|
||||
| `mode` | string | `default`（强制展示,不可关闭 = 上线至今的现状）/ `review`（可关闭,过审用）。仅供排查时看后台切成了哪态 |
|
||||
| `onboarding_closable` | bool | 快速设置权限步是否允许用户退出（= `mode == "review"`）。**客户端只读这一个字段决策** |
|
||||
|
||||
Mock 出参:
|
||||
```json
|
||||
{
|
||||
"mode": "review",
|
||||
"onboarding_closable": true
|
||||
}
|
||||
```
|
||||
|
||||
## 说明
|
||||
- **只有华为 ROM 客户端会来拉**（HarmonyOS / EMUI；荣耀 MagicOS 不拉）。机型判断在客户端做（`OemDetector`），服务端不看 UA，也就不用维护机型名单。
|
||||
- 客户端在进新手引导前拉一次并本地缓存；请求失败/超时用上次缓存值，从未拉到过则按 `onboarding_closable=false`（宁可不给退出按钮，也不误放开）。
|
||||
- 脏值兜底:DB 里 `mode` 不在枚举内时服务端一律回退 `default`。
|
||||
- 切回 `default` 即可一键收回退出按钮（审核通过后无需发版）。
|
||||
@@ -29,3 +29,12 @@
|
||||
## 注意
|
||||
- 不缓存:配置读频率低(每次福利操作读一次,主键查极快),admin 改了立即生效、跨进程一致(多 worker 也对)。
|
||||
- 新增可配项 = 在 `CONFIG_DEFS` 加一条 + 业务处改用 `app_config.get_value(db, key)` 读;不需要建迁移(行是动态插的,表结构不变)。
|
||||
|
||||
## 专用 key(借表不进 CONFIG_DEFS)
|
||||
有自己的语义与专用管理页的配置,复用本表但**不注册进 `CONFIG_DEFS`**——混进通用「系统配置」页只会显示成一个没头没尾的 on/off。它们各有一对 `get_*` / `set_*` 函数(仍在 `repositories/app_config.py`),`value` 存 dict,空行回退各自的模块内默认值。
|
||||
|
||||
| key | 管理页 / admin 端点 | C 端读取 | 说明 |
|
||||
|---|---|---|---|
|
||||
| `ad_config` | `GET/PATCH /admin/api/ad-config` | `GET /api/v1/platform/ad-config`(去密钥) | 穿山甲 app_id / 各代码位 / 各场景开关 |
|
||||
| `app_version` | 内部写入(`X-Internal-Secret`) | `GET /api/v1/platform/app-version` | OTA 最新版本信息 |
|
||||
| `huawei_review` | `GET/PATCH /admin/api/huawei-review` | `GET /api/v1/platform/huawei-review` | 华为审核开关:`{"mode": "default"｜"review"}`,决定新手引导「快速设置」权限步能否被用户关闭。脏值/空行一律回退 `default`(不给退出按钮) |
|
||||
|
||||
@@ -0,0 +1,897 @@
|
||||
# 接口 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`（先只放队列部分）**
|
||||
|
||||
> 注意:本步只放队列相关代码。中间件用到的 `os`/`time`/`Match` 及 `_SKIP_PATHS`/`_UNMATCHED`/`_SERVICE` 常量放到 Task 3 一并加入——否则本步提交时 ruff 会报 F401 未用导入。
|
||||
|
||||
新建 `app/core/observe.py`：
|
||||
|
||||
```python
|
||||
"""接口指标埋点:有界事件队列 + 纯 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: 提交**
|
||||
|
||||
```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 3a: 给 `app/core/observe.py` 补中间件用的导入与常量**
|
||||
|
||||
把顶部 import 段从
|
||||
|
||||
```python
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
from app.core.config import settings
|
||||
```
|
||||
|
||||
改成
|
||||
|
||||
```python
|
||||
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` 末尾）**
|
||||
|
||||
```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(仅本改动涉及文件)+ 全量测试**
|
||||
|
||||
> 说明:仓库基线有 ~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 之外的改动）**
|
||||
|
||||
```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 与文件结构表一致。
|
||||
@@ -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