UbuntuandClaude
bb9f55e93c
并发模型从「每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>
217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
"""请求日志辅助：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
|