diff --git a/config.toml b/config.toml index a61e8bf..bedad03 100644 --- a/config.toml +++ b/config.toml @@ -42,6 +42,9 @@ reload = true file = "logs" format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" level = "DEBUG" +# access 日志采样率:0.1 = 只记录 10% 的成功请求(错误 >=400 始终全量) +# 1.0 = 全量记录;配合异步日志队列降低高吞吐下主线程开销 +access_sample_rate = 0.1 [development.cors] allow_credentials = true diff --git a/pyproject.toml b/pyproject.toml index 246cc3c..0494471 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,16 +19,17 @@ dependencies = [ "tomli", "typing-extensions", "uvicorn", - "volcengine-python-sdk[ark]>=5.0.41", + "volcengine-python-sdk[ark]", "chongming-config", "chongming-password", "chongming-worker", - "amqtt>=0.11.3", - "aiofiles>=25.1.0", + "amqtt", + "aiofiles", "bytedance", "chongming_stream_manager", - "uvloop>=0.22.1", + "uvloop", "chongming-mqtt", + "httptools", ] description = "基于 modules-bank 的 fastapi 打包脚手架" name = "chongming-zenoh" diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index d913f85..0eb071f 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -9,18 +9,20 @@ chongming FastAPI 应用入口 import time import asyncio +import itertools from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse +from fastapi.middleware.gzip import GZipMiddleware from fastapi.middleware.cors import CORSMiddleware from chongming_config import get_config from .core.logger import get_logger from .core.static_files import SVFSStaticFiles -from .core import lifespan_setup -from .routes import storage_router +from .core import lifespan_setup, call_stats +from .routes import storage_router, stats_router config = get_config() @@ -112,6 +114,9 @@ async def lifespan(app: FastAPI): if worker_tasks: await asyncio.gather(*worker_tasks, return_exceptions=True) + # 4.5 取消后台统计/审计任务 + await call_stats.shutdown() + # 5. 关闭 Zenoh 会话 session.close() logger.info("Zenoh 会话已关闭") @@ -131,6 +136,24 @@ app = FastAPI( ) app.add_middleware(CORSMiddleware, **cors) +# GZip 压缩响应(<500B 不压缩;starlette 默认排除 text/event-stream, +# 保证 SSE 流式端点即时性;>=128KB 的响应在独立线程压缩,不阻塞事件循环) +app.add_middleware(GZipMiddleware) + + +# access 日志采样率(logging.access_sample_rate,0.1 = 只记录 10% 的成功请求; +# 错误 >=400 始终全量记录,保证可追溯;1.0/缺省 = 全量) +ACCESS_SAMPLE_RATE = float(config[env].get("logging", {}).get("access_sample_rate", 1.0)) +# 成功请求采样计数器(每进程独立;错误请求不消耗计数) +_access_log_counter = itertools.count() + + +def _should_log_access(status_code: int, sample_rate: float = ACCESS_SAMPLE_RATE) -> bool: + """错误全量;成功按采样率(采样为确定性计数,无随机数开销)""" + if status_code >= 400 or sample_rate >= 1.0: + return True + denom = max(1, int(round(1.0 / sample_rate))) + return next(_access_log_counter) % denom == 0 @app.middleware("http") @@ -144,6 +167,11 @@ async def access_log_middleware(request: Request, call_next): - 请求处理时长(毫秒) - 客户端IP地址 + 高吞吐优化: + - 采样:成功请求按 access_sample_rate 记录,错误(>=400)全量 + - 异步:% 参数化消息 + 队列日志(QueueHandler),格式化与文件 I/O + 在后台监听线程执行,主线程仅入队 + 参数: request: FastAPI请求对象 call_next: 下一个中间件或路由处理函数 @@ -156,11 +184,12 @@ async def access_log_middleware(request: Request, call_next): duration = (time.perf_counter() - start) * 1000 client = request.client.host if request.client else "-" - logger.info( - f"{request.method} {request.url.path} status={response.status_code} " - f"duration={duration:.2f}ms client={client}", - extra={"status_code": response.status_code}, - ) + if _should_log_access(response.status_code): + logger.info( + "%s %s status=%s duration=%.2fms client=%s", + request.method, request.url.path, response.status_code, duration, client, + extra={"status_code": response.status_code}, + ) return response @@ -240,3 +269,4 @@ if vfs_db_path := config[env].get("file_system", {}).get("path", None): app.mount("/static", static_files_handler, name="static") app.include_router(storage_router) +app.include_router(stats_router) diff --git a/src/chongming/app/core/auth.py b/src/chongming/app/core/auth.py index 99fd896..262e1f9 100644 --- a/src/chongming/app/core/auth.py +++ b/src/chongming/app/core/auth.py @@ -1,20 +1,64 @@ +import hashlib +import time + from fastapi import Depends, HTTPException, status, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from chongming_jwt import JWT +from chongming_cache import Cache from chongming_config import get_config +import orjson security_scheme = HTTPBearer(auto_error=False) +# JWT 解析共享缓存:与 WorkerManager 共用 [cache] 配置段。 +# 默认 diskcache(本地);横向拓展时 config.toml 切 backend = "redis", +# 多网关实例共享同一 Redis 即可全局只解密一次,代码零改动。 +# 注意:缓存存的是"已验证结果",信任模型与 worker 回复缓存一致 +# (缓存基础设施可信;Redis 请配置密码并置于内网)。 +_jwt_cache = Cache.from_config(get_config()["cache"]) +# 条目 TTL(秒):控制存储占用与密钥轮换后的自然清理; +# 安全性由命中时的 exp 复核保证,TTL 不承担过期判断 +_JWT_CACHE_TTL = 60 + + +def _cache_key(token: str, secret: str, algorithm: str) -> str: + """缓存 key:token 内容 + 密钥/算法指纹,不同密钥/算法不串用""" + return "jwt:" + hashlib.sha256(f"{secret}:{algorithm}:{token}".encode()).hexdigest() + + +def _cached_decode(token: str, secret: str, algorithm: str) -> tuple: + """JWT 解密结果缓存((payload, exp)),共享缓存避免高频请求重复密码学运算。 + + - 命中且未过期 -> 直接返回,跳过 RSA/ECDSA 验证与 JSON 解析 + - 命中但 exp 已过 -> 直接拒绝,不重解密(条目保留至 TTL 到期, + 过期 token 的重复请求零密码学开销;缓存不延长 token 有效期) + - 未命中 -> 完整解码后写入共享缓存(TTL=60s) + - decode 失败(无效 token)不缓存,每次都完整验证后拒绝 + """ + key = _cache_key(token, secret, algorithm) + entry = _jwt_cache.get(key) + if entry is not None: + payload = entry["payload"] + exp = entry.get("exp", 0) + if not exp or exp >= time.time(): + return payload, exp + raise ValueError("Token 已过期") + jwt = JWT(secret, algorithm) + payload_str = jwt.decode(token) + payload = orjson.loads(payload_str.encode("utf-8")) + exp = payload.get("exp", 0) + _jwt_cache.set(key, {"payload": payload, "exp": exp}, expire=_JWT_CACHE_TTL) + return payload, exp + def jwt_auth_dependency( credentials: HTTPAuthorizationCredentials = Depends(security_scheme) ) -> dict: """JWT 校验依赖,返回解码后的 payload,校验失败抛出 401""" - config =get_config() + config = get_config() secret = config["security"]["secret_key"] algorithm = config["security"]["algorithm"] - jwt = JWT(secret, algorithm) if not credentials: raise HTTPException( @@ -24,9 +68,8 @@ def jwt_auth_dependency( ) token = credentials.credentials try: - payload_str = jwt.decode(token) - import json - payload = json.loads(payload_str) + payload, _ = _cached_decode(token, secret, algorithm) + # 缓存后端为 pickle 序列化,返回的是独立对象,调用方修改不影响缓存 return payload except Exception as e: raise HTTPException( diff --git a/src/chongming/app/core/call_stats.py b/src/chongming/app/core/call_stats.py new file mode 100644 index 0000000..b1e100e --- /dev/null +++ b/src/chongming/app/core/call_stats.py @@ -0,0 +1,150 @@ +""" +调用统计与审计日志(后台任务执行) + +route_generator 端点在 call_raw 返回后仅做非关键后置操作,全部经 +spawn() 以 fire-and-forget 后台任务执行,端点立即返回,不增加接口 +响应延迟、不占用请求协程资源: + 1. 更新内存调用统计(per worker_type/subject/status 计数 + 延迟聚合 + 最近调用缓冲) + 2. 写审计日志(logs/audit-YYYY-MM-DD.log,复用 logger 按日切分 + 大小轮转设施) + +record_call 内部全兜底:统计或审计任一步失败仅记 warning,绝不影响响应。 +统计为进程内视角(gunicorn 多 worker 下各进程独立聚合)。 + +后台任务随应用生命周期管理:lifespan 关闭阶段调用 shutdown() 统一取消。 +""" +import asyncio +import datetime +import orjson +import time +from collections import deque +from typing import Any, Coroutine, Dict + +from .logger import get_logger + +logger = get_logger("call_stats") +# 审计日志独立文件(audit-YYYY-MM-DD.log),与业务日志分离便于检索 +audit_logger = get_logger("audit") + +# 最近调用环形缓冲上限 +MAX_RECENT = 1000 + +# 后台任务集合(fire-and-forget,shutdown 时统一取消) +_tasks: set[asyncio.Task] = set() + + +def spawn(coro: Coroutine) -> asyncio.Task: + """以 fire-and-forget 方式创建后台任务并纳入跟踪(shutdown 时取消)""" + task = asyncio.create_task(coro) + _tasks.add(task) + task.add_done_callback(_tasks.discard) + return task + + +async def shutdown() -> None: + """取消所有后台统计/审计任务(应用关闭时调用)""" + tasks = list(_tasks) + if not tasks: + return + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + +class CallStats: + """内存调用统计:计数 + 延迟聚合 + 最近调用环形缓冲。 + + asyncio 单线程事件循环内访问,无需加锁。 + """ + + def __init__(self, max_recent: int = MAX_RECENT): + self._counts: Dict[tuple, int] = {} + self._latency: Dict[str, Dict[str, float]] = {} + self._recent: deque = deque(maxlen=max_recent) + + def record(self, entry: Dict[str, Any]) -> None: + key = (entry["worker_type"], entry["subject"], entry["status"]) + self._counts[key] = self._counts.get(key, 0) + 1 + + wt = entry["worker_type"] + lat = self._latency.setdefault(wt, {"total": 0.0, "count": 0, "max": 0.0}) + dur = entry.get("duration_ms", 0.0) + lat["total"] += dur + lat["count"] += 1 + if dur > lat["max"]: + lat["max"] = dur + + self._recent.append(entry) + + def summary(self) -> Dict[str, Any]: + """聚合快照:总量、per worker/subject/status 计数、延迟、最近调用""" + calls = [ + {"worker_type": wt, "subject": subj, "status": status, "count": count} + for (wt, subj, status), count in sorted(self._counts.items()) + ] + latency = [ + { + "worker_type": wt, + "count": lat["count"], + "avg_ms": round(lat["total"] / lat["count"], 2) if lat["count"] else 0.0, + "max_ms": round(lat["max"], 2), + } + for wt, lat in sorted(self._latency.items()) + ] + return { + "total_calls": sum(self._counts.values()), + "calls": calls, + "latency": latency, + "recent": list(self._recent)[-50:], + } + + +_stats = CallStats() + + +def summary() -> Dict[str, Any]: + """全局统计快照(供 GET /stats 端点使用)""" + return _stats.summary() + + +def _format_audit(entry: Dict[str, Any]) -> str: + """审计日志单行 key=value 格式,便于 grep 检索""" + ts = entry.get("ts") or time.time() + ts_str = datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + user = entry.get("user") + if isinstance(user, dict): + # JWT payload(auth.py store_user_to_request 注入),orjson 默认紧凑单行输出 + user_str = orjson.dumps(user).decode() + elif isinstance(user, str): + user_str = user + else: + user_str = "-" + parts = [ + "AUDIT", + f"ts={ts_str}", + f"worker={entry.get('worker_type', '-')}", + f"subject={entry.get('subject', '-')}", + f"method={entry.get('method', '-')}", + f"path={entry.get('path', '-')}", + f"user={user_str}", + f"client={entry.get('client') or '-'}", + f"status={entry.get('status', '-')}", + f"duration_ms={entry.get('duration_ms', 0):.3f}", + f"req_bytes={entry.get('req_bytes', 0)}", + f"resp_bytes={entry.get('resp_bytes', 0)}", + ] + return " ".join(parts) + + +async def record_call(entry: Dict[str, Any]) -> None: + """后台执行体:更新内存统计 + 写审计日志。 + + 整体兜底:任一步失败仅记 warning,绝不向端点冒泡。 + """ + try: + _stats.record(entry) + except Exception as e: # pragma: no cover - 防御性兜底 + logger.warning("调用统计记录失败: %s", e) + try: + audit_logger.info(_format_audit(entry)) + except Exception as e: # pragma: no cover - 防御性兜底 + logger.warning("审计日志写入失败: %s", e) diff --git a/src/chongming/app/core/logger.py b/src/chongming/app/core/logger.py index 1c62671..b8a1129 100644 --- a/src/chongming/app/core/logger.py +++ b/src/chongming/app/core/logger.py @@ -1,8 +1,12 @@ import re +import os +import atexit +import queue import logging import datetime from pathlib import Path -from logging.handlers import RotatingFileHandler +from typing import List, Optional +from logging.handlers import RotatingFileHandler, QueueHandler, QueueListener from chongming_config import get_config @@ -177,6 +181,103 @@ class ExcludeWatchFilesFilter(logging.Filter): return not record.name.startswith(self._filter_starts_with_tulpe) +# ---- 异步日志队列(QueueHandler + QueueListener)---- +# 主线程/事件循环只做"入队"(LogRecord 创建 + put,µs 级),文件/控制台 I/O +# 由后台监听线程执行,日志记录不再阻塞业务协程。 +# 队列满时丢弃新日志(日志系统故障不影响业务);错误日志(>=400)不受采样影响。 + +_LOG_QUEUE_SIZE = 10000 + + +class _AsyncQueueHandler(QueueHandler): + """非阻塞入队 + 惰性格式化。 + + - 队列满时丢弃(而不是抛 queue.Full 打断业务) + - 不预格式化:% 样式消息的格式化推迟到监听线程(formatter 执行), + 主线程只承担 LogRecord 创建与入队 + - queue 经模块全局读取:gunicorn preload/fork 后自动指向子进程的新队列 + """ + + @property + def queue(self): + return _log_queue + + def __init__(self): + # queue 由 property 动态提供(fork 后自动指向新队列)。 + # 只初始化 Handler 基类(level/filters 等),不走 QueueHandler.__init__ + # 的 self.queue = queue 赋值(会与 property 冲突) + logging.Handler.__init__(self) + + def enqueue(self, record): + try: + self.queue.put_nowait(record) + except queue.Full: + pass # 队列满:丢弃该日志 + + def prepare(self, record): + return record # 格式化留给监听线程的 formatter + + +class _RegistryListener(QueueListener): + """handler 可动态注册的 QueueListener(每个 log_name 有独立 file handler)""" + + def __init__(self, q: queue.Queue): + super().__init__(q) + self.handlers: list = [] # 可变列表,运行期可追加 + + def register(self, *handlers) -> None: + self.handlers.extend(handlers) + + +_log_queue: queue.Queue = queue.Queue(maxsize=_LOG_QUEUE_SIZE) +_listener: Optional[_RegistryListener] = None +# 全部真实 handler(文件/控制台),供 fork 后重建监听线程 +_listener_handlers: List[logging.Handler] = [] + + +def _start_listener() -> None: + global _listener + _listener = _RegistryListener(_log_queue) + _listener.register(*_listener_handlers) + _listener.start() + + +def _after_fork_child() -> None: + """gunicorn preload/fork 场景:线程不跨 fork,子进程重建队列与监听线程。 + + 队列重建避免 fork 时锁状态不可用导致的死锁;旧队列中的记录在 fork + 瞬间丢失(启动窗口内可接受)。 + """ + global _log_queue + _log_queue = queue.Queue(maxsize=_LOG_QUEUE_SIZE) + _start_listener() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_after_fork_child) + + +def shutdown_logging() -> None: + """停止监听线程并冲刷剩余日志(进程退出 / 应用关闭时调用),幂等。 + + QueueListener.stop() 停止后会将 _thread 置 None,重复调用需跳过; + 幂等保证 atexit 与显式调用共存。 + """ + global _listener + if _listener is not None: + if _listener._thread is not None: + _listener.stop() # flush=True:处理完队列中已有记录再停止 + _listener = None + for h in _listener_handlers: + try: + h.close() + except Exception: # pragma: no cover - 关闭失败不影响退出 + pass + + +atexit.register(shutdown_logging) + + _configured_loggers = set() @@ -237,9 +338,14 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"): console_handler.setFormatter(color_formatter) console_handler.addFilter(ExcludeWatchFilesFilter()) - # 添加处理器到根日志记录器 - logger.addHandler(file_handler) - logger.addHandler(console_handler) + # 真实 handler(文件/控制台)挂到后台监听线程;logger 上只挂队列 handler, + # 日志调用在主线程仅入队,I/O 由监听线程执行 + _listener_handlers.extend([file_handler, console_handler]) + if _listener is None: + _start_listener() + else: + _listener.register(file_handler, console_handler) + logger.addHandler(_AsyncQueueHandler()) print( f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} " diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index 6fab0b9..e6112c7 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/src/chongming/app/core/mqtt_bridge.py @@ -1,6 +1,6 @@ """MQTT Bridge — 纯客户端模式,Broker 由子进程管理""" import asyncio -import json +import orjson import re from typing import Optional from urllib.parse import quote @@ -165,8 +165,8 @@ class MQTTBridge: t: str = m.topic p = m.data # type: ignore try: - pl = json.loads(p.decode("utf-8")) - except json.JSONDecodeError: + pl = orjson.loads(p) + except orjson.JSONDecodeError: return # 信封协议: {"meta": {...}, "data": {...}} @@ -243,8 +243,9 @@ class MQTTBridge: async def _pub(self, topic, data): try: - pd = json.dumps(data, ensure_ascii=False, default=str) - await self._client.publish(topic, pd.encode("utf-8"), QOS_1) # type: ignore + # data 均为 JSON 可序列化类型(worker 回复解析结果 / {"error": ...}), + # orjson.dumps 直接产出 UTF-8 字节,免去 str -> encode 两步 + await self._client.publish(topic, orjson.dumps(data), QOS_1) # type: ignore except Exception as e: logger.error(f"发布失败 [{topic}]: {e}") diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py index eae772b..21924f5 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -7,6 +7,8 @@ """ from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple from inspect import Parameter, Signature +import asyncio +import time import orjson from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends @@ -18,6 +20,7 @@ from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_ from .worker_manager import WorkerManager, StreamIdleTimeoutError from .logger import get_logger from .auth import store_user_to_request +from .call_stats import spawn, record_call logger = get_logger("route_generator") @@ -129,6 +132,7 @@ def register_route_group( prefix: str = "", extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, raw_body_models: Optional[Dict[Tuple[str, str], str]] = None, + response_models: Optional[Dict[Tuple[str, str], str]] = None, ) -> List[Any]: """ 为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。 @@ -145,6 +149,9 @@ def register_route_group( extra_schemas: 额外 schema 收集 dict(GET 请求模型注入 OpenAPI 用) raw_body_models: (path, method) -> 请求模型名 的收集 dict (validate=false 端点:请求体文档经 requestBody $ref 注入,见 _inject_schemas_to_openapi) + response_models: (path, method) -> 响应模型名 的收集 dict + (端点注册不传 response_model 以砍掉注册期模型反射;响应文档经 + responses[200].content $ref 手动注入,见 _inject_schemas_to_openapi) Returns: List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表 @@ -152,14 +159,17 @@ def register_route_group( """ router = APIRouter(prefix=prefix) for route_conf in routes: - _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models) + _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models, response_models) - # raw_body_models 记录的 (path, method) 为端点声明路径(不含 prefix), + # raw_body_models / response_models 记录的 (path, method) 为端点声明路径(不含 prefix), # 而 OpenAPI paths 键为最终挂载路径(prefix + path),这里统一补前缀 - if raw_body_models is not None and prefix: - for key in list(raw_body_models.keys()): + if prefix: + for key in list((raw_body_models or {}).keys()): path, method = key raw_body_models[(prefix + path, method)] = raw_body_models.pop(key) + for key in list((response_models or {}).keys()): + path, method = key + response_models[(prefix + path, method)] = response_models.pop(key) before = set(id(r) for r in app.router.routes) app.include_router(router) @@ -193,6 +203,8 @@ def register_dynamic_routes( extra_schemas: Dict[str, type[BaseModel]] = {} # validate=false 端点的 (path, method) -> 请求模型名(OpenAPI requestBody 注入用) raw_body_models: Dict[Tuple[str, str], str] = {} + # 非流式端点的 (path, method) -> 响应模型名(OpenAPI responses[200] $ref 注入用) + response_models: Dict[Tuple[str, str], str] = {} # 扫描所有 worker 子目录(每个子目录名 = worker_type) for worker_dir in sorted(workers_base.iterdir()): @@ -205,16 +217,16 @@ def register_dynamic_routes( if not routes: continue - register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models) + register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models, response_models) # 将额外 schema 注册到 OpenAPI 的 components/schemas if extra_schemas: - _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models) + _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models, lambda: response_models) logger.info(f"动态路由注册完成,worker 配置目录: {WORKER_CONFIG_DIR}") -def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None): +def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None, response_provider=None): """ 将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。 @@ -228,6 +240,9 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider raw_body_provider: 可选无参 callable,返回 {(path, method): model_name} 字典。 validate=false 端点未声明 body 参数(FastAPI 不会自动生成 requestBody), 这里按 (path, method) 补注入 requestBody $ref,保持 OpenAPI 请求体文档完整。 + response_provider: 可选无参 callable,返回 {(path, method): model_name} 字典。 + 端点注册未传 response_model(见 _register_single_route:砍掉注册期模型反射), + 这里按 (path, method) 补注入 responses[200] 的 $ref,保持响应文档完整。 """ original_openapi = app.openapi @@ -255,12 +270,55 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider } }, } + # 响应模型:补注入 responses[200] 的 $ref(保留 description,仅加 content) + if response_provider is not None: + for (path, method), model_name in response_provider().items(): + op = schema.get("paths", {}).get(path, {}).get(method.lower()) + if op is not None: + op.setdefault("responses", {}).setdefault("200", {}) + op["responses"]["200"].setdefault( + "content", {} + ).setdefault("application/json", {})["schema"] = { + "$ref": f"#/components/schemas/{model_name}" + } app.openapi_schema = schema return schema app.openapi = patched_openapi +def _spawn_call_record( + request: Request, + worker_type: str, + subject: str, + rc: RouteConfig, + status: int, + duration_ms: float, + req_bytes: int = 0, + resp_bytes: int = 0, +) -> None: + """ + 以后台任务记录一次调用(统计 + 审计),非关键后置操作。 + + 通过 spawn() 以 fire-and-forget 方式执行,端点立即返回, + 不增加接口响应延迟;后台失败由 record_call 内部兜底。 + """ + entry = { + "ts": time.time(), + "worker_type": worker_type, + "subject": subject, + "method": rc.method, + "path": request.url.path, + "user": getattr(request.state, "user", None), + "client": request.client.host if request.client else None, + "status": status, + "duration_ms": round(duration_ms, 3), + "req_bytes": req_bytes, + "resp_bytes": resp_bytes, + } + spawn(record_call(entry)) + + def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str, RequestModel: type[BaseModel]): """创建用于 Body 请求(POST/PUT/PATCH)的端点函数""" path = rc.path @@ -273,6 +331,8 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t request: Request, body: RequestModel, # type: ignore ) -> Any: + start: Optional[float] = None + payload_bytes = b"" try: # 信封协议: {"meta": {...}, "data": {...}} meta = {"method": http_method} @@ -280,19 +340,36 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t if user_info: meta["user_info"] = user_info payload = {"meta": meta, "data": body.model_dump()} + payload_bytes = orjson.dumps(payload) + start = time.perf_counter() result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, - payload=orjson.dumps(payload), + payload=payload_bytes, timeout=timeout, cacheable=cacheable, ) + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 200, duration_ms, + req_bytes=len(payload_bytes), resp_bytes=len(result), + ) # 回复为 worker 原始字节,零解析直接转发(response_model 仅用于 OpenAPI 文档) return Response(content=result, media_type="application/json") except RuntimeError as e: + duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0 + _spawn_call_record( + request, worker_type, subject, rc, 503, duration_ms, + req_bytes=len(payload_bytes), + ) logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) except Exception as e: + duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0 + _spawn_call_record( + request, worker_type, subject, rc, 500, duration_ms, + req_bytes=len(payload_bytes), + ) logger.error(f"路由 {path} 调用异常: {e}") raise HTTPException(status_code=500, detail="内部服务器错误") @@ -315,6 +392,8 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work cacheable = rc.cacheable async def endpoint(request: Request) -> Any: + start: Optional[float] = None + payload = b"" try: meta = {"method": http_method} user_info = getattr(request.state, "user", None) @@ -323,6 +402,7 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work raw = await request.body() data_part = raw.strip() or b"{}" payload = b'{"meta":' + orjson.dumps(meta) + b',"data":' + data_part + b'}' + start = time.perf_counter() result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, @@ -330,11 +410,26 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work timeout=timeout, cacheable=cacheable, ) + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 200, duration_ms, + req_bytes=len(payload), resp_bytes=len(result), + ) return Response(content=result, media_type="application/json") except RuntimeError as e: + duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0 + _spawn_call_record( + request, worker_type, subject, rc, 503, duration_ms, + req_bytes=len(payload), + ) logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) except Exception as e: + duration_ms = (time.perf_counter() - start) * 1000 if start else 0.0 + _spawn_call_record( + request, worker_type, subject, rc, 500, duration_ms, + req_bytes=len(payload), + ) logger.error(f"路由 {path} 调用异常: {e}") raise HTTPException(status_code=500, detail="内部服务器错误") @@ -368,20 +463,36 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_ user_info = getattr(request.state, "user", None) if user_info: meta["user_info"] = user_info - payload = {"meta": meta, "data": {}} + payload_bytes = orjson.dumps({"meta": meta, "data": {}}) + start = time.perf_counter() try: result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, - payload=orjson.dumps(payload), + payload=payload_bytes, timeout=timeout, cacheable=cacheable, ) + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 200, duration_ms, + req_bytes=len(payload_bytes), resp_bytes=len(result), + ) return Response(content=result, media_type="application/json") except RuntimeError as e: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 503, duration_ms, + req_bytes=len(payload_bytes), + ) logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) except Exception as e: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 500, duration_ms, + req_bytes=len(payload_bytes), + ) logger.error(f"路由 {path} 调用异常: {e}") raise HTTPException(status_code=500, detail="内部服务器错误") return _simple_endpoint @@ -395,20 +506,36 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_ user_info = getattr(request.state, "user", None) if user_info: meta["user_info"] = user_info - payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}} + payload_bytes = orjson.dumps({"meta": meta, "data": {k: kwargs[k] for k in param_names}}) + start = time.perf_counter() try: result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, - payload=orjson.dumps(payload), + payload=payload_bytes, timeout=timeout, cacheable=cacheable, ) + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 200, duration_ms, + req_bytes=len(payload_bytes), resp_bytes=len(result), + ) return Response(content=result, media_type="application/json") except RuntimeError as e: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 503, duration_ms, + req_bytes=len(payload_bytes), + ) logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) except Exception as e: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 500, duration_ms, + req_bytes=len(payload_bytes), + ) logger.error(f"路由 {path} 调用异常: {e}") raise HTTPException(status_code=500, detail="内部服务器错误") @@ -480,19 +607,40 @@ def _make_stream_endpoint( def _sse_frame(data: Dict[str, Any]) -> str: return f"data: {orjson.dumps(data).decode('utf-8')}\n\n" - async def _sse_body(gen) -> AsyncGenerator[str, None]: - """SSE 分帧;流中异常转为 SSE error 事件(HTTP 200 已提交)""" + async def _sse_body(gen, start: float, request: Request, req_bytes: int) -> AsyncGenerator[str, None]: + """SSE 分帧;流中异常转为 SSE error 事件(HTTP 200 已提交)。 + + 流结束(正常/空闲超时/中断/客户端断开)后在 finally 中后台记录 + 调用统计与审计,时长覆盖整个流。 + """ + status = 200 try: async for chunk in gen: yield f"data: {chunk.decode('utf-8')}\n\n" except StreamIdleTimeoutError as e: + status = 504 logger.warning("路由 %s 流式空闲超时: %s", path, e) yield _sse_frame({"error": str(e)}) + except asyncio.CancelledError: + status = 499 # 客户端断开,流未完成 + raise + except GeneratorExit: + status = 499 # 服务端主动关闭流(如连接回收),流未完成 + raise except Exception as e: + status = 500 logger.error("路由 %s 流式中断: %s", path, e) yield _sse_frame({"error": "内部服务器错误"}) + finally: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, status, duration_ms, + req_bytes=req_bytes, + ) - async def _start_stream(payload: Dict[str, Any]) -> StreamingResponse: + async def _start_stream(request: Request, payload: Dict[str, Any]) -> StreamingResponse: + req_bytes = len(orjson.dumps(payload)) + start = time.perf_counter() try: gen = await worker_manager.call_stream( worker_type=worker_type, @@ -502,27 +650,32 @@ def _make_stream_endpoint( idle_timeout=idle_timeout, ) except RuntimeError as e: + duration_ms = (time.perf_counter() - start) * 1000 + _spawn_call_record( + request, worker_type, subject, rc, 503, duration_ms, + req_bytes=req_bytes, + ) logger.warning("路由 %s [%s] 流式调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) return StreamingResponse( - _sse_body(gen), + _sse_body(gen, start, request, req_bytes), media_type="text/event-stream", headers=_STREAM_HEADERS, ) if RequestModel is not None: async def endpoint(request: Request, body: RequestModel) -> StreamingResponse: # type: ignore - return await _start_stream(build_payload(request, body=body)) + return await _start_stream(request, build_payload(request, body=body)) return endpoint # GET/DELETE 形态 if not param_names: async def _simple_endpoint(request: Request) -> StreamingResponse: - return await _start_stream(build_payload(request, kwargs=None)) + return await _start_stream(request, build_payload(request, kwargs=None)) return _simple_endpoint async def endpoint(request: Request, **kwargs: Any) -> StreamingResponse: - return await _start_stream(build_payload(request, kwargs=kwargs)) + return await _start_stream(request, build_payload(request, kwargs=kwargs)) # 附加类型注解与签名(与 _make_query_endpoint 一致),保证 OpenAPI 参数文档正确 params: List[Parameter] = [ @@ -550,6 +703,7 @@ def _register_single_route( route_conf: Dict[str, Any], extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, raw_body_models: Optional[Dict[Tuple[str, str], str]] = None, + response_models: Optional[Dict[Tuple[str, str], str]] = None, ): """注册单个路由,生成正确的函数签名以便 OpenAPI 正确生成文档""" rc = RouteConfig(route_conf) @@ -609,6 +763,15 @@ def _register_single_route( ResponseModel = generate_response_model( rc.response_model, response_model_name ) + # 响应模型仅注册到 components/schemas + responses[200] $ref 注入, + # 不传给 add_api_route:端点返回 Response 对象本就走透传(零运行时校验), + # 省掉 FastAPI 注册期对 response_model 的反射解析(external 插件频繁 + # 挂载/摘除时每次重建都会重复这一开销) + if ResponseModel is not None: + if extra_schemas is not None and response_model_name not in extra_schemas: + extra_schemas[response_model_name] = ResponseModel + if response_models is not None: + response_models[(rc.path, rc.method)] = response_model_name RequestModel = generate_request_model( param_spec, request_model_name ) @@ -627,6 +790,12 @@ def _register_single_route( ResponseModel = generate_response_model( rc.response_model, response_model_name ) + # 同上:响应模型走 extra_schemas + responses $ref 注入,不传 add_api_route + if ResponseModel is not None: + if extra_schemas is not None and response_model_name not in extra_schemas: + extra_schemas[response_model_name] = ResponseModel + if response_models is not None: + response_models[(rc.path, rc.method)] = response_model_name # 对 GET 端点也生成 Request 模型并注册到 extra_schemas if param_spec: RequestModel = generate_request_model( @@ -642,13 +811,14 @@ def _register_single_route( # 注册路由 - 使用 worker_type 作为 OpenAPI 分组标签 # 每个 worker 类型自动成为一个独立的分组,显示在 Swagger UI 中 + # 不传 response_model:端点返回 Response 对象走透传,运行时本就零校验; + # 响应文档由 response_models 收集 + _inject_schemas_to_openapi 补 $ref router.add_api_route( path=rc.path, endpoint=endpoint, methods=[rc.method], summary=rc.summary, description=rc.docstring, - response_model=ResponseModel, tags=[worker_type], include_in_schema=True, dependencies=dependencies, diff --git a/src/chongming/app/core/route_registry.py b/src/chongming/app/core/route_registry.py index 578a2b9..7ef3553 100644 --- a/src/chongming/app/core/route_registry.py +++ b/src/chongming/app/core/route_registry.py @@ -19,7 +19,7 @@ """ import asyncio import hashlib -import json +import orjson from typing import Any, Dict, List, Optional, Tuple import zenoh @@ -37,8 +37,8 @@ DEBOUNCE_SECONDS = 2.0 def _descriptor_hash(descriptor: Dict[str, Any]) -> str: """描述符内容 hash(多副本幂等去重用)""" - payload = json.dumps(descriptor, sort_keys=True, ensure_ascii=False) - return hashlib.sha1(payload.encode("utf-8")).hexdigest() + payload = orjson.dumps(descriptor, option=orjson.OPT_SORT_KEYS) + return hashlib.sha1(payload).hexdigest() class RouteRegistry: @@ -65,6 +65,8 @@ class RouteRegistry: self._schemas: Dict[str, Any] = {} # validate=false 端点的 (path, method) -> 请求模型名(OpenAPI requestBody 注入用) self._raw_body_models: Dict[Tuple[str, str], str] = {} + # 非流式端点的 (path, method) -> 响应模型名(OpenAPI responses[200] $ref 注入用) + self._response_models: Dict[Tuple[str, str], str] = {} # worker_type -> 防抖任务 self._debounce: Dict[str, asyncio.Task] = {} @@ -74,7 +76,12 @@ class RouteRegistry: self._inline_types: set = set() # 将动态 schema 注入 OpenAPI(provider 实时读取本注册表) - _inject_schemas_to_openapi(app, lambda: self._schemas, lambda: self._raw_body_models) + _inject_schemas_to_openapi( + app, + lambda: self._schemas, + lambda: self._raw_body_models, + lambda: self._response_models, + ) def set_inline_types(self, types: set): """设置 inline 插件类型集合(混合模式防双注册)。""" @@ -149,8 +156,8 @@ class RouteRegistry: return try: - descriptor = json.loads(payload.decode("utf-8")) - except (json.JSONDecodeError, UnicodeDecodeError): + descriptor = orjson.loads(payload) + except orjson.JSONDecodeError: logger.warning("丢弃非法路由描述符 [%s]: JSON 解析失败", key) return if not isinstance(descriptor, dict): @@ -238,7 +245,7 @@ class RouteRegistry: if http_routes: self._route_objs[wtype] = register_route_group( self._app, self._wm, wtype, http_routes, self._prefix, - self._schemas, self._raw_body_models, + self._schemas, self._raw_body_models, self._response_models, ) else: logger.info("[%s] 无 HTTP 路由可挂载(%d 个描述符,全部为 MQTT/internal)", diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py index c780ec2..b7444f8 100644 --- a/src/chongming/app/core/worker_manager.py +++ b/src/chongming/app/core/worker_manager.py @@ -1,7 +1,7 @@ import asyncio +import concurrent.futures import hashlib import random -import json import orjson from typing import Any, AsyncGenerator, Dict, List, Optional import zenoh @@ -12,6 +12,15 @@ logger = get_logger("worker_manager") _SENTINEL = object() # 哨兵值,用于标记"No reply received" +# Zenoh RPC 专用线程池:隔离于 asyncio.to_thread 的全局默认池 +# (默认池上限 min(32, cpu+4),高并发下会被其他阻塞调用挤占导致排队)。 +# zenoh session.get 是同步阻塞调用,每个进行中的查询占用一个线程; +# wait_for 超时后线程仍阻塞在 zenoh 内部直到其自身超时返回,故池上限即并发上限。 +# 按业务并发量调整(256 ≈ 256 个并发 Zenoh 查询);线程 daemon=True,进程退出不等待。 +_ZENOH_EXECUTOR = concurrent.futures.ThreadPoolExecutor( + max_workers=256, thread_name_prefix="zenoh-rpc" +) + class StreamIdleTimeoutError(TimeoutError): """流式调用空闲超时:在 idle_timeout 秒内未收到任何数据块""" @@ -216,11 +225,13 @@ class WorkerManager: async def _stream(): # 逐块消费;zenoh reply 迭代器是同步阻塞的,必须放线程执行, - # 避免阻塞事件循环(与 call() 的 _get_first 同模式) + # 避免阻塞事件循环(与 call() 的 _get_first 同模式)。 + # 走专用 _ZENOH_EXECUTOR 池,不与全局默认池的其它阻塞调用争抢 while True: try: + loop = asyncio.get_running_loop() reply = await asyncio.wait_for( - asyncio.to_thread(_next_reply, reply_iter), + loop.run_in_executor(_ZENOH_EXECUTOR, _next_reply, reply_iter), timeout=idle_timeout, ) except asyncio.TimeoutError: @@ -278,7 +289,7 @@ class WorkerManager: Returns: Dict[str, Any]: 调用结果(worker 回复解析后的 dict) """ - cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}" + cache_key = f"{worker_type}:{subject}:{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}" reply_bytes = await self._call_impl( worker_type, subject, orjson.dumps(payload), timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key, @@ -370,8 +381,10 @@ class WorkerManager: return item return _SENTINEL + # 走专用 _ZENOH_EXECUTOR 池(隔离于全局默认池,避免高并发下排队) + loop = asyncio.get_running_loop() first_reply = await asyncio.wait_for( - asyncio.to_thread(_get_first, reply_iter), + loop.run_in_executor(_ZENOH_EXECUTOR, _get_first, reply_iter), timeout=timeout, ) if first_reply is _SENTINEL: diff --git a/src/chongming/app/routes/__init__.py b/src/chongming/app/routes/__init__.py index acfd551..dabae44 100644 --- a/src/chongming/app/routes/__init__.py +++ b/src/chongming/app/routes/__init__.py @@ -1,3 +1,4 @@ from .storage import router as storage_router +from .stats import router as stats_router -__all__ = ["storage_router"] \ No newline at end of file +__all__ = ["storage_router", "stats_router"] \ No newline at end of file diff --git a/src/chongming/app/routes/stats.py b/src/chongming/app/routes/stats.py new file mode 100644 index 0000000..637ff42 --- /dev/null +++ b/src/chongming/app/routes/stats.py @@ -0,0 +1,23 @@ +# src/chongming/app/routes/stats.py +""" +网关级调用统计查看端点 + +数据来源为 core.call_stats 的内存统计(进程内视角): +动态路由每次调用在后台任务中更新计数/延迟聚合/最近调用缓冲, +此处提供只读快照,便于观测接口调用量、错误分布与延迟。 +""" +from fastapi import APIRouter + +from chongming_config import get_config + +from ..core import call_stats + +config = get_config() + +router = APIRouter(prefix=config["default"]["prefix"], tags=["stats"]) + + +@router.get("/stats", summary="调用统计") +async def get_call_stats(): + """返回内存调用统计快照:总量、per worker/subject/status 计数、延迟聚合、最近调用""" + return call_stats.summary() diff --git a/test_call_stats.py b/test_call_stats.py new file mode 100644 index 0000000..b6ae633 --- /dev/null +++ b/test_call_stats.py @@ -0,0 +1,333 @@ +""" +调用统计 + 审计日志(后台任务)测试(依赖隔离:桩掉 chongming 包链、auth、 +chongming_cache 与 worker_manager,加载真实 call_stats 与 route_generator) + +用 httpx.ASGITransport 验证: + 1. CallStats.record/summary 聚合正确性(计数 / 平均 / 最大延迟 / recent 上限) + 2. spawn/shutdown:后台任务被跟踪、shutdown 后全部取消 + 3. 动态端点请求后:响应与改动前一致(字节透传)+ 统计计数正确 + 审计日志出现 AUDIT 行 + 4. 错误路径(无可用 worker)-> HTTP 503 同样被记录 + 5. 后台任务失败兜底:record_call 内部异常不冒泡,端点响应不受影响 + +运行: uv run python test_call_stats.py +""" +import asyncio +import importlib.util +import logging +import sys +import types +from pathlib import Path + +import httpx +import orjson +from fastapi import FastAPI + +REPO_ROOT = Path(__file__).resolve().parent +sys.path.insert(0, str(REPO_ROOT / "src")) +sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src")) + +# ---- 依赖隔离:桩掉 chongming 包链 / auth / chongming_cache / worker_manager ---- +_pkg = types.ModuleType("chongming") +_pkg.__path__ = [] +_app = types.ModuleType("chongming.app") +_app.__path__ = [] +_core = types.ModuleType("chongming.app.core") +_core.__path__ = [] +_logger_stub = types.ModuleType("chongming.app.core.logger") +_logger_stub.get_logger = lambda name: logging.getLogger(name) +_auth_stub = types.ModuleType("chongming.app.core.auth") +_auth_stub.store_user_to_request = lambda x: x # 仅用于 Depends,桩掉鉴权 +_cache_stub = types.ModuleType("chongming_cache") +_cache_stub.Cache = object +# worker_manager 桩:route_generator 仅用其类型注解与异常类 +_wm_stub = types.ModuleType("chongming.app.core.worker_manager") +_wm_stub.WorkerManager = object +_wm_stub.StreamIdleTimeoutError = type("StreamIdleTimeoutError", (RuntimeError,), {}) +for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub, _wm_stub): + sys.modules[mod.__name__] = mod + +# 加载真实 call_stats(route_generator 依赖其 spawn/record_call) +_cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py" +_cs_spec = importlib.util.spec_from_file_location("chongming.app.core.call_stats", _cs_path) +_cs_mod = importlib.util.module_from_spec(_cs_spec) +sys.modules["chongming.app.core.call_stats"] = _cs_mod +_cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr] + +# 加载真实 route_generator +_rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py" +_rg_spec = importlib.util.spec_from_file_location("chongming.app.core.route_generator", _rg_path) +_rg_mod = importlib.util.module_from_spec(_rg_spec) +sys.modules["chongming.app.core.route_generator"] = _rg_mod +_rg_spec.loader.exec_module(_rg_mod) # type: ignore[union-attr] + +register_route_group = _rg_mod.register_route_group +record_call = _cs_mod.record_call +spawn = _cs_mod.spawn + +# ---- 审计日志捕获(桩 logger 直接挂到标准库 audit logger) ---- +_audit_records = [] + + +class _AuditHandler(logging.Handler): + def emit(self, record): + _audit_records.append(record.getMessage()) + + +_audit_logger = logging.getLogger("audit") +_audit_logger.setLevel(logging.INFO) +_audit_logger.addHandler(_AuditHandler()) +_audit_logger.propagate = False + +failures = [] + + +def check(name, cond, detail=""): + status = "PASS" if cond else "FAIL" + print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) + if not cond: + failures.append(name) + + +class FakeWorkerManager: + """模拟 WorkerManager:call_raw 返回原始字节;error="no_worker" 时抛 RuntimeError""" + + def __init__(self, reply=b'{"ok": true}', error=None): + self.reply = reply + self.error = error + + async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False): + if self.error == "no_worker": + raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。") + return self.reply + + async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0): + if self.error == "no_worker": + raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。") + + async def _gen(): + yield orjson.dumps({"chunk": 1}) + yield orjson.dumps({"chunk": 2}) + + return _gen() # 与真实实现一致:async 函数返回异步生成器 + + +ROUTES = [ + # body 端点(validate=true) + {"subject": "demo.echo", "path": "/echo", "method": "POST", + "params": ["msg: str"], "timeout": 3.0, "auth_required": False}, + # raw body 端点(validate=false) + {"subject": "demo.raw", "path": "/raw/echo", "method": "POST", + "params": ["msg: str"], "validate": False, "timeout": 3.0, "auth_required": False}, + # query 带参端点 + {"subject": "demo.add", "path": "/add", "method": "GET", + "params": ["a: int", "b: int"], "timeout": 3.0, "auth_required": False}, + # query 无参端点(_simple_endpoint 分支) + {"subject": "demo.ping", "path": "/ping", "method": "GET", + "timeout": 3.0, "auth_required": False}, + # 流式端点 + {"subject": "demo.stream", "path": "/stream", "method": "GET", + "stream": True, "timeout": 3.0, "auth_required": False}, +] + + +def build_app(fwm): + app = FastAPI() + register_route_group(app, fwm, "demo", ROUTES, prefix="/api/v1") + return app + + +async def _drain_background(): + """等待并清理所有后台统计/审计任务(每个异步测试末尾调用)""" + tasks = list(_cs_mod._tasks) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + await _cs_mod.shutdown() + + +def _reset_stats(): + """隔离:每个端点测试前重置全局统计实例""" + _cs_mod._stats = _cs_mod.CallStats() + + +# ---- 1. CallStats 单元 ---- + +def test_stats_aggregation(): + cs = _cs_mod.CallStats(max_recent=10) + cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 200, "duration_ms": 10.0}) + cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 200, "duration_ms": 30.0}) + cs.record({"worker_type": "demo", "subject": "demo.echo", "status": 503, "duration_ms": 5.0}) + cs.record({"worker_type": "other", "subject": "other.x", "status": 200, "duration_ms": 2.0}) + s = cs.summary() + check("total_calls", s["total_calls"] == 4, str(s["total_calls"])) + by_key = {(c["worker_type"], c["subject"], c["status"]): c["count"] for c in s["calls"]} + check( + "per-key 计数", + by_key == { + ("demo", "demo.echo", 200): 2, + ("demo", "demo.echo", 503): 1, + ("other", "other.x", 200): 1, + }, + str(by_key), + ) + lat = {l["worker_type"]: l for l in s["latency"]} + check("延迟聚合 count", lat["demo"]["count"] == 3, str(lat["demo"])) + check("延迟聚合 avg", lat["demo"]["avg_ms"] == 15.0, str(lat["demo"]["avg_ms"])) + check("延迟聚合 max", lat["demo"]["max_ms"] == 30.0, str(lat["demo"]["max_ms"])) + check("recent 保留", len(s["recent"]) == 4, str(len(s["recent"]))) + + +def test_recent_bounded(): + cs = _cs_mod.CallStats(max_recent=5) + for _ in range(20): + cs.record({"worker_type": "demo", "subject": "demo.x", "status": 200, "duration_ms": 1.0}) + s = cs.summary() + check("recent 环形缓冲上限", len(s["recent"]) == 5, str(len(s["recent"]))) + check("计数不受缓冲影响", s["total_calls"] == 20, str(s["total_calls"])) + + +# ---- 2. spawn / shutdown ---- + +def test_spawn_shutdown(): + async def run(): + async def never(): + await asyncio.sleep(3600) + + t = spawn(never()) + check("任务被跟踪", t in _cs_mod._tasks, str(list(_cs_mod._tasks))) + await _cs_mod.shutdown() + check("shutdown 后任务取消", t.cancelled(), str(t)) + check("任务集合清空", len(_cs_mod._tasks) == 0, str(len(_cs_mod._tasks))) + + asyncio.run(run()) + + +# ---- 3. 端点集成:统计 + 审计 ---- + +def test_endpoint_records_stats_and_audit(): + async def run(): + _audit_records.clear() + _reset_stats() + fwm = FakeWorkerManager(reply=b'{"ok": true}') + app = build_app(fwm) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac: + r1 = await ac.post("/api/v1/echo", json={"msg": "hi"}) + r2 = await ac.post("/api/v1/raw/echo", json={"msg": "hi"}) + r3 = await ac.get("/api/v1/add?a=1&b=2") + r4 = await ac.get("/api/v1/ping") + r5 = await ac.get("/api/v1/stream") + await _drain_background() + + # 响应与改动前一致:worker 原始字节零再序列化 + check("POST 响应字节透传", r1.status_code == 200 and r1.content == b'{"ok": true}', repr(r1.content)) + check("raw POST 响应字节透传", r2.status_code == 200 and r2.content == b'{"ok": true}', repr(r2.content)) + check("GET 响应字节透传", r3.status_code == 200 and r3.content == b'{"ok": true}', repr(r3.content)) + check("无参 GET 响应字节透传", r4.status_code == 200 and r4.content == b'{"ok": true}', repr(r4.content)) + check("流式响应 SSE 帧", r5.status_code == 200 and b"data:" in r5.content and b'"chunk":1' in r5.content, repr(r5.content[:80])) + + s = _cs_mod.summary() + check("total_calls == 5", s["total_calls"] == 5, str(s["total_calls"])) + subjects = {c["subject"] for c in s["calls"]} + check( + "5 种调用均记录", + subjects == {"demo.echo", "demo.raw", "demo.add", "demo.ping", "demo.stream"}, + str(subjects), + ) + check("全部状态 200", all(c["status"] == 200 for c in s["calls"]), str(s["calls"])) + + check("审计行数 == 5", len(_audit_records) == 5, str(len(_audit_records))) + echo_audit = next((a for a in _audit_records if "subject=demo.echo" in a), "") + check( + "审计行格式", + all(k in echo_audit for k in ( + "AUDIT", "ts=", "worker=demo", "method=POST", + "path=/api/v1/echo", "status=200", "req_bytes=", "resp_bytes=", + )), + echo_audit, + ) + stream_audit = next((a for a in _audit_records if "subject=demo.stream" in a), "") + check("流式审计已记录", "status=200" in stream_audit and "path=/api/v1/stream" in stream_audit, stream_audit) + + asyncio.run(run()) + + +def test_error_503_recorded(): + async def run(): + _audit_records.clear() + _reset_stats() + fwm = FakeWorkerManager(error="no_worker") + app = build_app(fwm) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac: + r = await ac.get("/api/v1/ping") + await _drain_background() + + check("503 响应", r.status_code == 503, str(r.status_code)) + s = _cs_mod.summary() + check( + "503 计数 1", + any(c["subject"] == "demo.ping" and c["status"] == 503 and c["count"] == 1 for c in s["calls"]), + str(s["calls"]), + ) + check( + "503 审计行", + any("subject=demo.ping" in a and "status=503" in a for a in _audit_records), + str(_audit_records), + ) + + asyncio.run(run()) + + +# ---- 4. 后台任务失败兜底 ---- + +def test_background_failure_isolated(): + async def run(): + _audit_records.clear() + _reset_stats() + original = _cs_mod._stats.record + + def boom(entry): + raise ValueError("stats broken") + + _cs_mod._stats.record = boom + try: + # 后台执行体自身不冒泡(统计故障仅记 warning,审计仍写入) + await record_call({ + "ts": 0.0, "worker_type": "demo", "subject": "demo.x", "status": 200, + "duration_ms": 1.0, "method": "GET", "path": "/x", + "user": None, "client": None, "req_bytes": 0, "resp_bytes": 0, + }) + check("record_call 异常被兜底(不冒泡)", True) + check("审计仍写入", any("subject=demo.x" in a for a in _audit_records), str(_audit_records)) + finally: + _cs_mod._stats.record = original + + # 统计故障下端点响应不受影响 + fwm = FakeWorkerManager(reply=b'{"ok": true}') + app = build_app(fwm) + _cs_mod._stats.record = boom + try: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac: + r = await ac.get("/api/v1/ping") + await _drain_background() + check("统计故障下响应仍 200", r.status_code == 200 and r.content == b'{"ok": true}', repr(r.content)) + finally: + _cs_mod._stats.record = original + + asyncio.run(run()) + + +# ---- 执行 ---- +test_stats_aggregation() +test_recent_bounded() +test_endpoint_records_stats_and_audit() +test_error_503_recorded() +test_background_failure_isolated() +test_spawn_shutdown() # 最后执行:shutdown 会取消所有未完成任务 + +print() +if failures: + print(f"共 {len(failures)} 项失败: {failures}") + sys.exit(1) +print("全部通过") diff --git a/test_jwt_cache.py b/test_jwt_cache.py new file mode 100644 index 0000000..9eba72f --- /dev/null +++ b/test_jwt_cache.py @@ -0,0 +1,238 @@ +""" +JWT 解析共享缓存测试(auth.py _cached_decode / jwt_auth_dependency) + +验证: + 1. 同 token 重复鉴权只解密一次(缓存命中),不同 token 独立缓存 + 2. TTL 过期后重新解码(条目自然失效) + 3. 缓存条目 exp 已过期:命中即拒绝,不重解密(缓存不延长 token 有效期) + 4. 无效 token 不缓存失败:每次都走完整验证(decode 调用计数递增) + 5. 缓存 key 含密钥/算法指纹:同 token 不同密钥不串用 + 6. 返回值为独立对象:调用方修改不影响缓存 + 7. 端到端:真实 chongming_jwt 签发 -> 鉴权成功 -> 二次命中缓存 + +运行: uv run python test_jwt_cache.py +""" +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) +sys.path.insert(0, str(Path(__file__).resolve().parent / "utils" / "worker" / "src")) + +import orjson +from fastapi.security import HTTPAuthorizationCredentials + +from chongming.app.core import auth as auth_mod +from chongming.app.core.auth import jwt_auth_dependency +from chongming_cache import Cache +from chongming_jwt import JWT +from chongming_config import get_config + +failures = [] + + +def check(name, cond, detail=""): + status = "PASS" if cond else "FAIL" + print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) + if not cond: + failures.append(name) + + +config = get_config() +SECRET = config["security"]["secret_key"] +ALGORITHM = config["security"]["algorithm"] + +# 每个测试使用独立临时 diskcache,隔离真实缓存目录 +_tmp = tempfile.TemporaryDirectory() + + +def fresh_cache(): + auth_mod._jwt_cache = Cache(backend="diskcache", cache_store_path=_tmp.name) + auth_mod._jwt_cache.clear() + + +def make_creds(token): + return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) + + +# ---- 1. 缓存命中与隔离 ---- +def test_cache_hit_and_isolation(): + fresh_cache() + + class CountingJWT: + calls = 0 + + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + CountingJWT.calls += 1 + return orjson.dumps({"sub": token, "exp": int(time.time()) + 3600}).decode() + + auth_mod.JWT = CountingJWT + t1, t2 = "token-1", "token-2" + p1 = jwt_auth_dependency(credentials=make_creds(t1)) + p2 = jwt_auth_dependency(credentials=make_creds(t1)) + p3 = jwt_auth_dependency(credentials=make_creds(t2)) + check("同 token 二次鉴权命中缓存(只解密一次)", CountingJWT.calls == 2, f"calls={CountingJWT.calls}") + check("payload 解析正确", p1["sub"] == "token-1" and p2["sub"] == "token-1", str(p1)) + check("不同 token 独立缓存", p3["sub"] == "token-2", str(p3)) + check("缓存中已写入条目", len(auth_mod._jwt_cache) >= 2, str(len(auth_mod._jwt_cache))) + + +# ---- 2. TTL 过期后重新解码 ---- +def test_ttl_expiry_redecodes(): + fresh_cache() + auth_mod._JWT_CACHE_TTL = 1 # 缩短 TTL 加速测试 + + class TtlJWT: + calls = 0 + + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + TtlJWT.calls += 1 + return orjson.dumps({"sub": "u1", "exp": int(time.time()) + 3600}).decode() + + auth_mod.JWT = TtlJWT + creds = make_creds("ttl-token") + p1 = jwt_auth_dependency(credentials=creds) + p2 = jwt_auth_dependency(credentials=creds) + check("TTL 内命中", TtlJWT.calls == 1 and p2["sub"] == "u1", f"calls={TtlJWT.calls}") + time.sleep(1.2) # 等待 TTL 过期 + p3 = jwt_auth_dependency(credentials=creds) + check("TTL 过期后重新解码", TtlJWT.calls == 2 and p3["sub"] == "u1", f"calls={TtlJWT.calls}") + + +# ---- 3. 缓存条目 exp 已过期:命中即拒绝,不重解密 ---- +def test_cached_expired_rejected(): + fresh_cache() + + class NeverJWT: + calls = 0 + + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + NeverJWT.calls += 1 + raise AssertionError("不应触发解码(命中缓存即应拒绝)") + + auth_mod.JWT = NeverJWT + # 预置一个 exp 已过期的缓存条目(模拟缓存写入后 token 到期) + key = auth_mod._cache_key("expired-token", SECRET, ALGORITHM) + auth_mod._jwt_cache.set( + key, {"payload": {"sub": "u1"}, "exp": int(time.time()) - 10}, expire=60 + ) + for i in range(2): + try: + jwt_auth_dependency(credentials=make_creds("expired-token")) + check(f"缓存过期条目第 {i+1} 次应 401", False, "未抛异常") + except Exception as e: + check(f"缓存过期条目第 {i+1} 次 401", getattr(e, "status_code", None) == 401, str(e)) + check("过期条目命中后未重解密(两次都直接拒绝)", NeverJWT.calls == 0, f"calls={NeverJWT.calls}") + check("过期条目保留在缓存(TTL 内持续拒绝)", key in auth_mod._jwt_cache) + + +# ---- 4. 无效 token 失败不缓存 ---- +def test_invalid_token_not_cached(): + fresh_cache() + + class BadJWT: + calls = 0 + + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + BadJWT.calls += 1 + raise ValueError("Invalid signature") + + auth_mod.JWT = BadJWT + creds = make_creds("bad-token") + for _ in range(2): + try: + jwt_auth_dependency(credentials=creds) + check("无效 token 应 401", False) + except Exception as e: + ok = getattr(e, "status_code", None) == 401 + check("无效 token 401", ok, str(e)) + check("无效 token 每次都完整验证(失败不缓存)", BadJWT.calls == 2, f"calls={BadJWT.calls}") + check("缓存无失败条目", len(auth_mod._jwt_cache) == 0, str(len(auth_mod._jwt_cache))) + + +# ---- 5. 缓存 key 含密钥/算法指纹 ---- +def test_key_scoped_to_secret(): + fresh_cache() + + class KeyJWT: + calls = 0 + + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + KeyJWT.calls += 1 + return orjson.dumps({"sub": token, "exp": int(time.time()) + 3600}).decode() + + auth_mod.JWT = KeyJWT + creds = make_creds("same-token") + jwt_auth_dependency(credentials=creds) # secret A + jwt_auth_dependency(credentials=creds) # secret A -> 命中 + auth_mod._cached_decode("same-token", "secret-B", ALGORITHM) # secret B -> miss + check("同 token 不同密钥独立缓存", KeyJWT.calls == 2, f"calls={KeyJWT.calls}") + + +# ---- 6. 返回值独立对象:修改不影响缓存 ---- +def test_payload_object_independent(): + fresh_cache() + + class FixedJWT: + def __init__(self, secret, algorithm): + pass + + def decode(self, token): + return orjson.dumps({"sub": "u1", "role": "admin", "exp": int(time.time()) + 3600}).decode() + + auth_mod.JWT = FixedJWT + p1 = jwt_auth_dependency(credentials=make_creds("t-copy")) + p1["role"] = "hacker" + p2 = jwt_auth_dependency(credentials=make_creds("t-copy")) + check("修改返回值不污染缓存", p2["role"] == "admin", str(p2)) + + +# ---- 7. 端到端:真实 JWT 签发 -> 鉴权 -> 命中缓存 ---- +def test_e2e_real_jwt(): + fresh_cache() + auth_mod.JWT = JWT # 恢复真实 JWT 类 + jwt = JWT(SECRET, ALGORITHM) + token = jwt.encode(orjson.dumps({"sub": "1", "username": "alice"}).decode(), expires_in_seconds=3600) + creds = make_creds(token) + p1 = jwt_auth_dependency(credentials=creds) + p2 = jwt_auth_dependency(credentials=creds) + check("真实 JWT 鉴权成功", p1["sub"] == "1" and p1["username"] == "alice", str(p1)) + check("二次调用命中缓存", auth_mod._cache_key(token, SECRET, ALGORITHM) in auth_mod._jwt_cache) + try: + jwt_auth_dependency(credentials=None) + check("缺 header 应 401", False) + except Exception as e: + check("缺 header 401", getattr(e, "status_code", None) == 401) + + +test_cache_hit_and_isolation() +test_ttl_expiry_redecodes() +test_cached_expired_rejected() +test_invalid_token_not_cached() +test_key_scoped_to_secret() +test_payload_object_independent() +test_e2e_real_jwt() + +_tmp.cleanup() + +print() +if failures: + print(f"共 {len(failures)} 项失败: {failures}") + sys.exit(1) +print("全部通过") diff --git a/test_logging_async.py b/test_logging_async.py new file mode 100644 index 0000000..e3c9c13 --- /dev/null +++ b/test_logging_async.py @@ -0,0 +1,143 @@ +""" +access 日志采样 + 异步日志队列(QueueHandler + QueueListener)测试 + +验证: + 1. _should_log_access:错误(>=400)全量;成功按采样率(确定性计数);1.0 全量 + 2. QueueHandler + QueueListener:logger 只挂队列 handler,I/O 由后台监听线程落盘 + 3. 队列满不抛异常(丢弃而非阻塞/打断业务) + 4. 端到端:真实 app 请求 -> access 日志异步落盘(含采样命中) + +运行: uv run python test_logging_async.py +""" +import importlib.util +import itertools +import queue +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) +sys.path.insert(0, str(Path(__file__).resolve().parent / "utils" / "worker" / "src")) + +_tmp = tempfile.TemporaryDirectory() + +# 预加载真实 logger 模块(不触发 chongming 包链导入),把日志目录重定向到 +# 临时目录,之后 import chongming.app 时 get_logger 即用新路径 +_logger_path = Path(__file__).resolve().parent / "src" / "chongming" / "app" / "core" / "logger.py" +_logger_spec = importlib.util.spec_from_file_location("chongming.app.core.logger", _logger_path) +logger_mod = importlib.util.module_from_spec(_logger_spec) +sys.modules["chongming.app.core.logger"] = logger_mod +_logger_spec.loader.exec_module(logger_mod) # type: ignore[union-attr] +logger_mod.log_config["file"] = _tmp.name + +import chongming.app as app_mod + +failures = [] + + +def check(name, cond, detail=""): + status = "PASS" if cond else "FAIL" + print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) + if not cond: + failures.append(name) + + +def _wait_for(pred, timeout=3.0): + deadline = time.time() + timeout + while time.time() < deadline: + if pred(): + return True + time.sleep(0.05) + return False + + +# ---- 1. 采样逻辑 ---- +def test_sampling(): + app_mod._access_log_counter = itertools.count() + check( + "错误 4xx/5xx 全量", + app_mod._should_log_access(404) + and app_mod._should_log_access(500) + and app_mod._should_log_access(503), + ) + app_mod._access_log_counter = itertools.count() + check("rate>=1.0 全量", app_mod._should_log_access(200, 1.0) and app_mod._should_log_access(200, 1.0)) + app_mod._access_log_counter = itertools.count() + logged = sum(1 for _ in range(10) if app_mod._should_log_access(200, 0.1)) + check("0.1 采样率 10 个成功请求记 1 条", logged == 1, f"logged={logged}") + + +# ---- 2. 异步队列:logger 只挂队列 handler,监听线程落盘 ---- +def test_async_queue_write(): + from logging.handlers import QueueHandler + + lg = logger_mod.get_logger("test_async") + check("logger 只挂队列 handler", len(lg.handlers) == 1 and isinstance(lg.handlers[0], QueueHandler)) + check("监听线程已启动", logger_mod._listener is not None and logger_mod._listener._thread.is_alive()) + + for i in range(5): + lg.info("async log %d", i) # % 参数化:格式化在监听线程执行 + ok = _wait_for( + lambda: any( + "async log 4" in f.read_text(errors="ignore") + for f in Path(_tmp.name).glob("test_async-*.log") + ) + ) + check("日志异步落盘(监听线程写入)", ok) + + +# ---- 3. 队列满:丢弃而非抛异常 ---- +def test_queue_full_drops(): + old = logger_mod._log_queue + logger_mod._log_queue = queue.Queue(maxsize=2) # 换小队列 + try: + lg = logger_mod.get_logger("test_full") + try: + for i in range(20): + lg.info("flood %d", i) + check("队列满不抛异常(丢弃新日志)", True) + except Exception as e: + check("队列满不抛异常(丢弃新日志)", False, str(e)) + finally: + logger_mod._log_queue = old + + +# ---- 4. 端到端:中间件采样 + 异步落盘 ---- +def test_e2e_middleware(): + import asyncio + import httpx + + app_mod._access_log_counter = itertools.count() # 重置采样计数 + + async def main(): + transport = httpx.ASGITransport(app=app_mod.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as ac: + r = await ac.get("/") + return r.status_code + + code = asyncio.run(main()) + check("请求返回 200", code == 200, str(code)) + ok = _wait_for( + lambda: any( + "GET / status=200" in f.read_text(errors="ignore") + for f in Path(_tmp.name).glob("app-*.log") + ) + ) + check("access 日志异步落盘(采样命中)", ok) + + +test_sampling() +test_async_queue_write() +test_queue_full_drops() +test_e2e_middleware() + +# 关闭监听线程并释放文件句柄,再清理临时目录 +logger_mod.shutdown_logging() +_tmp.cleanup() + +print() +if failures: + print(f"共 {len(failures)} 项失败: {failures}") + sys.exit(1) +print("全部通过") diff --git a/test_mqtt_auth.py b/test_mqtt_auth.py index 9ba9699..8ef4a73 100644 --- a/test_mqtt_auth.py +++ b/test_mqtt_auth.py @@ -10,7 +10,7 @@ MQTT Broker 认证集成测试 C. MQTTBridge 客户端凭据传递(含特殊字符密码的 URI 编码) """ import asyncio -import json +import orjson import multiprocessing import socket import sys @@ -143,7 +143,7 @@ async def scenario_bridge(port: int): assert m.topic == "chongming/demo/op", f"未收到请求: {m.topic}" m2 = await asyncio.wait_for(sub.deliver_message(), timeout=10) assert m2.topic == "chongming/demo/op/response", f"未收到响应: {m2.topic}" - assert json.loads(m2.data) == {"ok": True}, f"响应内容异常: {m2.data}" + assert orjson.loads(m2.data) == {"ok": True}, f"响应内容异常: {m2.data}" await bridge.stop() await pub.disconnect() print("PASS [C] MQTTBridge 凭据传递: 连接/路由/响应链路正常(含特殊字符密码)") diff --git a/test_mqtt_custom_topics.py b/test_mqtt_custom_topics.py index fc84d5d..7c589d6 100644 --- a/test_mqtt_custom_topics.py +++ b/test_mqtt_custom_topics.py @@ -7,7 +7,7 @@ MQTT Bridge 自定义订阅主题单元测试 3. 内置 chongming/... 主题兜底逻辑不变 """ import asyncio -import json +import orjson import sys from pathlib import Path from unittest.mock import MagicMock @@ -24,7 +24,7 @@ from chongming.app.core.mqtt_bridge import ( def make_msg(topic, payload): m = MagicMock() m.topic = topic - m.data = json.dumps(payload).encode("utf-8") + m.data = orjson.dumps(payload) return m diff --git a/test_mqtt_e2e.py b/test_mqtt_e2e.py index 1285987..ddc74ff 100644 --- a/test_mqtt_e2e.py +++ b/test_mqtt_e2e.py @@ -12,7 +12,7 @@ MQTT 端到端测试脚本 """ import asyncio -import json +import orjson import sys import logging from amqtt.client import MQTTClient @@ -51,7 +51,7 @@ async def run_test(): for i, case in enumerate(TEST_CASES): a, b, expected = case["a"], case["b"], case["expected"] payload = {"meta": {"method": "MQTT"}, "data": {"a": a, "b": b}} - payload_bytes = json.dumps(payload).encode("utf-8") + payload_bytes = orjson.dumps(payload) await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1) print(f"📤 [{i+1}/{len(TEST_CASES)}] 发送: {REQUEST_TOPIC} → {payload}") @@ -59,14 +59,14 @@ async def run_test(): # 等待响应(超时 5 秒) try: message = await asyncio.wait_for(client.deliver_message(), timeout=5.0) - data = json.loads(message.data.decode("utf-8")) # type: ignore + data = orjson.loads(message.data) # type: ignore actual = data.get("result") status = "✅ PASS" if abs(actual - expected) < 0.01 else "❌ FAIL" if status == "✅ PASS": PASS += 1 else: FAIL += 1 - print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {json.dumps(data)} → {status} (expect={expected}, got={actual})") + print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {orjson.dumps(data).decode()} → {status} (expect={expected}, got={actual})") except asyncio.TimeoutError: FAIL += 1 print(f"📥 [{i+1}/{len(TEST_CASES)}] ❌ FAIL: 超时无响应") diff --git a/test_passthrough_optimization.py b/test_passthrough_optimization.py index ea1b2f2..b67b0ef 100644 --- a/test_passthrough_optimization.py +++ b/test_passthrough_optimization.py @@ -61,6 +61,15 @@ _wm_mod = importlib.util.module_from_spec(_wm_spec) sys.modules["chongming.app.core.worker_manager"] = _wm_mod _wm_spec.loader.exec_module(_wm_mod) # type: ignore[union-attr] +# 加载真实 call_stats(route_generator 依赖其 spawn/record_call) +_cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py" +_cs_spec = importlib.util.spec_from_file_location( + "chongming.app.core.call_stats", _cs_path +) +_cs_mod = importlib.util.module_from_spec(_cs_spec) +sys.modules["chongming.app.core.call_stats"] = _cs_mod +_cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr] + # 加载真实 route_generator _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py" _rg_spec = importlib.util.spec_from_file_location( diff --git a/test_stream_route.py b/test_stream_route.py index 2112d32..436d06e 100644 --- a/test_stream_route.py +++ b/test_stream_route.py @@ -50,6 +50,15 @@ _wm_mod = importlib.util.module_from_spec(_wm_spec) sys.modules["chongming.app.core.worker_manager"] = _wm_mod _wm_spec.loader.exec_module(_wm_mod) # type: ignore[union-attr] +# 加载真实 call_stats(route_generator 依赖其 spawn/record_call) +_cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py" +_cs_spec = importlib.util.spec_from_file_location( + "chongming.app.core.call_stats", _cs_path +) +_cs_mod = importlib.util.module_from_spec(_cs_spec) +sys.modules["chongming.app.core.call_stats"] = _cs_mod +_cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr] + # 加载真实 route_generator _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py" _rg_spec = importlib.util.spec_from_file_location( diff --git a/uv.lock b/uv.lock index 52ae2a8..28d8b64 100644 --- a/uv.lock +++ b/uv.lock @@ -486,7 +486,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git [[package]] name = "chongming-worker" version = "0.3.0" -source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#c4a998084fc42572e121a27b84082579e66dd3d0" } +source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#7e4a3f66b30ee7ab56b2b066790a029d4a4ed6e8" } dependencies = [ { name = "eclipse-zenoh" }, { name = "minio" }, @@ -514,6 +514,7 @@ dependencies = [ { name = "eclipse-zenoh" }, { name = "fastapi" }, { name = "gunicorn" }, + { name = "httptools" }, { name = "module-bank" }, { name = "openai" }, { name = "orjson" }, @@ -537,8 +538,8 @@ dev = [ [package.metadata] requires-dist = [ - { name = "aiofiles", specifier = ">=25.1.0" }, - { name = "amqtt", specifier = ">=0.11.3" }, + { name = "aiofiles" }, + { name = "amqtt" }, { name = "bytedance", git = "http://124.71.68.6:3000/chongming/bytedance.git" }, { name = "chongming-cache", git = "http://124.71.68.6:3000/chongming/chongming-utils-cache.git" }, { name = "chongming-config", git = "http://124.71.68.6:3000/chongming/chongming-utils-config.git" }, @@ -552,6 +553,7 @@ requires-dist = [ { name = "eclipse-zenoh" }, { name = "fastapi", extras = ["uvicorn"] }, { name = "gunicorn" }, + { name = "httptools" }, { name = "module-bank" }, { name = "openai" }, { name = "orjson" }, @@ -560,8 +562,8 @@ requires-dist = [ { name = "tomli" }, { name = "typing-extensions" }, { name = "uvicorn" }, - { name = "uvloop", specifier = ">=0.22.1" }, - { name = "volcengine-python-sdk", extras = ["ark"], specifier = ">=5.0.41" }, + { name = "uvloop" }, + { name = "volcengine-python-sdk", extras = ["ark"] }, ] [package.metadata.requires-dev] @@ -793,6 +795,42 @@ wheels = [ { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/d2/74/d370e55600d9bcfa0d9794b0166126d49291a3d2b20c268fc98c453a4948/httpcore2-2.12.0-py3-none-any.whl", hash = "sha256:7e04258ce01013d7d615e5b910a3b27fac937d7a95038227e79652b4ba3b4ceb", size = 83074, upload-time = "2026-08-18T13:22:05.854Z" }, ] +[[package]] +name = "httptools" +version = "0.8.0" +source = { registry = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple" } +sdist = { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/43/e5/d471fcb0e14523fe1c3f4ba58ca52480e7bd70ad7109a3846bc75892f7fb/httptools-0.8.0.tar.gz", hash = "sha256:6b2a32f18d97e16e90827d7a819ffa8dbd8cc245fc4e1fa9d1095b54ef4bd999", size = 271342, upload-time = "2026-05-25T22:17:48.841Z" } +wheels = [ + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/14/88/1d21a36da8f5cb0fa49eafd4b169eba5608d57e75bbcf61845cbc6243216/httptools-0.8.0-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:880490234c10f70a9830743097e8958d6e4b9f5a0ffc24515023afeef984054d", size = 208247, upload-time = "2026-05-25T22:17:07.843Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/a5/42/cc4feea2945cb3051038f090c9b36bd5b8a9d7f5a894a506a8983e33fd1c/httptools-0.8.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:5931891fb7b441b8a3853cf1b85c82c903defce084dd5f6771ca46e31bf862c5", size = 113064, upload-time = "2026-05-25T22:17:09.136Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/e3/a6/febbb8b8db0f58b38e44ad6cb946e6a255ae49b55f2e8543408fb7501ccd/httptools-0.8.0-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:b15fc622b0f869d19207c4089a501d9bcc63ca5e071ffdd2f03f922df882dcb2", size = 523851, upload-time = "2026-05-25T22:17:10.106Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/b7/e4/f90a0df0b83beff265b7e3b65f2a4cefd95792d4be0ac3e16049f2acd3c2/httptools-0.8.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:425f83884fd6343828d8c565f046cb72b6d19063f6924093e11bcd8e1548cd09", size = 518842, upload-time = "2026-05-25T22:17:11.218Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/9e/2d/0c9ac76dd2c893841fbf6498d6acec4f2442e1b7067f6e3e316a80e494e8/httptools-0.8.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:ef7c3c97f4311c7be57e2986629df89d49cb434dbff78eafcd48c2bff986b15a", size = 501238, upload-time = "2026-05-25T22:17:12.728Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ca/42/906adc91ae3a5fa9c59c0a2f21c139725bd7e5b41ae6acd485cd14123ebf/httptools-0.8.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a1afd7c9fbff0d9f5d489c4ce2768bd09c84a46ddefc7161e6aa82ae35c85745", size = 509567, upload-time = "2026-05-25T22:17:13.842Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/05/0b/4240efeb672751ee5b9b380cb0e3fdc050bc05f68adc7a8aefc4fcd9a69a/httptools-0.8.0-cp312-cp312-win_amd64.whl", hash = "sha256:cd96f29b4bab1d42fa6e3d008711c75e0f79e94e06827330160e3a304227f150", size = 90918, upload-time = "2026-05-25T22:17:15.155Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/5e/e5/8cfcabc5546e8022f168be28bcdaa128a240a0befdd03b59d558b4f18bd6/httptools-0.8.0-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:614ceea8ea606848bece2338ac03b3ce5324bcb4be8dc7d377ed708012fa4db8", size = 205148, upload-time = "2026-05-25T22:17:16.333Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/2a/0e/0fb14848c19a686c8062ff9067c1a48793e3224b47bc5b201535b6036fce/httptools-0.8.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:2d689918c15a013c65ef52d9fd495d766893ab831a2c8d89f2ac5940a5df847c", size = 111368, upload-time = "2026-05-25T22:17:17.586Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/2e/1b/46f1cecf06b9bbde8e4b8c88034ac7908989e5ff7a3a388ef38392949c1f/httptools-0.8.0-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:eb3028cca2fc0a6d720e52ef61d8ebb62fcbfeb1de56874546d858d3f25a26b7", size = 486447, upload-time = "2026-05-25T22:17:18.564Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/77/00/258bfc0837221f81d9725c45f9b948a6a6b2994a147a4fb66e85100c668f/httptools-0.8.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:88bdd940f2b5d487b4d032c6afa5489a7dc4694410d43de3c38c4fb3af0dc45d", size = 482448, upload-time = "2026-05-25T22:17:19.912Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/04/ab/d1cef3b5523f4d272a70f42a776c3169a2dddfe3a54de4b2ce4a36341528/httptools-0.8.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:6a43c9dd399758ccc0531acb0a3c4a6c299ee893ee9400e9c893b7bdcfae0681", size = 464460, upload-time = "2026-05-25T22:17:20.882Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ce/48/5d1d072442277bb2b3434e0e60690b8e8c23840ef7de8b6ea54040a536d3/httptools-0.8.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:0770728beb05094c809b98e814edff5fef69d26ad7d21185f2f6d5884a0ba683", size = 471312, upload-time = "2026-05-25T22:17:22.085Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/0d/66/b96623b27e51a68199ef4efdda0613cced9233fe3062ac74e50749c5ad37/httptools-0.8.0-cp313-cp313-win_amd64.whl", hash = "sha256:7685df791fad561384bfb139e77fde27a1ffd93134e016f95a0db424ffbf77b1", size = 90117, upload-time = "2026-05-25T22:17:23.074Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/1a/12/fa3fbf5f9517b273edea2dc982aa82a8c634091e67c590792b729017bc6f/httptools-0.8.0-cp314-cp314-macosx_10_13_universal2.whl", hash = "sha256:de242a49b5d18e0a8776e654e9f6bf6d89f3875a5c35b425a0e7ce940feb3fd6", size = 206183, upload-time = "2026-05-25T22:17:24.004Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/30/fc/5e7c4cb443370f2090a3aba0453a07384d29ff66b7435bb90e77e1037599/httptools-0.8.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:159e9ab5f701ccd42e555a12f1ad8ff69702910fc1c996cf2bb66e5fcb7a231b", size = 112079, upload-time = "2026-05-25T22:17:25.216Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/ba/53/771bd891eb0f236f32145d6a1775777ec85745f3cc983a1f23d1a3b8ddfe/httptools-0.8.0-cp314-cp314-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:c4a9f1707e4823d54dfec6c33fa3697d302aed536ed352a7ebb5a061ddb869d0", size = 481596, upload-time = "2026-05-25T22:17:26.186Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/62/42/94e15bc68ce3d423243c45d7f1b0c7561f13844f97dc52ae23182fb65628/httptools-0.8.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d76ad7b951387e3632c8716a9bb03ac5b45c5f16119aa409db0459520887944e", size = 480865, upload-time = "2026-05-25T22:17:27.542Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/1c/7c/fe2980fc03723272e30f135b62360b075f513dfe7cc73aef36c7f04012bd/httptools-0.8.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:a3b7387147361c3fd47a0bde763c5c91b5b4cd4dc9989b8ece84ff436c99843b", size = 463189, upload-time = "2026-05-25T22:17:28.546Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/15/1b/47fc5fff68acd1bfa20b4734059c9a06cadb88119dcd5258b5b0d21d91c8/httptools-0.8.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:f256d6ce930c52ca1cb2a960b7da03548c454e7d28b06059ad41bfe789036ce0", size = 466610, upload-time = "2026-05-25T22:17:29.816Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/60/bd/07b13c93ffd9bec9546e0d43f8e19378dd696dbd278511406bc07371ef1f/httptools-0.8.0-cp314-cp314-win_amd64.whl", hash = "sha256:19d1ee275bb59ba2643ba9a3a1e51cc0c788caf2b8df506368e03f56fdd08527", size = 92705, upload-time = "2026-05-25T22:17:31.133Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fd/c4/121648f68ce066d7bd762d6b6d97e620847642d38d54f3d90ff11d947629/httptools-0.8.0-cp314-cp314t-macosx_10_13_universal2.whl", hash = "sha256:de1ed58a974e75d56560acc7e7fed01a454994429456f65209789992e41f2568", size = 215023, upload-time = "2026-05-25T22:17:32.401Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/b9/b0/312a062ae741ae3e8baa8c8bf20be81b2e67337b259ab4349bebc7b6142e/httptools-0.8.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:e93c227b595c6926c1acee96891dd9da4be338cfbe82e5cd3bb9d8dd7dc4ac0b", size = 117405, upload-time = "2026-05-25T22:17:33.742Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/fc/37/fccd705f795386bb05bf413012fecff2a33e5aa8c2f069096de3e9fd8702/httptools-0.8.0-cp314-cp314t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:2a021c3a8e65cc125390d72f59b968afca3bdcaff25bd67965e0a055a14946ca", size = 558497, upload-time = "2026-05-25T22:17:34.732Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/bd/39/f172e8003576de35f5ba77ff417cf0e34429d35dc014deef15afa337a72c/httptools-0.8.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:48774d39cbb70e2b1f71f88852a3087ae1d3a1eb80482bb48c13067ab080c14f", size = 571585, upload-time = "2026-05-25T22:17:35.813Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/3e/b9/f5564760af99f3dbbf3f9104dc00e5da27e96cf433c6bdcf77617f70bf3f/httptools-0.8.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:88eead8ec8680a9f146c655bc88445a325bd7921cfd8194c7337e9467282427d", size = 543297, upload-time = "2026-05-25T22:17:37.08Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/99/67/8d9f2c313618e161b82f3873188e7196126da1d6e29688df40eb3997c77a/httptools-0.8.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:2c032fa028f46871ec7e1fc59fc15e8023eab3e6bbe6ece786a1611719a5d081", size = 539535, upload-time = "2026-05-25T22:17:38.032Z" }, + { url = "https://mirrors.tuna.tsinghua.edu.cn/pypi/web/packages/48/63/b906c01e53f50d432c0defe43ce52764a111dc1bdd028bafbeb54dcfd008/httptools-0.8.0-cp314-cp314t-win_amd64.whl", hash = "sha256:384c17174464c8e873398b7af24f0b1f44d992c820328413951a625323155d77", size = 108209, upload-time = "2026-05-25T22:17:39.473Z" }, +] + [[package]] name = "httpx" version = "0.28.1"