modify: 日志重复问题

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",

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)
 
 

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:
+        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)

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)

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,6 +369,9 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"):
         _listener.register(file_handler, console_handler)
     logger.addHandler(_AsyncQueueHandler())
 
+    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})"

src/chongming/app/core/worker_manager.py

@@ -110,8 +110,11 @@ class WorkerManager:
             await asyncio.sleep(self._refresh_interval)
 
     async def get_worker(self, worker_type: str) -> Optional[str]:
-        """获取一个可用的 worker 地址（简单随机负载均衡）"""
-        async with self._lock:
+        """获取一个可用的 worker 地址（简单随机负载均衡）
+
+        读路径无锁：_workers_cache 由 refresh_workers 整表原子替换，
+        _local_workers 由同步方法修改，单事件循环内读取安全。
+        """
         workers = self._workers_cache.get(worker_type, [])
         # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时）
         if not workers:
@@ -134,7 +137,6 @@ class WorkerManager:
         Returns:
             可用的 worker key_expr，若无其他可用则返回 None
         """
-        async with self._lock:
         workers = self._workers_cache.get(worker_type, [])
         candidates = [w for w in workers if w not in exclude]
         if candidates:
@@ -162,7 +164,6 @@ class WorkerManager:
 
         供 call() / call_stream() 共用，保证两类调用错误信息一致。
         """
-        async with self._lock:
         all_known = list(self._workers_cache.keys())
         local_known = list(self._local_workers.keys())
         all_types = sorted(set(all_known + local_known))
@@ -289,6 +290,9 @@ class WorkerManager:
         Returns:
             Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
         """
+        # 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),
@@ -325,6 +329,9 @@ class WorkerManager:
         Returns:
             bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
         """
+        # 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,

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()))

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()
 
 # 关闭监听线程并释放文件句柄，再清理临时目录