From bb9f55e93c5f6916583c52ff03e421429485eb20 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Thu, 23 Jul 2026 22:56:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E5=85=A8=E5=B1=80=E4=B8=B2?= =?UTF-8?q?=E8=A1=8C=E5=8C=96(=E5=B9=B6=E5=8F=91=3D1)=20+=20=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E5=85=A5=E5=8F=82/=E5=87=BA=E5=8F=82/=E8=80=97?= =?UTF-8?q?=E6=97=B6=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 并发模型从「每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 --- gateway/app.py | 39 +++++ gateway/config.example.json | 2 + gateway/config.py | 7 +- gateway/forward.py | 50 +++++- gateway/logging_middleware.py | 262 ++++++++++++++++++++++++++++ gateway/pool.py | 70 +++++++- gateway/reqlog.py | 216 +++++++++++++++++++++++ static/api_timing_dashboard.html | 231 ++++++++++++++++++++++++ tests/test_gateway_serialization.py | 58 ++++++ tests/test_reqlog.py | 109 ++++++++++++ 10 files changed, 1037 insertions(+), 7 deletions(-) create mode 100644 gateway/reqlog.py create mode 100644 static/api_timing_dashboard.html create mode 100644 tests/test_gateway_serialization.py create mode 100644 tests/test_reqlog.py diff --git a/gateway/app.py b/gateway/app.py index eed7c3f..fb34f13 100644 --- a/gateway/app.py +++ b/gateway/app.py @@ -20,6 +20,8 @@ from fastapi.staticfiles import StaticFiles from gateway.config import load_config from gateway.logging_middleware import ( + get_available_dates, + get_daily_stats, get_stats, init_logging as _init_req_logging, request_logging_middleware, @@ -185,6 +187,9 @@ async def gateway_health(): "workers_total": status["total"], "workers_healthy": status["healthy"], "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", "docs": "/docs", "stats": "/admin/stats", + "timing_dashboard": "/static/api_timing_dashboard.html", "integration_guide": "/static/integration.html", "test_pages": { "if1_measure": "/static/test_interface1.html", @@ -470,6 +476,18 @@ async def admin_stats_json(): 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=["人脸分析"]) async def face_features( + request: Request, image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"), image_url: Optional[str] = Form(default=None, description="图片 URL"), 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, }) + # 入参日志（接口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 face_features import analyze_features diff --git a/gateway/config.example.json b/gateway/config.example.json index d47dd26..b76a439 100644 --- a/gateway/config.example.json +++ b/gateway/config.example.json @@ -16,6 +16,8 @@ "ark_api_key": "", "dispatch": { "per_worker_concurrency": 1, + "max_global_concurrency": 1, + "max_queue_wait_seconds": 590, "queue_wait_seconds": 30, "request_timeout_seconds": 600, "retry_on_failure": true, diff --git a/gateway/config.py b/gateway/config.py index 8198955..629075e 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -24,6 +24,8 @@ DEFAULTS = { }, "dispatch": { "per_worker_concurrency": 1, + "max_global_concurrency": 1, + "max_queue_wait_seconds": 590, "queue_wait_seconds": 30, "request_timeout_seconds": 600, "retry_on_failure": True, @@ -105,12 +107,15 @@ def load_config() -> dict: logger.info( "配置加载完成 | 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["public_base_url"], cfg["health_check"]["interval_seconds"], cfg["dispatch"]["request_timeout_seconds"], cfg["dispatch"]["queue_wait_seconds"], + cfg["dispatch"]["max_global_concurrency"], + cfg["dispatch"]["max_queue_wait_seconds"], ) _config_cache = cfg diff --git a/gateway/forward.py b/gateway/forward.py index 2e6aa40..db17b07 100644 --- a/gateway/forward.py +++ b/gateway/forward.py @@ -19,10 +19,13 @@ from fastapi.responses import JSONResponse from gateway.pool import ( NoWorkerAvailable, + acquire_global_slot, acquire_worker, mark_worker_unhealthy, + release_global_slot, release_worker, ) +from gateway.reqlog import extract_form_params logger = logging.getLogger("gateway.forward") @@ -143,17 +146,56 @@ async def proxy_request(request: Request, path: str) -> JSONResponse: """ from gateway.config import 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"] token = cfg["shared_password"] request_timeout = dispatch_cfg["request_timeout_seconds"] 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"] static_dir = cfg["static_dir"] - # --- 1. 读取客户端请求体（原始字节，不做解析） --- - body = await request.body() - # --- 2. 获取 worker（最多重试 max_retries+1 次） --- attempts = max_retries + 1 last_error_response = None diff --git a/gateway/logging_middleware.py b/gateway/logging_middleware.py index 2003c21..4184856 100644 --- a/gateway/logging_middleware.py +++ b/gateway/logging_middleware.py @@ -14,6 +14,8 @@ from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional +from gateway.reqlog import summarize_for_log + logger = logging.getLogger("gateway.logging_middleware") @@ -34,6 +36,8 @@ class RequestLogEntry: worker: str = "" # 处理请求的 worker URL（空串表示网关本地处理） response_code: Optional[int] = None # 响应 JSON 中的 code 字段 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", ""), response_code=data.get("response_code"), request_id=data.get("request_id"), + request_params=data.get("request_params"), + response_data=data.get("response_data"), ) _buffer.append(entry) count += 1 @@ -248,6 +254,7 @@ async def request_logging_middleware(request, call_next): # 提取响应 body 并解析业务字段（仅 JSON 响应） response_code = None request_id = None + response_data = None content_type = response.headers.get("content-type", "") 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) response_code = data.get("code") request_id = data.get("request_id") + response_data = summarize_for_log(data) except (json.JSONDecodeError, UnicodeDecodeError): pass @@ -274,6 +282,9 @@ async def request_logging_middleware(request, call_next): media_type=response.media_type, ) + # 入参（由 forward.py / 接口4 handler 挂到 request.state） + request_params = getattr(request.state, "log_request_params", None) + # 记录 now = datetime.datetime.utcnow() entry = RequestLogEntry( @@ -287,6 +298,8 @@ async def request_logging_middleware(request, call_next): worker=worker_url, response_code=response_code, request_id=request_id, + request_params=request_params, + response_data=response_data, ) _buffer.append(entry) @@ -366,6 +379,8 @@ def get_stats() -> Dict[str, Any]: "worker": e.worker, "response_code": e.response_code, "request_id": e.request_id, + "request_params": e.request_params, + "response_data": e.response_data, } for e in recent_100 ] @@ -404,3 +419,250 @@ def get_stats() -> Dict[str, Any]: "recent": recent, "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, + } diff --git a/gateway/pool.py b/gateway/pool.py index e2e2e6d..9bf790e 100644 --- a/gateway/pool.py +++ b/gateway/pool.py @@ -50,6 +50,14 @@ _pool_condition: Optional[asyncio.Condition] = None _health_task: Optional[asyncio.Task] = 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() _shutdown_event = asyncio.Event() + # 全局串行槽（独立于 worker busy 标志） + init_global_slot(cfg) + # 立即做一轮健康检查以快速上线 hc_cfg = cfg["health_check"] token = cfg["shared_password"] @@ -206,6 +217,56 @@ async def shutdown_pool() -> None: 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: """返回当前池状态（供 /gateway-health 使用）。""" + global_info = { + "global_max": _global_max, + "global_busy": _global_busy, + "global_waiting": _global_waiting, + } if not _workers: - return {"total": 0, "healthy": 0, "busy": 0} + return {"total": 0, "healthy": 0, "busy": 0, **global_info} total = len(_workers) healthy = sum(1 for w in _workers.values() if w.online) 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} diff --git a/gateway/reqlog.py b/gateway/reqlog.py new file mode 100644 index 0000000..17bdeb6 --- /dev/null +++ b/gateway/reqlog.py @@ -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-Length：parse_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 diff --git a/static/api_timing_dashboard.html b/static/api_timing_dashboard.html new file mode 100644 index 0000000..e71b743 --- /dev/null +++ b/static/api_timing_dashboard.html @@ -0,0 +1,231 @@ + + + + + +API 耗时看板 + + + +
+

⏱️ API 耗时看板

+

按天统计各接口调用次数与耗时分布，多发型接口额外拆解「单次换发型耗时」与「整体耗时」

+
+ ← 返回首页 + 实时请求统计 + API 文档 +
+ +
+ + + + +
+ +
+ +
+
说明：「整个 API 耗时」为网关实测的请求总耗时（收到请求到返回响应），已排除网络传输本身的额外延迟。
+
单次换发型耗时为估算值：C端生发 / C端生发v2 / 发际线PNG生成 这几个接口，对每个勾选的发型会串行调用一次 ComfyUI 重绘， + 总耗时 ≈ 固定开销（人脸检测等，通常 < 1s）+ 单次换发型耗时 × 勾选的发型数量。日志未记录具体勾选了几个发型， + 这里基于当天所有请求耗时的分布，用迭代最小二乘回归拆解出「单次换发型耗时」与「固定开销」两个分量，供参考。
+
+
+ + + + diff --git a/tests/test_gateway_serialization.py b/tests/test_gateway_serialization.py new file mode 100644 index 0000000..df09a8a --- /dev/null +++ b/tests/test_gateway_serialization.py @@ -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 diff --git a/tests/test_reqlog.py b/tests/test_reqlog.py new file mode 100644 index 0000000..12c5436 --- /dev/null +++ b/tests/test_reqlog.py @@ -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