feat(gateway): 全局串行化(并发=1) + 记录入参/出参/耗时日志

并发模型从「每worker并发1 + 多worker并行」改为全局串行:同一时间
只处理1个请求,其余排队;多 worker 仅作热备(主 worker 坏了才用备机)。
接口4(face/features) 走豆包、不占 GPU,不纳入串行。

- pool: asyncio.Semaphore(max_global_concurrency=1) + acquire/release_global_slot
  (依赖单进程 uvicorn 部署,已在注释中标注)
- forward: proxy_request 最外层 acquire 全局槽、try/finally 全路径释放;
  入参 multipart 解析挂 request.state;原重试/故障转移逻辑抽到 _dispatch_with_retries
- reqlog(新): 标量入参保留;图片(file/base64)存盘转URL,绝不内嵌base64;
  出参递归摘要截断(landmarks/长串/大数组)
- logging_middleware: RequestLogEntry 加 request_params/response_data,
  jsonl 全量记录;_load_from_logfile 同步映射防重启丢字段;get_stats recent 暴露
- /gateway-health 暴露 global_max/global_busy/global_waiting
- config: dispatch 新增 max_global_concurrency / max_queue_wait_seconds
- tests: test_reqlog + test_gateway_serialization(9 用例)

顺带提交此前未提交的网关统计(daily stats 接口)与耗时看板(api_timing_dashboard.html)。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Ubuntu
2026-07-23 22:56:48 +08:00
co-authored by Claude
parent 98b9108837
commit bb9f55e93c
10 changed files with 1037 additions and 7 deletions
+39
View File
@@ -20,6 +20,8 @@ from fastapi.staticfiles import StaticFiles
from gateway.config import load_config from gateway.config import load_config
from gateway.logging_middleware import ( from gateway.logging_middleware import (
get_available_dates,
get_daily_stats,
get_stats, get_stats,
init_logging as _init_req_logging, init_logging as _init_req_logging,
request_logging_middleware, request_logging_middleware,
@@ -185,6 +187,9 @@ async def gateway_health():
"workers_total": status["total"], "workers_total": status["total"],
"workers_healthy": status["healthy"], "workers_healthy": status["healthy"],
"workers_busy": status["busy"], "workers_busy": status["busy"],
"global_max": status.get("global_max", 1),
"global_busy": status.get("global_busy", 0),
"global_waiting": status.get("global_waiting", 0),
} }
@@ -201,6 +206,7 @@ async def index():
"version": "0.1.0", "version": "0.1.0",
"docs": "/docs", "docs": "/docs",
"stats": "/admin/stats", "stats": "/admin/stats",
"timing_dashboard": "/static/api_timing_dashboard.html",
"integration_guide": "/static/integration.html", "integration_guide": "/static/integration.html",
"test_pages": { "test_pages": {
"if1_measure": "/static/test_interface1.html", "if1_measure": "/static/test_interface1.html",
@@ -470,6 +476,18 @@ async def admin_stats_json():
return get_stats() return get_stats()
@app.get("/admin/stats/dates", include_in_schema=False)
async def admin_stats_dates():
"""返回日志中出现过的所有本地日期(供耗时看板日期选择器使用)。"""
return {"dates": get_available_dates()}
@app.get("/admin/stats/daily", include_in_schema=False)
async def admin_stats_daily(date: str):
"""按天统计各接口调用耗时(含多发型接口的单次换发型耗时拆解)。"""
return get_daily_stats(date)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# 代理路由 # 代理路由
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -529,6 +547,7 @@ async def hair_grow_b(request: Request):
@app.post("/api/v1/face/features", tags=["人脸分析"]) @app.post("/api/v1/face/features", tags=["人脸分析"])
async def face_features( async def face_features(
request: Request,
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件(JPG/PNG"), image_file: Optional[UploadFile] = File(default=None, description="上传图片文件(JPG/PNG"),
image_url: Optional[str] = Form(default=None, description="图片 URL"), image_url: Optional[str] = Form(default=None, description="图片 URL"),
image_base64: Optional[str] = Form(default=None, description="图片 base64(需带前缀)"), image_base64: Optional[str] = Form(default=None, description="图片 base64(需带前缀)"),
@@ -559,6 +578,26 @@ async def face_features(
"request_id": f"gw-{_uuid.uuid4().hex[:8]}", "data": None, "request_id": f"gw-{_uuid.uuid4().hex[:8]}", "data": None,
}) })
# 入参日志(接口4 不走 forward.py,自行记录;不获取全局串行槽,因不占 GPU)
try:
from gateway.config import get_config as _get_cfg
from gateway.reqlog import save_image_bytes
_cfg = _get_cfg()
_params: dict = {}
if image_file:
_url = save_image_bytes(img_bytes, _cfg["static_dir"], _cfg["public_base_url"]) if img_bytes else None
_params["image_file"] = {"_image": True, "source": "file",
"size": len(img_bytes) if img_bytes else 0, "url": _url}
elif image_base64:
_url = save_image_bytes(img_bytes, _cfg["static_dir"], _cfg["public_base_url"]) if img_bytes else None
_params["image_base64"] = {"_image": True, "source": "base64",
"size": len(img_bytes) if img_bytes else 0, "url": _url}
elif image_url:
_params["image_url"] = image_url
request.state.log_request_params = _params
except Exception: # noqa: BLE001
request.state.log_request_params = {"_parse_error": "iface4"}
from fastapi.concurrency import run_in_threadpool from fastapi.concurrency import run_in_threadpool
from face_features import analyze_features from face_features import analyze_features
+2
View File
@@ -16,6 +16,8 @@
"ark_api_key": "", "ark_api_key": "",
"dispatch": { "dispatch": {
"per_worker_concurrency": 1, "per_worker_concurrency": 1,
"max_global_concurrency": 1,
"max_queue_wait_seconds": 590,
"queue_wait_seconds": 30, "queue_wait_seconds": 30,
"request_timeout_seconds": 600, "request_timeout_seconds": 600,
"retry_on_failure": true, "retry_on_failure": true,
+6 -1
View File
@@ -24,6 +24,8 @@ DEFAULTS = {
}, },
"dispatch": { "dispatch": {
"per_worker_concurrency": 1, "per_worker_concurrency": 1,
"max_global_concurrency": 1,
"max_queue_wait_seconds": 590,
"queue_wait_seconds": 30, "queue_wait_seconds": 30,
"request_timeout_seconds": 600, "request_timeout_seconds": 600,
"retry_on_failure": True, "retry_on_failure": True,
@@ -105,12 +107,15 @@ def load_config() -> dict:
logger.info( logger.info(
"配置加载完成 | workers=%s | public_base_url=%s | " "配置加载完成 | workers=%s | public_base_url=%s | "
"hc_interval=%ds | dispatch_timeout=%ds | queue_wait=%ds", "hc_interval=%ds | dispatch_timeout=%ds | queue_wait=%ds | "
"global_concurrency=%d | max_queue_wait=%ds",
cfg["workers"], cfg["workers"],
cfg["public_base_url"], cfg["public_base_url"],
cfg["health_check"]["interval_seconds"], cfg["health_check"]["interval_seconds"],
cfg["dispatch"]["request_timeout_seconds"], cfg["dispatch"]["request_timeout_seconds"],
cfg["dispatch"]["queue_wait_seconds"], cfg["dispatch"]["queue_wait_seconds"],
cfg["dispatch"]["max_global_concurrency"],
cfg["dispatch"]["max_queue_wait_seconds"],
) )
_config_cache = cfg _config_cache = cfg
+46 -4
View File
@@ -19,10 +19,13 @@ from fastapi.responses import JSONResponse
from gateway.pool import ( from gateway.pool import (
NoWorkerAvailable, NoWorkerAvailable,
acquire_global_slot,
acquire_worker, acquire_worker,
mark_worker_unhealthy, mark_worker_unhealthy,
release_global_slot,
release_worker, release_worker,
) )
from gateway.reqlog import extract_form_params
logger = logging.getLogger("gateway.forward") logger = logging.getLogger("gateway.forward")
@@ -143,17 +146,56 @@ async def proxy_request(request: Request, path: str) -> JSONResponse:
""" """
from gateway.config import get_config from gateway.config import get_config
cfg = get_config() cfg = get_config()
# --- 1. 读取客户端请求体(原始字节,不做解析) ---
body = await request.body()
# --- 入参日志解析(非致命,绝不影响转发) ---
try:
request.state.log_request_params = extract_form_params(
request.headers.get("content-type", ""),
body,
cfg["static_dir"],
cfg["public_base_url"],
)
except Exception as ex: # noqa: BLE001
request.state.log_request_params = {"_parse_error": str(ex)}
# --- 全局串行槽(GPU 请求在此排队;超时拒绝) ---
if not await acquire_global_slot(cfg):
request.state.worker_url = ""
return JSONResponse(
status_code=503,
content={
"code": 1007,
"message": "排队超时,请稍后重试",
"request_id": f"gw-{uuid.uuid4().hex[:8]}",
"data": None,
},
)
try:
return await _dispatch_with_retries(request, path, cfg, body)
finally:
release_global_slot()
async def _dispatch_with_retries(
request: Request, path: str, cfg: dict, body: bytes
) -> JSONResponse:
"""在已持有全局串行槽的前提下:选 worker → 转发 → 重试 → 改写 base64 → 返回。
故障转移:worker 失败 → mark_worker_unhealthy → release_worker → 循环再 acquire
(此时主 worker 已下线,自动选备机)。全程在同一个全局槽持有期内,故
「平时只打主 worker,坏了才用备机」。
"""
dispatch_cfg = cfg["dispatch"] dispatch_cfg = cfg["dispatch"]
token = cfg["shared_password"] token = cfg["shared_password"]
request_timeout = dispatch_cfg["request_timeout_seconds"] request_timeout = dispatch_cfg["request_timeout_seconds"]
max_retries = dispatch_cfg.get("max_retries", 1) max_retries = dispatch_cfg.get("max_retries", 1)
retry_on_failure = dispatch_cfg.get("retry_on_failure", True)
public_base_url = cfg["public_base_url"] public_base_url = cfg["public_base_url"]
static_dir = cfg["static_dir"] static_dir = cfg["static_dir"]
# --- 1. 读取客户端请求体(原始字节,不做解析) ---
body = await request.body()
# --- 2. 获取 worker(最多重试 max_retries+1 次) --- # --- 2. 获取 worker(最多重试 max_retries+1 次) ---
attempts = max_retries + 1 attempts = max_retries + 1
last_error_response = None last_error_response = None
+262
View File
@@ -14,6 +14,8 @@ from dataclasses import asdict, dataclass, field
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from gateway.reqlog import summarize_for_log
logger = logging.getLogger("gateway.logging_middleware") logger = logging.getLogger("gateway.logging_middleware")
@@ -34,6 +36,8 @@ class RequestLogEntry:
worker: str = "" # 处理请求的 worker URL(空串表示网关本地处理) worker: str = "" # 处理请求的 worker URL(空串表示网关本地处理)
response_code: Optional[int] = None # 响应 JSON 中的 code 字段 response_code: Optional[int] = None # 响应 JSON 中的 code 字段
request_id: Optional[str] = None # 响应 JSON 中的 request_id request_id: Optional[str] = None # 响应 JSON 中的 request_id
request_params: Optional[dict] = None # 入参(图片用 URL 引用,无 base64)
response_data: Optional[Any] = None # 出参摘要(递归截断)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -195,6 +199,8 @@ def _load_from_logfile(filepath: str, max_entries: int) -> int:
worker=data.get("worker", ""), worker=data.get("worker", ""),
response_code=data.get("response_code"), response_code=data.get("response_code"),
request_id=data.get("request_id"), request_id=data.get("request_id"),
request_params=data.get("request_params"),
response_data=data.get("response_data"),
) )
_buffer.append(entry) _buffer.append(entry)
count += 1 count += 1
@@ -248,6 +254,7 @@ async def request_logging_middleware(request, call_next):
# 提取响应 body 并解析业务字段(仅 JSON 响应) # 提取响应 body 并解析业务字段(仅 JSON 响应)
response_code = None response_code = None
request_id = None request_id = None
response_data = None
content_type = response.headers.get("content-type", "") content_type = response.headers.get("content-type", "")
if "application/json" in content_type or "application/json" in (response.media_type or ""): if "application/json" in content_type or "application/json" in (response.media_type or ""):
@@ -261,6 +268,7 @@ async def request_logging_middleware(request, call_next):
data = json.loads(body) data = json.loads(body)
response_code = data.get("code") response_code = data.get("code")
request_id = data.get("request_id") request_id = data.get("request_id")
response_data = summarize_for_log(data)
except (json.JSONDecodeError, UnicodeDecodeError): except (json.JSONDecodeError, UnicodeDecodeError):
pass pass
@@ -274,6 +282,9 @@ async def request_logging_middleware(request, call_next):
media_type=response.media_type, media_type=response.media_type,
) )
# 入参(由 forward.py / 接口4 handler 挂到 request.state
request_params = getattr(request.state, "log_request_params", None)
# 记录 # 记录
now = datetime.datetime.utcnow() now = datetime.datetime.utcnow()
entry = RequestLogEntry( entry = RequestLogEntry(
@@ -287,6 +298,8 @@ async def request_logging_middleware(request, call_next):
worker=worker_url, worker=worker_url,
response_code=response_code, response_code=response_code,
request_id=request_id, request_id=request_id,
request_params=request_params,
response_data=response_data,
) )
_buffer.append(entry) _buffer.append(entry)
@@ -366,6 +379,8 @@ def get_stats() -> Dict[str, Any]:
"worker": e.worker, "worker": e.worker,
"response_code": e.response_code, "response_code": e.response_code,
"request_id": e.request_id, "request_id": e.request_id,
"request_params": e.request_params,
"response_data": e.response_data,
} }
for e in recent_100 for e in recent_100
] ]
@@ -404,3 +419,250 @@ def get_stats() -> Dict[str, Any]:
"recent": recent, "recent": recent,
"last_updated": datetime.datetime.utcnow().isoformat() + "Z", "last_updated": datetime.datetime.utcnow().isoformat() + "Z",
} }
# ---------------------------------------------------------------------------
# 按天统计(读取磁盘日志全量,支持任意历史日期 + 单次换发型耗时拆解)
# ---------------------------------------------------------------------------
# 接口路径 → 编号/名称(对齐 gateway/app.py 里各路由 docstring 中的「接口N」编号)
PATH_INTERFACE_MAP: Dict[str, Dict[str, Any]] = {
"/api/v1/face/measure": {"num": 1, "name": "四庭七眼测量标注"},
"/api/v1/hair/grow": {"num": 2, "name": "C端生发"},
"/api/v1/hair/grow-b": {"num": 3, "name": "B端生发(医生/操作端)"},
"/api/v1/face/features": {"num": 4, "name": "用户特征分析"},
"/api/v1/hairline/generate": {"num": 5, "name": "发际线PNG生成"},
"/api/v1/face/measure-v2": {"num": 6, "name": "四庭七眼测量标注 v2"},
"/api/v1/hair/grow-v2": {"num": 7, "name": "C端生发 v2"},
"/api/v1/head/mask": {"num": 9, "name": "头发遮罩生成"},
"/api/v1/head/band": {"num": 10, "name": "头部外缘膨胀带遮罩"},
}
# 会按「勾选发型数量」串行多次调用 ComfyUI 的接口(耗时 ≈ 固定开销 + 单次换发型耗时 × 发型数)
MULTI_STYLE_PATHS = {
"/api/v1/hair/grow",
"/api/v1/hair/grow-v2",
"/api/v1/hairline/generate",
}
def _log_file_path() -> Path:
"""当前日志文件路径(若尚未 init_logging,用默认路径兜底)。"""
if _writer is not None:
return _writer.filepath
return Path(__file__).resolve().parent.parent / "gateway" / "request_log.jsonl"
def _read_all_log_entries() -> List[dict]:
"""读取磁盘上完整的日志历史(含轮转备份 .jsonl.1,按时间顺序:备份在前)。"""
log_path = _log_file_path()
paths = []
backup = log_path.with_suffix(".jsonl.1")
if backup.exists():
paths.append(backup)
if log_path.exists():
paths.append(log_path)
entries: List[dict] = []
for p in paths:
try:
with open(p, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
entries.append(json.loads(line))
except json.JSONDecodeError:
continue
except Exception:
logger.warning("读取日志文件失败: %s", p, exc_info=True)
return entries
def _parse_ts(ts: str) -> Optional[datetime.datetime]:
for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ"):
try:
return datetime.datetime.strptime(ts, fmt).replace(tzinfo=datetime.timezone.utc)
except ValueError:
continue
return None
def _percentile(values: List[float], p: float) -> float:
"""线性插值百分位数,values 需已排序。"""
if not values:
return 0.0
if len(values) == 1:
return values[0]
k = (len(values) - 1) * p
f = int(k)
c = min(f + 1, len(values) - 1)
if f == c:
return values[f]
return values[f] + (values[c] - values[f]) * (k - f)
def get_available_dates(tz_offset_hours: float = 8) -> List[str]:
"""返回日志中出现过的所有本地日期(YYYY-MM-DD),最新在前。"""
tz = datetime.timezone(datetime.timedelta(hours=tz_offset_hours))
dates = set()
for e in _read_all_log_entries():
t = _parse_ts(e.get("timestamp", ""))
if t is None:
continue
dates.add(t.astimezone(tz).date().isoformat())
return sorted(dates, reverse=True)
def _estimate_per_style(durations_ms: List[float]) -> Optional[Dict[str, Any]]:
"""从一组请求耗时里拆解出「单次换发型边际耗时」与「固定开销」。
原理:这类接口对每个勾选的发型串行跑一次 ComfyUI,总耗时 ≈ 固定开销(人脸检测等)
+ 单次换发型耗时 × 发型数量。发型数量未被记录,这里用迭代最小二乘估计"量子"
(近似每多选 1 个发型多花多少秒),再据此把样本归类、做线性回归得到最终估计。
样本耗时种类过少(不足以分辨固定开销与边际耗时)时返回 None。
"""
xs = sorted(d / 1000.0 for d in durations_ms if d and d > 0)
if len(xs) < 3:
return None
guess = xs[0]
if guess <= 0:
return None
def _fit(ns: List[int], xs_: List[float]):
n_mean = sum(ns) / len(ns)
x_mean = sum(xs_) / len(xs_)
num = sum((n - n_mean) * (x - x_mean) for n, x in zip(ns, xs_))
den = sum((n - n_mean) ** 2 for n in ns)
if den == 0:
return None
slope = num / den
intercept = x_mean - slope * n_mean
return slope, intercept
for _ in range(8):
ns = [max(1, round(x / guess)) for x in xs]
fit = _fit(ns, xs)
if fit is None or fit[0] <= 0:
break
guess = fit[0]
ns_final = [max(1, round(x / guess)) for x in xs]
if len(set(ns_final)) < 2:
return None # 样本都挤在同一档,无法拆分固定开销 / 边际耗时
fit = _fit(ns_final, xs)
if fit is None or fit[0] <= 0:
return None
slope, intercept = fit
groups: Dict[int, List[float]] = {}
for x, n in zip(xs, ns_final):
groups.setdefault(n, []).append(x)
clusters = [
{"styles": n, "count": len(v), "avg_seconds": round(sum(v) / len(v), 2)}
for n, v in sorted(groups.items())
]
return {
"per_style_seconds": round(slope, 2),
"fixed_overhead_seconds": round(max(intercept, 0.0), 2),
"sample_count": len(xs),
"clusters": clusters,
}
def get_daily_stats(date_str: str, tz_offset_hours: float = 8) -> Dict[str, Any]:
"""统计某一天(本地时区,默认 UTC+8)的接口调用情况。
读取磁盘上的完整日志历史(不受内存环形缓冲区大小限制),返回:
- summary:当天总请求数/成功率/总耗时/平均耗时
- endpoints:按接口分组的详细统计(含接口编号/名称、耗时分布、
对「多发型串行」接口额外给出单次换发型耗时拆解)
- hourly:按小时的请求量分布,供画图用
"""
tz = datetime.timezone(datetime.timedelta(hours=tz_offset_hours))
try:
target_date = datetime.date.fromisoformat(date_str)
except ValueError:
return {"error": f"日期格式错误: {date_str!r},应为 YYYY-MM-DD"}
day_entries = []
for e in _read_all_log_entries():
if not str(e.get("path", "")).startswith("/api/"):
continue
t = _parse_ts(e.get("timestamp", ""))
if t is None:
continue
local = t.astimezone(tz)
if local.date() == target_date:
e = dict(e)
e["_local_hour"] = local.hour
day_entries.append(e)
if not day_entries:
return {
"date": date_str,
"summary": {"total": 0, "success_rate": 0, "avg_duration_ms": 0,
"total_duration_seconds": 0, "min_duration_ms": 0, "max_duration_ms": 0},
"endpoints": [],
"hourly": [],
}
total = len(day_entries)
ok_count = sum(1 for e in day_entries
if e.get("status_code") == 200 and e.get("response_code") in (0, None))
all_durations = sorted(e.get("duration_ms", 0.0) for e in day_entries)
by_path: Dict[str, List[dict]] = {}
for e in day_entries:
by_path.setdefault(e["path"], []).append(e)
endpoints = []
for path, items in by_path.items():
durs = sorted(it.get("duration_ms", 0.0) for it in items)
ok = sum(1 for it in items
if it.get("status_code") == 200 and it.get("response_code") in (0, None))
info = PATH_INTERFACE_MAP.get(path, {"num": None, "name": path})
entry: Dict[str, Any] = {
"path": path,
"interface_num": info["num"],
"interface_name": info["name"],
"count": len(items),
"ok_count": ok,
"fail_count": len(items) - ok,
"success_rate": round(ok / len(items) * 100, 1),
"avg_duration_ms": round(sum(durs) / len(durs), 1),
"median_duration_ms": round(_percentile(durs, 0.5), 1),
"p95_duration_ms": round(_percentile(durs, 0.95), 1),
"min_duration_ms": round(durs[0], 1),
"max_duration_ms": round(durs[-1], 1),
"total_duration_seconds": round(sum(durs) / 1000, 1),
"per_style": None,
}
if path in MULTI_STYLE_PATHS:
entry["per_style"] = _estimate_per_style(durs)
endpoints.append(entry)
endpoints.sort(key=lambda x: -x["count"])
hourly_counts: Dict[int, int] = {}
for e in day_entries:
hourly_counts[e["_local_hour"]] = hourly_counts.get(e["_local_hour"], 0) + 1
hourly = [{"hour": h, "count": hourly_counts.get(h, 0)} for h in range(24) if hourly_counts.get(h, 0) > 0]
return {
"date": date_str,
"summary": {
"total": total,
"success_rate": round(ok_count / total * 100, 1),
"avg_duration_ms": round(sum(all_durations) / total, 1),
"total_duration_seconds": round(sum(all_durations) / 1000, 1),
"min_duration_ms": round(all_durations[0], 1),
"max_duration_ms": round(all_durations[-1], 1),
},
"endpoints": endpoints,
"hourly": hourly,
}
+68 -2
View File
@@ -50,6 +50,14 @@ _pool_condition: Optional[asyncio.Condition] = None
_health_task: Optional[asyncio.Task] = None _health_task: Optional[asyncio.Task] = None
_shutdown_event: Optional[asyncio.Event] = None _shutdown_event: Optional[asyncio.Event] = None
# 全局串行槽:同一时间最多 max_global_concurrency 个请求进入 worker 派发,其余排队。
# ⚠️ 依赖单进程 uvicorn 部署(hair-gateway.service 无 --workers)。
# 若改多 worker / gunicorn,进程内信号量会失效,需换成跨进程锁(文件锁/Redis)。
_global_sem: Optional[asyncio.Semaphore] = None
_global_max: int = 1
_global_busy: int = 0
_global_waiting: int = 0
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# 健康检查后台任务 # 健康检查后台任务
@@ -162,6 +170,9 @@ async def init_pool(cfg: dict) -> None:
_pool_condition = asyncio.Condition() _pool_condition = asyncio.Condition()
_shutdown_event = asyncio.Event() _shutdown_event = asyncio.Event()
# 全局串行槽(独立于 worker busy 标志)
init_global_slot(cfg)
# 立即做一轮健康检查以快速上线 # 立即做一轮健康检查以快速上线
hc_cfg = cfg["health_check"] hc_cfg = cfg["health_check"]
token = cfg["shared_password"] token = cfg["shared_password"]
@@ -206,6 +217,56 @@ async def shutdown_pool() -> None:
logger.info("Worker 池已关闭") logger.info("Worker 池已关闭")
# ---------------------------------------------------------------------------
# 全局串行槽
# ---------------------------------------------------------------------------
def init_global_slot(cfg: dict) -> None:
"""初始化全局串行信号量。
max_global_concurrency=1 → 同一时间只有一个请求进入 worker 派发,其余在信号量上排队。
依赖单进程 uvicorn 部署;多进程需换跨进程锁。
"""
global _global_sem, _global_max, _global_busy, _global_waiting
n = int(cfg.get("dispatch", {}).get("max_global_concurrency", 1))
if n < 1:
n = 1
_global_sem = asyncio.Semaphore(n)
_global_max = n
_global_busy = 0
_global_waiting = 0
logger.info("全局并发槽初始化: max=%d(单进程生效)", n)
async def acquire_global_slot(cfg: dict) -> bool:
"""获取一个全局处理槽。True=获得;False=排队超时。
超时由 dispatch.max_queue_wait_seconds 控制(需 < nginx proxy_read_timeout 600s)。
"""
global _global_busy, _global_waiting
if _global_sem is None:
return True # 未初始化,不限流
timeout = float(cfg.get("dispatch", {}).get("max_queue_wait_seconds", 590))
_global_waiting += 1
try:
await asyncio.wait_for(_global_sem.acquire(), timeout=timeout)
_global_busy += 1
return True
except asyncio.TimeoutError:
logger.warning("全局排队超时(%.1fs),拒绝请求 | waiting=%d", timeout, _global_waiting)
return False
finally:
_global_waiting -= 1
def release_global_slot() -> None:
"""释放全局处理槽。"""
global _global_busy
if _global_sem is not None:
_global_busy -= 1
_global_sem.release()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# 派发 # 派发
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -270,9 +331,14 @@ async def release_worker(w: WorkerState) -> None:
def get_pool_status() -> dict: def get_pool_status() -> dict:
"""返回当前池状态(供 /gateway-health 使用)。""" """返回当前池状态(供 /gateway-health 使用)。"""
global_info = {
"global_max": _global_max,
"global_busy": _global_busy,
"global_waiting": _global_waiting,
}
if not _workers: if not _workers:
return {"total": 0, "healthy": 0, "busy": 0} return {"total": 0, "healthy": 0, "busy": 0, **global_info}
total = len(_workers) total = len(_workers)
healthy = sum(1 for w in _workers.values() if w.online) healthy = sum(1 for w in _workers.values() if w.online)
busy = sum(1 for w in _workers.values() if w.busy) busy = sum(1 for w in _workers.values() if w.busy)
return {"total": total, "healthy": healthy, "busy": busy} return {"total": total, "healthy": healthy, "busy": busy, **global_info}
+216
View File
@@ -0,0 +1,216 @@
"""请求日志辅助:multipart 入参解析(标量保留 + 图片存盘转 URL)、出参摘要。
设计原则(与用户约定):
- 入参/出参日志里**绝不内嵌图片 base64**。图片统一存盘后用 URL 引用,保持日志短小。
- 入参图片(image_file / *_base64)当前网关不存盘——这里补存到 static_dir(in_ 前缀),
复用现有 24h 清理循环自动回收;url 输入图本身就在远端,不重新下载,直接记 URL。
- 出参响应在 forward.py 已把 base64 改写成 *_url(无大图),记录完整 data,
但用递归摘要器截断超长结构(landmarks、长字符串),单条上限 ~8KB。
所有解析/存盘均包 try/except,**绝不影响请求转发**。
"""
import base64
import io
import json
import logging
import uuid
from pathlib import Path
from typing import Any, Optional
from python_multipart.multipart import parse_form, parse_options_header
logger = logging.getLogger("gateway.reqlog")
_PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
def _sniff_ext(data: bytes) -> str:
"""按 magic byte 嗅探图片扩展名(PNG / 否则 JPEG)。与 forward.py 一致。"""
return "png" if data[:8] == _PNG_MAGIC else "jpg"
def save_image_bytes(
data: bytes,
static_dir: str,
public_base_url: str,
prefix: str = "in_",
) -> Optional[str]:
"""把图片字节存盘并返回公网 URL。失败返回 None(不抛异常)。
存到 static_dir/{prefix}{uuid}.{ext}URL = {public_base_url}/static/annotations/{file}
与 forward.py 的 rewrite_base64_to_url 落盘路径/URL 规则一致,可被同一清理循环回收。
"""
if not data:
return None
try:
ext = _sniff_ext(data)
filename = f"{prefix}{uuid.uuid4().hex}.{ext}"
target = Path(static_dir)
target.mkdir(parents=True, exist_ok=True)
(target / filename).write_bytes(data)
return f"{public_base_url}/static/annotations/{filename}"
except Exception as e: # noqa: BLE001
logger.warning("保存输入图片失败: %s", e)
return None
def _decode_b64_text(text: str) -> Optional[bytes]:
"""解码 base64 文本(兼容 data: URI 前缀和裸 base64)。失败返回 None。"""
s = text.strip()
if s.startswith("data:") and "," in s:
s = s.split(",", 1)[1]
try:
return base64.b64decode(s)
except Exception:
return None
def _name(b) -> str:
"""把 field_name(可能是 bytes)安全解码为 str。"""
if isinstance(b, (bytes, bytearray)):
return b.decode("utf-8", "replace")
return str(b)
def extract_form_params(
content_type: str,
body: bytes,
static_dir: str,
public_base_url: str,
) -> dict:
"""解析 multipart/form-data 请求体,返回可安全写入日志的入参 dict。
- 标量字段(gender/hair_style/use_mask/prompt/erode_cm/...):utf-8 解码,超 500 字符截断。
- *_base64 字段:视为图片 → 解码存盘 → 记 {_image, source:"base64", size, url},绝不保留 base64 文本。
- file 字段:读字节存盘 → 记 {_image, source:"file", filename?, ctype?, size, url}。
- 非 multipart:记 {_content_type, _body_bytes}。
任何异常都吞掉并在结果里记 _parse_error,不抛。
"""
result: dict = {}
if not content_type or "multipart/form-data" not in content_type:
result["_content_type"] = content_type or ""
result["_body_bytes"] = len(body)
return result
try:
_mime, options = parse_options_header(content_type)
except Exception as e: # noqa: BLE001
result["_parse_error"] = f"options_header: {e}"
return result
if not options.get(b"boundary"):
result["_parse_error"] = "no boundary in content-type"
return result
def on_field(field) -> None:
try:
name = _name(field.field_name)
val = field.value
val_b = val if isinstance(val, (bytes, bytearray)) else (
val.encode("utf-8") if isinstance(val, str) else b""
)
if name.endswith("_base64"):
text = val_b.decode("utf-8", "replace")
data = _decode_b64_text(text)
if data:
url = save_image_bytes(data, static_dir, public_base_url)
result[name] = {"_image": True, "source": "base64",
"size": len(data), "url": url}
else:
result[name] = {"_image": True, "source": "base64",
"size": len(val_b), "url": None,
"_save_error": "decode failed"}
else:
text = val_b.decode("utf-8", "replace")
if len(text) > 500:
text = text[:500] + f"…(+{len(text) - 500} chars)"
result[name] = text
except Exception as e: # noqa: BLE001
logger.warning("on_field 解析失败 (%s): %s", getattr(field, "field_name", "?"), e)
def on_file(file) -> None:
try:
name = _name(file.field_name)
data = b""
fo = getattr(file, "file_object", None)
if fo is not None:
try:
fo.seek(0)
data = fo.read()
except Exception: # noqa: BLE001
data = b""
size = getattr(file, "size", None)
if size is None:
size = len(data)
entry: dict = {"_image": True, "source": "file", "size": size}
fn = getattr(file, "file_name", None)
if fn:
entry["filename"] = _name(fn)
ct = getattr(file, "content_type", None)
if ct:
entry["ctype"] = _name(ct)
entry["url"] = save_image_bytes(data, static_dir, public_base_url) if data else None
result[name] = entry
except Exception as e: # noqa: BLE001
logger.warning("on_file 解析失败 (%s): %s", getattr(file, "field_name", "?"), e)
try:
# headers: dict[str, bytes],仅需 Content-Type(含 boundary)。
# 不传 Content-Lengthparse_form 会从 BytesIO 读到 EOF。
parse_form(
{"Content-Type": content_type.encode("utf-8")},
io.BytesIO(body),
on_field,
on_file,
)
except Exception as e: # noqa: BLE001
result["_parse_error"] = f"parse_form: {e}"
return result
def summarize_for_log(
obj: Any,
max_str: int = 200,
max_list: int = 5,
max_depth: int = 6,
max_bytes: int = 8192,
) -> Any:
"""递归摘要:截断超长字符串 / 大数组 / 深层嵌套,保证日志短小。
- str 超 max_str → 截断并标注溢出字符数。
- list/tuple 超 max_list → 保留前 max_list 项 + "…(+N items)"
- 嵌套深度超 max_depth → "<…>"
- 摘要后 JSON 序列化若超 max_bytes → 退化为 {_truncated, size_bytes, keys} 桩。
"""
def _trunc(o: Any, depth: int) -> Any:
if depth > max_depth:
return "<…>"
if isinstance(o, bool) or o is None:
return o
if isinstance(o, (int, float)):
return o
if isinstance(o, str):
return o[:max_str] + f"…(+{len(o) - max_str} chars)" if len(o) > max_str else o
if isinstance(o, dict):
return {str(k): _trunc(v, depth + 1) for k, v in o.items()}
if isinstance(o, (list, tuple)):
n = len(o)
if n <= max_list:
return [_trunc(v, depth + 1) for v in o]
return [_trunc(v, depth + 1) for v in o[:max_list]] + [f"…(+{n - max_list} items)"]
return str(o)[:max_str]
summarized = _trunc(obj, 0)
try:
raw = json.dumps(summarized, ensure_ascii=False).encode("utf-8")
if len(raw) > max_bytes:
keys = list(summarized.keys()) if isinstance(summarized, dict) else None
return {"_truncated": True, "size_bytes": len(raw), "keys": keys}
except Exception: # noqa: BLE001
return {"_truncated": True}
return summarized
+231
View File
@@ -0,0 +1,231 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>API 耗时看板</title>
<style>
* { box-sizing: border-box; margin: 0; padding: 0; }
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; background: #f5f5f5; color: #333; }
.container { max-width: 1200px; margin: 0 auto; padding: 24px; }
h1 { font-size: 22px; margin-bottom: 4px; }
.subtitle { color: #888; font-size: 13px; margin-bottom: 18px; }
.nav { margin-bottom: 18px; }
.nav a { color: #2563eb; text-decoration: none; font-size: 13px; margin-right: 14px; }
.nav a:hover { text-decoration: underline; }
.toolbar { display: flex; align-items: center; gap: 12px; margin-bottom: 20px; flex-wrap: wrap; }
.toolbar label { font-size: 13px; font-weight: 600; color: #374151; }
.toolbar select, .toolbar button { padding: 8px 14px; border: 1px solid #d1d5db; border-radius: 8px; font-size: 14px; background: #fff; cursor: pointer; }
.toolbar select { min-width: 160px; }
.toolbar button { background: #2563eb; color: #fff; border: none; font-weight: 600; }
.toolbar button:hover { background: #1d4ed8; }
#loadingHint { font-size: 12px; color: #9ca3af; }
.stats-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(170px, 1fr)); gap: 14px; margin-bottom: 24px; }
.stat-card { background: #fff; border-radius: 12px; padding: 18px 20px; box-shadow: 0 1px 4px rgba(0,0,0,.06); }
.stat-card .value { font-size: 26px; font-weight: 700; color: #111827; }
.stat-card .label { font-size: 11px; color: #9ca3af; text-transform: uppercase; letter-spacing: .5px; margin-top: 4px; }
.stat-card.ok .value { color: #059669; }
.hourly-wrap { background: #fff; border-radius: 12px; padding: 16px 20px; box-shadow: 0 1px 4px rgba(0,0,0,.06); margin-bottom: 24px; }
.hourly-wrap h2 { font-size: 14px; color: #374151; margin-bottom: 12px; }
.hourly-bars { display: flex; align-items: flex-end; gap: 4px; height: 80px; }
.hourly-bar { flex: 1; background: #93c5fd; border-radius: 3px 3px 0 0; position: relative; min-height: 2px; }
.hourly-bar .cnt { position: absolute; top: -16px; left: 0; right: 0; text-align: center; font-size: 10px; color: #6b7280; }
.hourly-labels { display: flex; gap: 4px; margin-top: 4px; }
.hourly-labels span { flex: 1; text-align: center; font-size: 10px; color: #9ca3af; }
.ep-card { background: #fff; border-radius: 12px; box-shadow: 0 1px 4px rgba(0,0,0,.06); margin-bottom: 16px; overflow: hidden; }
.ep-header { padding: 14px 18px; background: #fafafa; border-bottom: 1px solid #f0f0f0; display: flex; align-items: center; gap: 10px; flex-wrap: wrap; }
.ep-badge { background: #2563eb; color: #fff; padding: 3px 12px; border-radius: 12px; font-size: 12px; font-weight: 700; white-space: nowrap; }
.ep-name { font-size: 15px; font-weight: 700; color: #111827; }
.ep-path { font-family: "SF Mono", "Fira Code", monospace; font-size: 12px; color: #9ca3af; }
.ep-success { margin-left: auto; font-size: 12px; }
.ep-body { padding: 16px 18px; }
.metrics-row { display: grid; grid-template-columns: repeat(auto-fit, minmax(110px, 1fr)); gap: 10px; margin-bottom: 4px; }
.metric { text-align: center; padding: 10px 4px; background: #f9fafb; border-radius: 8px; }
.metric .v { font-size: 17px; font-weight: 700; color: #1f2937; }
.metric .l { font-size: 10px; color: #9ca3af; margin-top: 2px; text-transform: uppercase; letter-spacing: .3px; }
.style-box { margin-top: 14px; background: #eff6ff; border: 1px solid #bfdbfe; border-radius: 10px; padding: 14px 16px; }
.style-box .title { font-size: 13px; font-weight: 700; color: #1e40af; margin-bottom: 8px; display: flex; align-items: center; gap: 6px; }
.style-box .formula { font-size: 12px; color: #1e3a8a; margin-bottom: 10px; font-family: "SF Mono", "Fira Code", monospace; background: #dbeafe; padding: 6px 10px; border-radius: 6px; display: inline-block; }
.style-metrics { display: flex; gap: 24px; margin-bottom: 10px; flex-wrap: wrap; }
.style-metrics .sm { }
.style-metrics .sm .v { font-size: 22px; font-weight: 700; color: #1d4ed8; }
.style-metrics .sm .l { font-size: 11px; color: #3b82f6; margin-top: 2px; }
.cluster-table { width: 100%; border-collapse: collapse; font-size: 12px; margin-top: 6px; }
.cluster-table th, .cluster-table td { padding: 5px 10px; text-align: center; border-bottom: 1px solid #dbeafe; }
.cluster-table th { color: #1e40af; font-weight: 700; }
.no-breakdown { font-size: 12px; color: #9ca3af; margin-top: 10px; }
.empty-state { text-align: center; padding: 60px 20px; color: #9ca3af; }
.footer-note { font-size: 12px; color: #9ca3af; margin-top: 20px; line-height: 1.8; }
.footer-note b { color: #6b7280; }
@media (max-width: 768px) {
.stats-grid { grid-template-columns: repeat(2, 1fr); }
.metrics-row { grid-template-columns: repeat(3, 1fr); }
}
</style>
</head>
<body>
<div class="container">
<h1>⏱️ API 耗时看板</h1>
<p class="subtitle">按天统计各接口调用次数与耗时分布,多发型接口额外拆解「单次换发型耗时」与「整体耗时」</p>
<div class="nav">
<a href="/">← 返回首页</a>
<a href="/admin/stats">实时请求统计</a>
<a href="/docs">API 文档</a>
</div>
<div class="toolbar">
<label for="dateSelect">选择日期</label>
<select id="dateSelect" onchange="loadDate()"></select>
<button onclick="loadDate()">🔄 刷新</button>
<span id="loadingHint"></span>
</div>
<div id="content"></div>
<div class="footer-note">
<div><b>说明:</b>「整个 API 耗时」为网关实测的请求总耗时(收到请求到返回响应),已排除网络传输本身的额外延迟。</div>
<div><b>单次换发型耗时</b>为估算值:C端生发 / C端生发v2 / 发际线PNG生成 这几个接口,对每个勾选的发型会串行调用一次 ComfyUI 重绘,
总耗时 ≈ 固定开销(人脸检测等,通常 &lt; 1s)+ 单次换发型耗时 × 勾选的发型数量。日志未记录具体勾选了几个发型,
这里基于当天所有请求耗时的分布,用迭代最小二乘回归拆解出「单次换发型耗时」与「固定开销」两个分量,供参考。</div>
</div>
</div>
<script>
function fmtSeconds(ms) {
if (ms == null) return '—';
const s = ms / 1000;
if (s < 60) return s.toFixed(1) + 's';
return (s / 60).toFixed(1) + 'min';
}
function fmtSecondsRaw(s) {
if (s == null) return '—';
if (s < 60) return s.toFixed(1) + 's';
return (s / 60).toFixed(1) + 'min';
}
async function loadDates() {
const r = await fetch('/admin/stats/dates');
const data = await r.json();
const sel = document.getElementById('dateSelect');
sel.innerHTML = '';
if (!data.dates || data.dates.length === 0) {
sel.innerHTML = '<option>暂无数据</option>';
return;
}
data.dates.forEach((d, i) => {
const opt = document.createElement('option');
opt.value = d;
opt.textContent = d + (i === 0 ? '(今天)' : '');
sel.appendChild(opt);
});
loadDate();
}
async function loadDate() {
const date = document.getElementById('dateSelect').value;
if (!date) return;
document.getElementById('loadingHint').textContent = '⏳ 加载中...';
try {
const r = await fetch('/admin/stats/daily?date=' + encodeURIComponent(date));
const data = await r.json();
render(data);
document.getElementById('loadingHint').textContent = '✅ 已更新 ' + new Date().toLocaleTimeString();
} catch (e) {
document.getElementById('loadingHint').textContent = '❌ 加载失败: ' + e.message;
}
}
function render(data) {
const el = document.getElementById('content');
if (data.error) {
el.innerHTML = '<div class="empty-state">⚠ ' + data.error + '</div>';
return;
}
if (!data.summary || data.summary.total === 0) {
el.innerHTML = '<div class="empty-state">📭 ' + data.date + ' 当天没有调用记录</div>';
return;
}
const s = data.summary;
let html = '';
// 汇总卡片
html += '<div class="stats-grid">' +
'<div class="stat-card"><div class="value">' + s.total + '</div><div class="label">当天请求总数</div></div>' +
'<div class="stat-card ok"><div class="value">' + s.success_rate + '%</div><div class="label">成功率</div></div>' +
'<div class="stat-card"><div class="value">' + fmtSeconds(s.avg_duration_ms) + '</div><div class="label">平均耗时/次</div></div>' +
'<div class="stat-card"><div class="value">' + fmtSecondsRaw(s.total_duration_seconds) + '</div><div class="label">累计耗时</div></div>' +
'<div class="stat-card"><div class="value">' + fmtSeconds(s.max_duration_ms) + '</div><div class="label">最长单次耗时</div></div>' +
'</div>';
// 按小时分布
if (data.hourly && data.hourly.length > 0) {
const maxCnt = Math.max(...data.hourly.map(h => h.count));
html += '<div class="hourly-wrap"><h2>📅 按小时请求量分布</h2><div class="hourly-bars">';
data.hourly.forEach(h => {
const pct = Math.max(4, Math.round(h.count / maxCnt * 100));
html += '<div class="hourly-bar" style="height:' + pct + '%"><div class="cnt">' + h.count + '</div></div>';
});
html += '</div><div class="hourly-labels">';
data.hourly.forEach(h => {
html += '<span>' + String(h.hour).padStart(2, '0') + '时</span>';
});
html += '</div></div>';
}
// 按接口
data.endpoints.forEach(ep => {
const badge = ep.interface_num != null ? ('接口' + ep.interface_num) : '未知接口';
html += '<div class="ep-card">' +
'<div class="ep-header">' +
'<span class="ep-badge">' + badge + '</span>' +
'<span class="ep-name">' + ep.interface_name + '</span>' +
'<span class="ep-path">' + ep.path + '</span>' +
'<span class="ep-success">调用 <b>' + ep.count + '</b> 次 · 成功率 <b style="color:' + (ep.success_rate >= 99 ? '#059669' : '#d97706') + '">' + ep.success_rate + '%</b></span>' +
'</div>' +
'<div class="ep-body">' +
'<div class="metrics-row">' +
'<div class="metric"><div class="v">' + fmtSeconds(ep.avg_duration_ms) + '</div><div class="l">平均耗时</div></div>' +
'<div class="metric"><div class="v">' + fmtSeconds(ep.median_duration_ms) + '</div><div class="l">中位数</div></div>' +
'<div class="metric"><div class="v">' + fmtSeconds(ep.p95_duration_ms) + '</div><div class="l">P95</div></div>' +
'<div class="metric"><div class="v">' + fmtSeconds(ep.min_duration_ms) + '</div><div class="l">最快</div></div>' +
'<div class="metric"><div class="v">' + fmtSeconds(ep.max_duration_ms) + '</div><div class="l">最慢</div></div>' +
'<div class="metric"><div class="v">' + fmtSecondsRaw(ep.total_duration_seconds) + '</div><div class="l">累计耗时</div></div>' +
'</div>';
if (ep.per_style) {
const ps = ep.per_style;
html += '<div class="style-box">' +
'<div class="title">✂️ 单次换发型耗时拆解(估算)</div>' +
'<div class="style-metrics">' +
'<div class="sm"><div class="v">' + ps.per_style_seconds.toFixed(1) + 's</div><div class="l">单次换发型耗时</div></div>' +
'<div class="sm"><div class="v">' + ps.fixed_overhead_seconds.toFixed(1) + 's</div><div class="l">固定开销</div></div>' +
'</div>' +
'<div class="formula">整体 API 耗时 ≈ ' + ps.fixed_overhead_seconds.toFixed(1) + 's + ' + ps.per_style_seconds.toFixed(1) + 's × 发型数量</div>' +
'<table class="cluster-table"><thead><tr><th>推断发型数</th><th>样本数</th><th>平均耗时</th></tr></thead><tbody>';
ps.clusters.forEach(c => {
html += '<tr><td>' + c.styles + '</td><td>' + c.count + '</td><td>' + c.avg_seconds.toFixed(1) + 's</td></tr>';
});
html += '</tbody></table></div>';
} else if (['/api/v1/hair/grow', '/api/v1/hair/grow-v2', '/api/v1/hairline/generate'].includes(ep.path)) {
html += '<div class="no-breakdown">️ 当天该接口耗时分布过于集中,样本不足以拆解出单次换发型耗时(换个天试试,或等积累更多调用)</div>';
}
html += '</div></div>';
});
el.innerHTML = html;
}
loadDates();
</script>
</body>
</html>
+58
View File
@@ -0,0 +1,58 @@
"""gateway.pool 全局串行槽单测:验证 max_global_concurrency=1 时同一时间只跑1个请求、
排队超时返回 False。
用 asyncio.run 套同步测试,避免引入 pytest-asyncio 依赖。
"""
import asyncio
from gateway import pool
_CFG = {"dispatch": {"max_global_concurrency": 1, "max_queue_wait_seconds": 590}}
_CFG_FAST_TIMEOUT = {"dispatch": {"max_global_concurrency": 1, "max_queue_wait_seconds": 0.2}}
def test_global_slot_serializes_to_one():
pool.init_global_slot(_CFG)
state = {"active": 0, "peak": 0}
async def task():
got = await pool.acquire_global_slot(_CFG)
assert got is True
state["active"] += 1
state["peak"] = max(state["peak"], state["active"])
await asyncio.sleep(0.03)
state["active"] -= 1
pool.release_global_slot()
async def run():
await asyncio.gather(*(task() for _ in range(4)))
asyncio.run(run())
assert state["peak"] == 1, f"期望峰值并发=1,实际={state['peak']}"
assert pool.get_pool_status()["global_busy"] == 0
def test_global_slot_timeout_returns_false():
pool.init_global_slot(_CFG_FAST_TIMEOUT)
async def run():
first = await pool.acquire_global_slot(_CFG_FAST_TIMEOUT)
assert first is True
# 槽已被占用,第二个应在 0.2s 内超时
second = await pool.acquire_global_slot(_CFG_FAST_TIMEOUT)
assert second is False
pool.release_global_slot()
asyncio.run(run())
assert pool.get_pool_status()["global_busy"] == 0
def test_get_pool_status_has_global_fields():
pool.init_global_slot(_CFG)
status = pool.get_pool_status()
assert "global_max" in status
assert "global_busy" in status
assert "global_waiting" in status
assert status["global_max"] == 1
+109
View File
@@ -0,0 +1,109 @@
"""gateway.reqlog 单测:multipart 入参解析(标量 + 图片存盘转 URL)+ 出参摘要。
不依赖网络 / GPU,纯函数级。
"""
import base64
import json
from gateway.reqlog import extract_form_params, summarize_for_log
# 1x1 透明 PNG(合法 magic byte
_PNG = (
b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01"
b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01"
b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82"
)
def _multipart(fields, files, boundary=b"----testbnd"):
"""手搓 multipart/form-data body。fields: {name:str}; files: {name:(fname,ctype,bytes)}."""
parts = []
for name, val in fields.items():
parts.append(b"--" + boundary + b"\r\n")
parts.append(b'Content-Disposition: form-data; name="' + name.encode() + b'"\r\n\r\n')
parts.append(val.encode("utf-8") + b"\r\n")
for name, (fname, ctype, data) in files.items():
parts.append(b"--" + boundary + b"\r\n")
parts.append(b'Content-Disposition: form-data; name="' + name.encode() + b'"; filename="'
+ fname.encode() + b'"\r\n')
parts.append(b"Content-Type: " + ctype.encode() + b"\r\n\r\n")
parts.append(data + b"\r\n")
parts.append(b"--" + boundary + b"--\r\n")
return b"".join(parts)
def test_extract_scalars_and_images(tmp_path):
b64 = "data:image/png;base64," + base64.b64encode(_PNG).decode()
body = _multipart(
fields={"gender": "female", "hair_style": "1,3", "use_mask": "True",
"prompt": "填充遮罩区域的头发", "image_base64": b64},
files={"image_file": ("x.png", "image/png", _PNG)},
)
ct = "multipart/form-data; boundary=----testbnd"
res = extract_form_params(ct, body, str(tmp_path), "https://test.local")
# 标量原样
assert res["gender"] == "female"
assert res["hair_style"] == "1,3"
assert res["use_mask"] == "True"
# 图片字段是元信息 dict(不是 base64 文本)
assert res["image_file"]["_image"] is True
assert res["image_file"]["source"] == "file"
assert res["image_file"]["filename"] == "x.png"
assert res["image_file"]["url"] is not None
assert res["image_base64"]["_image"] is True
assert res["image_base64"]["source"] == "base64"
assert res["image_base64"]["url"] is not None
# 图片已落盘
saved = [p.name for p in tmp_path.iterdir()]
assert any(n.startswith("in_") and n.endswith(".png") for n in saved)
# URL 指向正确路径
assert res["image_file"]["url"].startswith("https://test.local/static/annotations/in_")
# 关键:日志里绝不包含 base64 文本
blob = json.dumps(res, ensure_ascii=False)
assert b64.split(",", 1)[1][:40] not in blob
def test_extract_non_multipart(tmp_path):
res = extract_form_params("application/json", b'{"a":1}', str(tmp_path), "https://t.local")
assert res == {"_content_type": "application/json", "_body_bytes": 7}
def test_extract_empty_content_type(tmp_path):
res = extract_form_params("", b"hello", str(tmp_path), "https://t.local")
assert res["_body_bytes"] == 5
def test_summarize_truncates_long_values():
big = {
"landmarks": [[i, i] for i in range(500)],
"name": "x" * 1000,
"deep": {"a": {"b": {"c": {"d": {"e": {"f": {"g": 1}}}}}}},
}
sm = summarize_for_log(big)
# 长数组被截断 + 标注剩余数
assert isinstance(sm["landmarks"], list)
assert sm["landmarks"][-1].startswith("…(+")
assert len(sm["landmarks"]) == 6 # 5 项 + 1 个标注
# 长字符串被截断
assert "…(+" in sm["name"]
# 总体积受控
assert len(json.dumps(sm, ensure_ascii=False).encode()) <= 8192
def test_summarize_keeps_small_intact():
obj = {"code": 0, "data": {"hairline_images": [{"order": 1, "image_middle_url": "https://x/y.png"}]}}
assert summarize_for_log(obj) == obj
def test_summarize_byte_cap_stub():
# 构造大量小字段触发字节上限
obj = {f"k{i}": {f"j{j}": "abcdefghij" for j in range(50)} for i in range(50)}
sm = summarize_for_log(obj, max_bytes=2048)
assert isinstance(sm, dict) and sm.get("_truncated") is True
assert "size_bytes" in sm