diff --git a/config.toml b/config.toml index bedad03..0faee59 100644 --- a/config.toml +++ b/config.toml @@ -88,6 +88,9 @@ base = [ "utils.config", "zenoh", "chongming_worker", + "chongming_worker.contracts", + "chongming_worker.route_config", + "chongming_worker.routes_publisher", "chongming_request", "server", "module_bank", # 模块化打包工具 @@ -96,6 +99,7 @@ base = [ "fastapi", # web 应用依赖 "fastapi.staticfiles", "fastapi.middleware.cors", + "fastapi.middleware.gzip", "uvicorn", "apscheduler", "diskcache", diff --git a/public/utils/launch.py b/public/utils/launch.py index 1084601..6db5547 100644 --- a/public/utils/launch.py +++ b/public/utils/launch.py @@ -3,6 +3,12 @@ import os from module_bank import PythonToSQLite from typing import Callable +# module_bank 依赖在每次加载 PyArmor 加密模块时都会 print 一行 +# "已部署 PyArmor 运行时"，启动期会刷屏（约 50 行）。依赖包不便改动， +# 这里用模块级 print 遮蔽（模块内名字解析 globals 优先于 builtins）静默掉。 +import module_bank.sqlite_module_loader as _sqlite_module_loader + +_sqlite_module_loader.print = lambda *args, **kwargs: None def init_module_bank(config): @@ -10,7 +16,7 @@ def init_module_bank(config): 初始化模块银行系统，将指定路径下的Python模块打包并安装导入器。 该函数会验证所有模块路径的有效性，创建PythonToSQLite打包器实例， - 列出所有可用模块，并为每个打包器安装导入器以支持模块的动态加载。 + 并为每个打包器安装导入器以支持模块的动态加载。 Args: config (dict): 配置字典，必须包含 'production' -> 'module_system' -> 'path' 键， @@ -28,12 +34,9 @@ def init_module_bank(config): packer_list = [ PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST ] - # 遍历所有打包器，列出模块信息并安装导入器 + # 遍历所有打包器，安装导入器（模块清单不再打印，避免启动刷屏） for packer in packer_list: key = packer.get_key() - modules = packer.list_modules() - for module in modules: - print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") packer.install_importer(key=key) diff --git a/src/chongming/app/core/call_stats.py b/src/chongming/app/core/call_stats.py index b1e100e..1a64a6e 100644 --- a/src/chongming/app/core/call_stats.py +++ b/src/chongming/app/core/call_stats.py @@ -15,10 +15,13 @@ record_call 内部全兜底：统计或审计任一步失败仅记 warning，绝 import asyncio import datetime import orjson +import random import time from collections import deque from typing import Any, Coroutine, Dict +from chongming_config import get_config + from .logger import get_logger logger = get_logger("call_stats") @@ -28,6 +31,22 @@ audit_logger = get_logger("audit") # 最近调用环形缓冲上限 MAX_RECENT = 1000 +# 审计日志采样率：与 access 日志同源配置（access_sample_rate，0.1 = 只写 10% 的成功请求； +# 错误 status >= 400 始终全量）。压测实测全量审计写盘 8.8MB/4.3 万行，是高吞吐下的隐藏 I/O 大头。 +_audit_config = get_config() +_audit_env = _audit_config["default"]["env"] +_AUDIT_SAMPLE_RATE = float( + _audit_config[_audit_env].get("logging", {}).get("access_sample_rate", 1.0) +) + + +def _should_record_audit(status: int) -> bool: + """审计日志采样判定（与 access 日志同模式）""" + if status >= 400 or _AUDIT_SAMPLE_RATE >= 1.0: + return True + denom = max(1, int(round(1.0 / _AUDIT_SAMPLE_RATE))) + return random.randrange(denom) == 0 + # 后台任务集合（fire-and-forget，shutdown 时统一取消） _tasks: set[asyncio.Task] = set() @@ -145,6 +164,7 @@ async def record_call(entry: Dict[str, Any]) -> None: except Exception as e: # pragma: no cover - 防御性兜底 logger.warning("调用统计记录失败: %s", e) try: - audit_logger.info(_format_audit(entry)) + if _should_record_audit(entry.get("status", 200)): + audit_logger.info(_format_audit(entry)) except Exception as e: # pragma: no cover - 防御性兜底 logger.warning("审计日志写入失败: %s", e) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 33409f0..2ff0ae6 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -83,6 +83,14 @@ def _run_mqtt_broker(host: str, port: int, username: str | None = None, import asyncio as _asyncio from amqtt.broker import Broker + # amqtt 在 broker 进程内无 handler，WARNING 会经 lastResort 直接刷到 stderr： + # - sys_interval=0（禁用 $SYS）被 amqtt 当 falsy 报警 + # - 客户端连上却不发 CONNECT 时每次刷 2 行 "No data from"/"Failed to initialize..." + # 统一压到 ERROR：认证失败本身是 DEBUG 级不受影响，真实故障仍会输出 + import logging as _logging + + _logging.getLogger("amqtt").setLevel(_logging.ERROR) + # ---- monkey-patch 1: amqtt StreamWriterAdapter.close() 忽略断开的传输层 ---- from amqtt.adapters import StreamWriterAdapter as _SWA @@ -209,7 +217,7 @@ def _run_mqtt_broker(host: str, port: int, username: str | None = None, except OSError: pass auth_note = f" (认证用户: {username})" if username else " (匿名)" - print(f"[MQTT Broker] 已启动 @ {host}:{port}{auth_note}") + logger.info("MQTT Broker 已启动 @ %s:%s%s", host, port, auth_note) try: while True: await _asyncio.sleep(1) diff --git a/src/chongming/app/core/logger.py b/src/chongming/app/core/logger.py index b8a1129..92688b0 100644 --- a/src/chongming/app/core/logger.py +++ b/src/chongming/app/core/logger.py @@ -145,6 +145,24 @@ class DailyRotatingFileHandler(RotatingFileHandler): super().emit(record) +class LoggerNameFilter(logging.Filter): + """ + 按 log_name 路由日志记录的过滤器 + + 只放行本 log_name 自身及其子 logger 的记录（如 chongming_worker.runtime + 归入 chongming_worker），保证共享 QueueListener 下每个 handler 只处理 + 属于自己的记录：控制台每行只打印一次，各日志文件按名隔离。 + """ + + def __init__(self, log_name: str): + super().__init__() + self._name = log_name + self._prefix = log_name + "." + + def filter(self, record): + return record.name == self._name or record.name.startswith(self._prefix) + + class ExcludeWatchFilesFilter(logging.Filter): """ 日志过滤器，用于排除第三方库的日志输出 @@ -279,6 +297,8 @@ atexit.register(shutdown_logging) _configured_loggers = set() +# 每个进程只打印一次日志配置信息（setup_logging 每个 log_name 都会调用一次） +_config_info_printed = False def setup_logging(logger: logging.Logger, log_name: str = "app"): @@ -332,11 +352,13 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"): ) file_handler.setFormatter(plain_formatter) file_handler.addFilter(ExcludeWatchFilesFilter()) + file_handler.addFilter(LoggerNameFilter(log_name)) # 创建控制台处理器并添加过滤器 console_handler = logging.StreamHandler() console_handler.setFormatter(color_formatter) console_handler.addFilter(ExcludeWatchFilesFilter()) + console_handler.addFilter(LoggerNameFilter(log_name)) # 真实 handler（文件/控制台）挂到后台监听线程；logger 上只挂队列 handler， # 日志调用在主线程仅入队，I/O 由监听线程执行 @@ -347,10 +369,13 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"): _listener.register(file_handler, console_handler) logger.addHandler(_AsyncQueueHandler()) - print( - f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} " - f"(最大：{max_bytes / 1024 / 1024:.1f}MB, 备份：{backup_count})" - ) + global _config_info_printed + if not _config_info_printed: + _config_info_printed = True + print( + f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} " + f"(最大：{max_bytes / 1024 / 1024:.1f}MB, 备份：{backup_count})" + ) def get_logger(name: str) -> logging.Logger: diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py index b7444f8..6af4239 100644 --- a/src/chongming/app/core/worker_manager.py +++ b/src/chongming/app/core/worker_manager.py @@ -110,9 +110,12 @@ class WorkerManager: await asyncio.sleep(self._refresh_interval) async def get_worker(self, worker_type: str) -> Optional[str]: - """获取一个可用的 worker 地址（简单随机负载均衡）""" - async with self._lock: - workers = self._workers_cache.get(worker_type, []) + """获取一个可用的 worker 地址（简单随机负载均衡） + + 读路径无锁：_workers_cache 由 refresh_workers 整表原子替换， + _local_workers 由同步方法修改，单事件循环内读取安全。 + """ + workers = self._workers_cache.get(worker_type, []) # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时） if not workers: local = self._local_workers.get(worker_type, []) @@ -134,8 +137,7 @@ class WorkerManager: Returns: 可用的 worker key_expr，若无其他可用则返回 None """ - async with self._lock: - workers = self._workers_cache.get(worker_type, []) + workers = self._workers_cache.get(worker_type, []) candidates = [w for w in workers if w not in exclude] if candidates: return random.choice(candidates) @@ -162,8 +164,7 @@ class WorkerManager: 供 call() / call_stream() 共用，保证两类调用错误信息一致。 """ - async with self._lock: - all_known = list(self._workers_cache.keys()) + all_known = list(self._workers_cache.keys()) local_known = list(self._local_workers.keys()) all_types = sorted(set(all_known + local_known)) msg = ( @@ -289,7 +290,10 @@ class WorkerManager: Returns: Dict[str, Any]: 调用结果（worker 回复解析后的 dict） """ - cache_key = f"{worker_type}:{subject}:{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}" + # cacheable=False 时不计算 cache_key，省一次排序序列化 + cache_key = None + if cacheable: + 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, @@ -325,7 +329,10 @@ class WorkerManager: Returns: bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致） """ - cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}" + # cacheable=False 时不计算 cache_key（当前唯一用途），省一次 sha256 + cache_key = None + if cacheable: + cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}" return await self._call_impl( worker_type, subject, payload, timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key, diff --git a/test/bench_http.py b/test/bench_http.py new file mode 100644 index 0000000..6735be6 --- /dev/null +++ b/test/bench_http.py @@ -0,0 +1,203 @@ +""" +HTTP 端到端 RPC 压测脚本（chongming 网关全链路） + +链路: 压测客户端(asyncio+httpx) → FastAPI 网关 :8000 → Zenoh Query/Reply → worker 进程 + +前置启动（均后台运行，需在仓库根目录）: + 1. Zenoh router: uv run python utils/worker-rs/scripts/zenoh_router.py + 2. 网关(无热重载): uv run uvicorn chongming.app:app --host 0.0.0.0 --port 8000 --no-access-log + 3. Worker: uv run worker_serve --type demo --count 2 + +运行: + uv run python test/bench_http.py --concurrency 1,10,50,100,200 --duration 30 + +指标: RPS(总请求/秒)、p50/p95/p99/max 延迟(成功请求, ms)、错误率(非 2xx/超时/连接异常)、 + 结束后拉取网关侧 GET /api/v1/stats 做服务端对照。 +""" +import argparse +import asyncio +import json +import statistics +import sys +import time +from dataclasses import dataclass, field + +import httpx + +DEFAULT_URL = "http://localhost:8000/api/v1/demo/hello?name=bench" + + +@dataclass +class Bench: + """单档压测共享状态（asyncio 单线程内访问，无需加锁）""" + start: float = 0.0 + total: int = 0 # 全部完成的请求（含错误/超时） + ok: int = 0 # 2xx 成功请求 + timeouts: int = 0 + errors: int = 0 # 非 2xx / 连接异常 + status_dist: dict = field(default_factory=dict) + lats_ms: list = field(default_factory=list) # 成功请求延迟(ms) + + +def percentile(sorted_lats: list, p: float) -> float: + """有序列表上的线性插值分位数（p 为 0~1）""" + if not sorted_lats: + return 0.0 + idx = (len(sorted_lats) - 1) * p + lo = int(idx) + hi = min(lo + 1, len(sorted_lats) - 1) + frac = idx - lo + return sorted_lats[lo] * (1 - frac) + sorted_lats[hi] * frac + + +async def worker(client: httpx.AsyncClient, url: str, bench: Bench, end_time: float) -> None: + loop = asyncio.get_running_loop() + while True: + t0 = loop.time() + if t0 >= end_time: + return + try: + resp = await client.get(url, timeout=10.0) + except httpx.TimeoutException: + bench.timeouts += 1 + continue + except httpx.HTTPError: + bench.errors += 1 + continue + lat_ms = (loop.time() - t0) * 1000 + if resp.status_code == 200: + bench.ok += 1 + bench.lats_ms.append(lat_ms) + else: + bench.errors += 1 + bench.status_dist[resp.status_code] = bench.status_dist.get(resp.status_code, 0) + 1 + bench.total += 1 + + +async def progress_reporter(bench: Bench, duration: float) -> None: + """每 5s 打印进度（RPS/成功/错误），供长压测观测""" + while True: + await asyncio.sleep(5) + elapsed = time.perf_counter() - bench.start + if elapsed <= 0: + continue + print(f" [{elapsed:5.0f}s] RPS={bench.total / elapsed:7.1f} " + f"ok={bench.ok} err={bench.errors} timeout={bench.timeouts}", flush=True) + + +async def run_level(client: httpx.AsyncClient, url: str, concurrency: int, + duration: float, warmup: float) -> dict: + """单档并发：warmup 预热（结果丢弃）→ duration 计时统计""" + bench = Bench() + + async def phase(end_time: float) -> None: + tasks = [asyncio.create_task(worker(client, url, bench, end_time)) + for _ in range(concurrency)] + await asyncio.gather(*tasks) + + loop = asyncio.get_running_loop() + print(f" 预热 {warmup:.0f}s ...", flush=True) + bench.start = time.perf_counter() + await phase(loop.time() + warmup) + + bench = Bench() # 丢弃预热结果，正式计时 + bench.start = time.perf_counter() + print(f" 压测 {duration:.0f}s (并发 {concurrency}) ...", flush=True) + reporter = asyncio.create_task(progress_reporter(bench, duration)) + await phase(loop.time() + duration) + reporter.cancel() + await asyncio.gather(reporter, return_exceptions=True) + + elapsed = time.perf_counter() - bench.start + sorted_lats = sorted(bench.lats_ms) + total = bench.total + return { + "concurrency": concurrency, + "duration_s": round(elapsed, 2), + "total": total, + "ok": bench.ok, + "errors": bench.errors, + "timeouts": bench.timeouts, + "error_rate_pct": round((bench.errors + bench.timeouts) / total * 100, 2) if total else 0.0, + "rps": round(total / elapsed, 1) if elapsed else 0.0, + "p50_ms": round(percentile(sorted_lats, 0.50), 2), + "p95_ms": round(percentile(sorted_lats, 0.95), 2), + "p99_ms": round(percentile(sorted_lats, 0.99), 2), + "max_ms": round(sorted_lats[-1], 2) if sorted_lats else 0.0, + "avg_ms": round(statistics.fmean(bench.lats_ms), 2) if bench.lats_ms else 0.0, + "status_dist": dict(sorted(bench.status_dist.items())), + } + + +async def fetch_stats_snapshot(client: httpx.AsyncClient) -> dict: + """拉取网关侧 /api/v1/stats 快照（失败返回空 dict）""" + try: + resp = await client.get("http://localhost:8000/api/v1/stats", timeout=5.0) + return resp.json() if resp.status_code == 200 else {} + except Exception: + return {} + + +async def main() -> int: + parser = argparse.ArgumentParser(description="chongming HTTP 端到端 RPC 压测") + parser.add_argument("--url", default=DEFAULT_URL, help="压测目标 URL") + parser.add_argument("--concurrency", default="100", help="并发数，逗号分隔多档（如 1,10,50,100,200）") + parser.add_argument("--duration", type=float, default=30.0, help="每档压测时长(秒)") + parser.add_argument("--warmup", type=float, default=3.0, help="每档预热时长(秒)") + parser.add_argument("--max-connections", type=int, default=100, + help="httpx 连接池上限（并发数超过此值时客户端会排队）") + parser.add_argument("--output", default=None, help="结果保存为 JSON 文件") + args = parser.parse_args() + + levels = [int(c.strip()) for c in args.concurrency.split(",") if c.strip()] + print(f"压测目标: {args.url}") + print(f"并发档位: {levels} | 每档 {args.duration:.0f}s + 预热 {args.warmup:.0f}s") + + results = [] + limits = httpx.Limits(max_connections=args.max_connections, + max_keepalive_connections=args.max_connections) + async with httpx.AsyncClient(http2=False, limits=limits) as client: + for level in levels: + stats_before = await fetch_stats_snapshot(client) + result = await run_level(client, args.url, level, args.duration, args.warmup) + stats_after = await fetch_stats_snapshot(client) + delta = max(0, stats_after.get("total_calls", 0) - stats_before.get("total_calls", 0)) + result["gateway_total_calls_delta"] = delta + results.append(result) + + # ---- 汇总表格 ---- + print("\n" + "=" * 88) + print(f"{'并发':>5} | {'RPS':>8} | {'avg(ms)':>8} | {'p50':>7} | {'p95':>7} | " + f"{'p99':>7} | {'max':>7} | {'错误率%':>7} | 网关侧计数差") + print("-" * 88) + for r in results: + print(f"{r['concurrency']:>5} | {r['rps']:>8.1f} | {r['avg_ms']:>8.2f} | {r['p50_ms']:>7.2f} | " + f"{r['p95_ms']:>7.2f} | {r['p99_ms']:>7.2f} | {r['max_ms']:>7.2f} | " + f"{r['error_rate_pct']:>7.2f} | {r.get('gateway_total_calls_delta', '-'):>10}") + print("=" * 88) + + # ---- 网关侧统计对照 ---- + async with httpx.AsyncClient() as client: + stats = await fetch_stats_snapshot(client) + if stats.get("latency"): + print("\n网关侧 /api/v1/stats 延迟统计（进程内聚合，全部档累计）:") + for lat in stats["latency"]: + print(f" worker_type={lat['worker_type']:<8} count={lat['count']:<8} " + f"avg={lat['avg_ms']:<8.2f}ms max={lat['max_ms']:.2f}ms") + print(f" 累计调用: {stats.get('total_calls', 0)}") + for c in stats.get("calls", []): + if c["count"]: + print(f" subject={c['subject']:<24} status={c['status']} count={c['count']}") + else: + print("\n（网关侧 /api/v1/stats 无数据或不可用）") + + if args.output: + with open(args.output, "w") as f: + json.dump(results, f, indent=2, ensure_ascii=False) + print(f"\n结果已保存: {args.output}") + + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/test/test_logging_async.py b/test/test_logging_async.py index e3c9c13..aa13ecf 100644 --- a/test/test_logging_async.py +++ b/test/test_logging_async.py @@ -17,14 +17,16 @@ 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")) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) +sys.path.insert(0, str(Path(__file__).resolve().parent.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_path = ( + Path(__file__).resolve().parent.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 @@ -103,7 +105,27 @@ def test_queue_full_drops(): logger_mod._log_queue = old -# ---- 4. 端到端：中间件采样 + 异步落盘 ---- +# ---- 4. 按名隔离：不同 log_name 的记录不串文件 ---- +def test_name_isolation(): + logger_mod.get_logger("test_b") + lg_a = logger_mod.get_logger("test_async") + lg_b = logger_mod.get_logger("test_b") + lg_a.info("隔离测试 A 专属 %d", 1) + lg_b.info("隔离测试 B 专属 %d", 2) + ok = _wait_for( + lambda: any( + "隔离测试 B 专属 2" in f.read_text(errors="ignore") + for f in Path(_tmp.name).glob("test_b-*.log") + ) + ) + check("test_b 记录写入 test_b 文件", ok) + a_text = "".join(f.read_text(errors="ignore") for f in Path(_tmp.name).glob("test_async-*.log")) + b_text = "".join(f.read_text(errors="ignore") for f in Path(_tmp.name).glob("test_b-*.log")) + check("test_async 记录不串入 test_b 文件", "隔离测试 A 专属" not in b_text) + check("test_b 记录不串入 test_async 文件", "隔离测试 B 专属" not in a_text) + + +# ---- 5. 端到端：中间件采样 + 异步落盘 ---- def test_e2e_middleware(): import asyncio import httpx @@ -130,6 +152,7 @@ def test_e2e_middleware(): test_sampling() test_async_queue_write() test_queue_full_drops() +test_name_isolation() test_e2e_middleware() # 关闭监听线程并释放文件句柄，再清理临时目录