fix: 修复任务分页总数错误 + 补齐可观测性基建
分页总数（#5）： - get_all_tasks 的 total 原为 len(flat_tasks)（当前页条数），移动端 「我的任务」用 tasks.length < total 判断 hasMore，首页满员时恒为 false，列表永远停在第一页 20 条。改为独立 COUNT 查询（与 notifications.py 已有写法保持一致）。 可观测性（#9）： - 新增 core/logging.py：单行 JSON 结构化日志 + request_id/user 上下文 注入；零第三方依赖；接管 uvicorn 自带 handler 避免格式绕过。 - 新增 core/middleware.py：RequestContextMiddleware 生成/透传 X-Request-ID 并回写响应头，输出含耗时/用户的结构化访问日志。 - 新增 core/health.py：拆分存活/就绪探针。/health/live 不触依赖； /health/ready 探主库，不可用返回 503；MOM 挂掉仅降级不摘流量。 - main.py：全局异常处理器只把堆栈写日志，响应体仅回 request_id； 接入可选 Sentry（未装 SDK 时静默跳过）。 - auth_service：解析 Token 后写入 user 上下文，日志自动带操作人。 注：异常处理器由 ServerErrorMiddleware 调用，此时 contextvar 已被重置， 故 request_id 同时写入 request.state（由 ASGI scope 承载）再读取。 已用 TestClient 验证：探针状态码/检查项、X-Request-ID 透传与生成、 500 响应携带可对账的 request_id 且不泄露堆栈。
This commit is contained in:
openhands
@ -16,6 +16,17 @@ class Settings(BaseSettings):
|
||||
# ---- 调试 ----
|
||||
DEBUG: bool = True
|
||||
|
||||
# ---- 应用元信息 ----
|
||||
APP_VERSION: str = "1.0.0"
|
||||
|
||||
# ---- 日志 ----
|
||||
LOG_LEVEL: str = "INFO"
|
||||
LOG_JSON: bool = True # 生产保持 True（便于采集）；本地调试可设 False 换可读格式
|
||||
|
||||
# ---- 错误追踪（可选，不装 sentry-sdk 则自动跳过）----
|
||||
SENTRY_DSN: str | None = None
|
||||
SENTRY_TRACES_SAMPLE_RATE: float = 0.0
|
||||
|
||||
# ---- CORS 跨域白名单（JSON 数组字符串，直接从 .env 的 CORS_ORIGINS 读取） ----
|
||||
CORS_ORIGINS: str = '["http://localhost:1420", "tauri://localhost"]'
|
||||
|
||||
|
||||
93
backend/app/core/health.py
Normal file
93
backend/app/core/health.py
Normal file
@ -0,0 +1,93 @@
|
||||
"""健康检查 — 存活探针与就绪探针分离
|
||||
|
||||
为什么必须拆开：
|
||||
- 存活探针（liveness）只回答「进程还活着吗」，绝不能探测外部依赖。
|
||||
否则数据库抖一下，编排系统会判定进程已死并反复重启容器，
|
||||
把一次依赖故障放大成全站雪崩。
|
||||
- 就绪探针（readiness）回答「现在能对外服务吗」。依赖不可用时返回 503，
|
||||
由负载均衡把该实例摘掉，依赖恢复后自动回来。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import anyio
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.database import AsyncSessionLocal
|
||||
from app.core.mom_database import mom_engine
|
||||
|
||||
logger = logging.getLogger("track.health")
|
||||
|
||||
router = APIRouter(tags=["健康检查"])
|
||||
|
||||
|
||||
async def _probe_primary_db() -> bool:
|
||||
"""主库探活 — 业务强依赖，失败即不就绪"""
|
||||
try:
|
||||
async with AsyncSessionLocal() as session:
|
||||
await session.execute(text("SELECT 1"))
|
||||
return True
|
||||
except Exception:
|
||||
logger.exception("主库探活失败")
|
||||
return False
|
||||
|
||||
|
||||
def _probe_mom_db_sync() -> bool:
|
||||
try:
|
||||
with mom_engine.connect() as conn:
|
||||
conn.execute(text("SELECT 1"))
|
||||
return True
|
||||
except Exception:
|
||||
logger.exception("MOM 库探活失败")
|
||||
return False
|
||||
|
||||
|
||||
async def _probe_mom_db() -> bool:
|
||||
# MOM 用的是同步引擎，放线程池执行，避免阻塞事件循环
|
||||
return await anyio.to_thread.run_sync(_probe_mom_db_sync)
|
||||
|
||||
|
||||
async def _collect() -> tuple[bool, dict[str, str]]:
|
||||
primary_ok = await _probe_primary_db()
|
||||
mom_ok = await _probe_mom_db()
|
||||
checks = {
|
||||
"database": "ok" if primary_ok else "fail",
|
||||
# MOM 是外部只读依赖：挂掉时登录/选料降级，但扫码、流转、看板仍可用。
|
||||
# 因此只标记 degraded、不摘流量 —— 否则 MOM 一抖就让在产车间全线停摆。
|
||||
"mom_database": "ok" if mom_ok else "degraded",
|
||||
}
|
||||
return primary_ok, checks
|
||||
|
||||
|
||||
@router.get("/health/live", include_in_schema=False)
|
||||
async def liveness() -> dict:
|
||||
"""存活探针：不触碰任何依赖，恒定快速返回"""
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@router.get("/health/ready", include_in_schema=False)
|
||||
async def readiness() -> JSONResponse:
|
||||
"""就绪探针：主库不可用时返回 503，让负载均衡摘流量"""
|
||||
ready, checks = await _collect()
|
||||
return JSONResponse(
|
||||
{"status": "ready" if ready else "not_ready", "checks": checks},
|
||||
status_code=200 if ready else 503,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/health", include_in_schema=False)
|
||||
async def health() -> JSONResponse:
|
||||
"""兼容旧监控脚本：语义等同就绪探针，并附带版本号"""
|
||||
ready, checks = await _collect()
|
||||
return JSONResponse(
|
||||
{
|
||||
"status": "ok" if ready else "unavailable",
|
||||
"version": settings.APP_VERSION,
|
||||
"checks": checks,
|
||||
},
|
||||
status_code=200 if ready else 503,
|
||||
)
|
||||
89
backend/app/core/logging.py
Normal file
89
backend/app/core/logging.py
Normal file
@ -0,0 +1,89 @@
|
||||
"""结构化日志 — 单行 JSON 输出 + 请求上下文注入
|
||||
|
||||
设计要点：
|
||||
1. 零第三方依赖，只用 stdlib（logging + json + contextvars）。
|
||||
2. 业务代码通过 `extra={"extra_fields": {...}}` 附加结构化字段，
|
||||
不要把可检索的字段拼进 msg 字符串 —— 拼进去就只能靠正则捞了。
|
||||
3. request_id / user 走 contextvar。contextvar 在 asyncio 下按任务隔离，
|
||||
并发请求之间不会串号；由 RequestContextMiddleware 与 get_current_user 写入。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from contextvars import ContextVar
|
||||
from datetime import datetime, timezone
|
||||
|
||||
# 请求级上下文
|
||||
request_id_var: ContextVar[str | None] = ContextVar("request_id", default=None)
|
||||
user_var: ContextVar[str | None] = ContextVar("user", default=None)
|
||||
|
||||
|
||||
class _ContextFilter(logging.Filter):
|
||||
"""把 contextvar 注入每条 record，使 JSON 自带 request_id / user"""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
record.request_id = request_id_var.get()
|
||||
record.user = user_var.get()
|
||||
return True
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
"""单行 JSON — 便于 Loki / ELK / CloudWatch 直接解析，无需正则"""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
payload: dict = {
|
||||
"ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
|
||||
"level": record.levelname,
|
||||
"logger": record.name,
|
||||
"msg": record.getMessage(),
|
||||
}
|
||||
if getattr(record, "request_id", None):
|
||||
payload["request_id"] = record.request_id
|
||||
if getattr(record, "user", None):
|
||||
payload["user"] = record.user
|
||||
payload.update(getattr(record, "extra_fields", None) or {})
|
||||
if record.exc_info:
|
||||
payload["exc"] = self.formatException(record.exc_info)
|
||||
return json.dumps(payload, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
class TextFormatter(logging.Formatter):
|
||||
"""本地开发可读格式（LOG_JSON=false 时启用）"""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
line = (
|
||||
f"{self.formatTime(record, '%H:%M:%S')} "
|
||||
f"{record.levelname:<5} {record.name} - {record.getMessage()}"
|
||||
)
|
||||
extras = getattr(record, "extra_fields", None)
|
||||
if extras:
|
||||
line += " | " + " ".join(f"{k}={v}" for k, v in extras.items())
|
||||
if record.exc_info:
|
||||
line += "\n" + self.formatException(record.exc_info)
|
||||
return line
|
||||
|
||||
|
||||
def setup_logging(level: str = "INFO", json_output: bool = True) -> None:
|
||||
"""配置根 logger。必须在应用启动前调用一次。"""
|
||||
handler = logging.StreamHandler(sys.stdout)
|
||||
handler.setFormatter(JsonFormatter() if json_output else TextFormatter())
|
||||
handler.addFilter(_ContextFilter())
|
||||
|
||||
root = logging.getLogger()
|
||||
# 清空既有 handler：uvicorn --reload / 多 worker 下模块可能被重复导入，
|
||||
# 不清会看到每条日志打印 N 遍
|
||||
root.handlers.clear()
|
||||
root.addHandler(handler)
|
||||
root.setLevel(level.upper())
|
||||
|
||||
# uvicorn 自带 handler 会绕过上面的 formatter，必须清掉并让它向根传播
|
||||
for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
|
||||
lg = logging.getLogger(name)
|
||||
lg.handlers.clear()
|
||||
lg.propagate = True
|
||||
|
||||
# 访问日志统一由 RequestContextMiddleware 输出（含耗时 / 用户 / request_id），
|
||||
# 故关闭 uvicorn 自带的访问日志，避免重复
|
||||
logging.getLogger("uvicorn.access").disabled = True
|
||||
83
backend/app/core/middleware.py
Normal file
83
backend/app/core/middleware.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""请求上下文中间件 — request_id 生成/透传 + 结构化访问日志"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
from starlette.requests import Request
|
||||
|
||||
from app.core.logging import request_id_var, user_var
|
||||
|
||||
access_log = logging.getLogger("track.access")
|
||||
|
||||
# 探针被高频轮询，降级为 DEBUG 避免把有价值的信息淹掉
|
||||
_QUIET_PATHS = frozenset({"/health", "/health/live", "/health/ready"})
|
||||
|
||||
|
||||
class RequestContextMiddleware(BaseHTTPMiddleware):
|
||||
"""为每个请求建立可追踪上下文。
|
||||
|
||||
- request_id：优先沿用上游网关传来的 X-Request-ID，实现全链路追踪；
|
||||
没有就生成一个。响应头回写该 ID，前端报错时可直接带上，
|
||||
运维拿 ID 就能在日志里精确定位到这一次请求。
|
||||
- 访问日志：method / path / status / duration_ms / client / user。
|
||||
"""
|
||||
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
|
||||
# 同时写入 request.state：它由 ASGI scope 承载，作用域比 contextvar 更长。
|
||||
# FastAPI 把 Exception 处理器交给 ServerErrorMiddleware（位于本中间件外层），
|
||||
# 异常传播到那里时 contextvar 已在 finally 中被重置，只有 state 还留着 ID。
|
||||
request.state.request_id = request_id
|
||||
rid_token = request_id_var.set(request_id)
|
||||
user_token = user_var.set(None)
|
||||
started = time.perf_counter()
|
||||
logged = False
|
||||
status_code = 500
|
||||
try:
|
||||
response = await call_next(request)
|
||||
status_code = response.status_code
|
||||
response.headers["X-Request-ID"] = request_id
|
||||
self._log_access(request, status_code, started)
|
||||
logged = True
|
||||
return response
|
||||
finally:
|
||||
# 异常路径也要留下访问记录，否则接口 500 时日志里反而没有痕迹
|
||||
if not logged:
|
||||
self._log_access(request, status_code, started)
|
||||
request_id_var.reset(rid_token)
|
||||
user_var.reset(user_token)
|
||||
|
||||
def _log_access(self, request: Request, status_code: int, started: float) -> None:
|
||||
path = request.url.path
|
||||
duration_ms = round((time.perf_counter() - started) * 1000, 1)
|
||||
|
||||
if status_code >= 500:
|
||||
level = logging.ERROR
|
||||
elif status_code >= 400:
|
||||
level = logging.WARNING
|
||||
elif path in _QUIET_PATHS:
|
||||
level = logging.DEBUG
|
||||
else:
|
||||
level = logging.INFO
|
||||
|
||||
access_log.log(
|
||||
level,
|
||||
"%s %s -> %s (%.1fms)",
|
||||
request.method,
|
||||
path,
|
||||
status_code,
|
||||
duration_ms,
|
||||
extra={
|
||||
"extra_fields": {
|
||||
"method": request.method,
|
||||
"path": path,
|
||||
"status": status_code,
|
||||
"duration_ms": duration_ms,
|
||||
"client": request.client.host if request.client else None,
|
||||
"user": user_var.get(),
|
||||
}
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user