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>
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
"""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
|