diff --git a/README.md b/README.md index 5d13a6c..42838f9 100644 --- a/README.md +++ b/README.md @@ -113,11 +113,22 @@ uv sync # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手 **修改 `utils/`（工具包）**： -1. 在 `utils/` 内提交并 push 到对应远程 +1. 在 `utils/` 内提交并 push 到对应远程（并发布新版本，见下） 2. 回主仓库执行 `uv lock --upgrade-package <包名>` 刷新版本 pin 3. `uv sync` 后运行相关测试 4. 主仓库提交 `uv.lock` +> ⚠️ **打包陷阱（"改了 utils/* 但产物行为没变"几乎都是它）**：`uv run build` 会先按 +> 锁文件同步环境，**`uv pip install -e utils/` 的可编辑安装会被替换回锁文件里的 +> 发布版**。因此本地改了工具包后，必须走上面 1-4 步（push + 发版 → 刷新 pin → sync）， +> 否则 `build/` 产物里仍是旧版代码。仅本地联调时可临时用可编辑安装，但打包前要确认 +> venv 指向的是准备发布的版本： +> +> ```bash +> uv pip install -e utils/worker --no-deps # 仅联调：让 venv 用本地源码 +> python -c "import chongming_worker, inspect; print(inspect.getsourcefile(chongming_worker))" +> ``` + **修改 `src/plugins/` / `config/` / `src/chongming-web/`**： 1. 在对应子仓库内提交并 push @@ -127,6 +138,8 @@ uv sync # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手 - [ ] 所有子仓库 `git status` 干净且已 push - [ ] `uv.lock` 与各 git 源最新提交一致 +- [ ] 打包前确认 venv 里 `chongming-worker` 等工具包为目标版本（改了 `utils/*` 却仍是旧版时， + 产物行为不会变，见上一节打包陷阱） - [ ] 配置中无硬编码凭据——密钥一律走环境变量： `VOLC_ACCESSKEY`（豆包 Ark Key）、`AIBOX_MQTT_URI`（AI Box 外部 Broker URI） - [ ] `uv run python test_mqtt_e2e.py` 等测试通过 @@ -297,6 +310,8 @@ worker///__admin__ 两种模式都不回写 `config.toml`：管理接口的改动只进缓存。`cache` 模式下启动日志会打印 toml 与缓存的差异键（以缓存为准）；如需让 toml 的改动生效，先删除 `worker:cm:configs:`（运行时配置）或 `worker:cm:counts:`（副本数）缓存键再重启，或改用 `"toml"` 模式。`GET /admin/workers//config` 的响应带 `source`（cache/toml）与 `mode`，前端据此标注展示值的来路。 +- **对齐时机（两层，规则一致且幂等）**：inline 插件由**网关**在创建副本之前对齐——`WorkerManager.align_inline_config()` 在收到展平配置与 `[workers.config]` 原始段后按模式取缓存/写缓存，再交给插件的 `create_workers`；worker SDK 的 `Plugin.create_workers` 在 `_bootstrap()`/`on_init()` 之前用 `resolve_startup_config()` 再解析一次。网关侧这层是为了让网关与 worker 包版本解耦：worker 包仍是旧版时 cache 模式同样生效（否则旧版 `on_init` 只拿到 toml 展平配置，表现为"前端显示缓存值、实际跑 toml"）。**external 插件**没有网关这层兜底，由插件进程内的 SDK 解析，需要 worker 包为含该改动的版本。 + - **进程插件（external）实例粒度参与分摊**：网关只对 **inline** 插件用 cache 解析启动副本数（见「启动流程」第 5 步 `load_workers`），external 插件的启动副本数由外部编排按自身配置决定；但**运行时的缩放**上，external 并不退化为“每进程全额”，而是同样在**共享 cache 上登记实例参与方**来分摊，避免多实例把副本翻倍： - **实例登记 + 心跳租约**：每个 external 进程启动时向 `worker:cm:instances:` 登记自身参与方（`[instance_id, ts]`），并每 15s 心跳续约；超过租约（60s）未心跳的死实例会被清理。`divmod` 按存活参与方 N 均分全局 count（余量补靠前实例），`N==1` 时退化为全额（单部署零影响）。 - **实例标识默认唯一**：`--instance` / 环境变量 `CHONGMING_WORKER_INSTANCE` 显式指定（跨重启稳定、逐实例精细扩缩可持续命中）；缺省回退 `{hostname}:{pid}`——**同一机器并行多个未指定 `--instance` 的进程也能靠 pid 区分**，不会因为 id 相同被并成一个参与方导致副本翻倍或前端只显示一个实例。注意缺省 id 随进程重启变化。 diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index c7435e5..40fc38f 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -399,6 +399,9 @@ def _discover_plugins() -> list[dict]: "load_mode": load_mode, "mqtt_topics": mqtt_topics, "config": runtime_config, + # [workers.config] 展平前的原始段：供 load_workers 按「配置优先来源模式」 + # 引导/覆盖共享缓存，使 inline 插件初始化即拿到缓存值（cache 模式） + "config_section": dict(workers_section.get("config") or {}), }) # 2. 兼容旧写法: config/workers/config.toml（仅加载尚未发现的插件） @@ -475,7 +478,12 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana # 否则用配置值引导并写回 cache（多网关共享同一 cache 时自动对齐）。 count = wc["count"] worker_type = wc["type"] - worker_config = wc["config"] + # [workers.config] 单一事实来源：按配置优先来源模式对齐（cache 模式取缓存值、 + # toml 模式以文件为准并覆盖缓存），必须在注册 starter/创建副本之前完成—— + # 插件实例（on_init 重建客户端等）用的就是这份配置。 + worker_config = manager.align_inline_config( + worker_type, wc["config"], wc.get("config_section") + ) if not module_name or not factory_name: logger.warning(f"Worker 配置不完整: {wc}") diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py index af7d768..b763b73 100644 --- a/src/chongming/app/core/worker_manager.py +++ b/src/chongming/app/core/worker_manager.py @@ -14,6 +14,11 @@ from .logger import get_logger logger = get_logger("worker_manager") +# ── 插件运行时配置（chongming-cache 单一事实来源）── +# 键与 worker SDK（chongming_worker.runtime._configs_cache_key）一致，全局唯一一份 +# [workers.config]；读写规则见 WorkerManager.align_inline_config。 +_CONFIGS_KEY = "worker:cm:configs:{type}" + # ── 全局副本数（chongming-cache 单一事实来源）── # 多网关共用同一 cache（单实例=本地 diskcache，多实例=共享卷/redis）， # 「count」语义 = 该类 worker 的**全局目标总量**，各网关按在线数量分摊。 @@ -134,6 +139,51 @@ class WorkerManager: """ return "cache" if str(self._launch_worker_count).lower() == "cache" else "toml" + def align_inline_config( + self, + worker_type: str, + worker_config: Dict[str, Any], + config_section: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """inline 插件初始化前，按配置优先来源模式对齐其 [workers.config]（网关侧兜底）。 + + worker SDK 的 Plugin.create_workers 会做同样的事（权威实现，见 + chongming_worker.runtime.resolve_startup_config）；网关侧再兜一层，是为了让 + 网关（随本仓库发布）与 worker 包版本解耦：worker 包仍是旧版时，旧版 on_init + 只拿到 config/workers//config.toml 展平出来的配置，会出现“前端显示缓存值、 + 实际跑 toml”。规则与 SDK 一致： + + - "cache"：缓存里有该段就用它覆盖传入配置（toml 仅作缓存缺失时的引导并写回， + 之后以缓存为准、跨重启保持）； + - "toml" ：以 config.toml 为准，把该段覆盖写入缓存（前端读缓存 → 随之显示文件值）； + - 未配置缓存后端（cache 为 None）时保持传入配置不动。 + + Args: + worker_type: 插件类型（如 aibox_sophon_adapter） + worker_config: _discover_plugins 展平出来的运行时配置 + config_section: 该插件 config.toml 的 [workers.config] 原始段（引导/覆盖缓存用） + """ + if self.cache is None: + return worker_config + key = _CONFIGS_KEY.format(type=worker_type) + section = config_section if isinstance(config_section, dict) else None + if self.config_mode == "cache": + cached = self.cache.get(key) + if isinstance(cached, dict): + logger.info( + "%s [workers.config] 取自缓存（cache 模式，%d 键）", + worker_type, len(cached), + ) + return {**worker_config, **cached} + if section: + self.cache[key] = dict(section) # 缓存缺失：用 toml 引导并写回 + logger.info("%s [workers.config] 缓存缺失，已用 config.toml 引导", worker_type) + return worker_config + if section is not None: + self.cache[key] = dict(section) # 以文件为准：覆盖缓存，前端随之显示文件值 + logger.info("%s [workers.config] 以 config.toml 为准并覆盖缓存（toml 模式）", worker_type) + return worker_config + # ── 全局副本数协调（chongming-cache 单一事实来源）──────────── # count/quota 的读写与分摊已收归 worker 层（chongming_worker.runtime）： # 网关不再直写这些缓存键，仅维护「在线网关列表」（register/heartbeat/unregister）。 diff --git a/test/test_inline_config_align.py b/test/test_inline_config_align.py new file mode 100644 index 0000000..af32881 --- /dev/null +++ b/test/test_inline_config_align.py @@ -0,0 +1,163 @@ +""" +inline 插件 [workers.config] 对齐（网关侧兜底）单测 —— WorkerManager.align_inline_config。 + +网关侧在创建 inline 副本之前，按「配置优先来源模式」（根 config.toml 的 +[cache].launch_worker_count）对齐插件配置：cache 模式取共享缓存里的段， +toml 模式以 config/workers//config.toml 为准并覆盖缓存。目的是让网关与 +worker 包版本解耦——worker 包仍是旧版时，旧版 on_init 只拿到 toml 展平配置， +会出现“前端显示缓存值、实际跑 toml”。 + +用 dict 作缓存桩，验证两种模式下插件拿到的配置与缓存/文件两侧的取值一致。 +""" +import os +import sys + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(REPO, "src")) +os.chdir(REPO) # 网关模块导入期读根 config.toml / 日志配置 + +from chongming.app.core.worker_manager import _CONFIGS_KEY, WorkerManager # noqa: E402 + +PASS = 0 +FAIL = 0 +WT = "aibox_sophon_adapter" + +# _discover_plugins 展平出来的形态（toml 值） +FLAT = {"type": WT, "count": 1, "data_dir": "./data", "mediamtx_addr": "rtsp://toml:8554"} +# 该插件 config.toml 的 [workers.config] 原始段 +SECTION = {"data_dir": "./data", "mediamtx_addr": "rtsp://toml:8554", "heartbeat": {"timeout": 15}} +# 前端下发后缓存里的段（改过 mediamtx_addr 与 heartbeat） +CACHED = {"data_dir": "./data", "mediamtx_addr": "rtsp://cache:9999", "heartbeat": {"timeout": 30}} + + +def report(name, cond, detail=""): + global PASS, FAIL + if cond: + PASS += 1 + print(f"✅ {name}" + (f" — {detail}" if detail else "")) + else: + FAIL += 1 + print(f"❌ {name}" + (f" — {detail}" if detail else "")) + + +def manager(cache, mode): + m = WorkerManager.__new__(WorkerManager) # 免会话构造：仅用到 cache + config_mode + m.cache = cache + m._launch_worker_count = mode + return m + + +def run() -> int: + key = _CONFIGS_KEY.format(type=WT) + + # ── 1. cache 模式：缓存有段 → 插件拿到缓存值（本次修复目标）────────── + cache = {key: dict(CACHED)} + got = manager(cache, "cache").align_inline_config(WT, dict(FLAT), dict(SECTION)) + report( + "cache 模式：插件配置取缓存值（覆盖 toml 同名键）", + got.get("mediamtx_addr") == "rtsp://cache:9999" + and got.get("heartbeat") == {"timeout": 30}, + f"mediamtx_addr={got.get('mediamtx_addr')}", + ) + report("cache 模式：不覆盖已存在的缓存段", cache[key] == CACHED, f"cache={cache[key]}") + + # ── 2. cache 模式：缓存缺失 → 用 toml 段引导并写回 ────────────────── + cache2 = {} + got2 = manager(cache2, "cache").align_inline_config(WT, dict(FLAT), dict(SECTION)) + report( + "cache 模式：缓存缺失用 config.toml 引导并写回", + cache2.get(key) == SECTION and got2.get("mediamtx_addr") == "rtsp://toml:8554", + f"cache={cache2.get(key)}", + ) + + # ── 3. toml 模式：以文件为准并覆盖缓存（前端随之显示文件值）───────── + cache3 = {key: dict(CACHED)} + got3 = manager(cache3, "toml").align_inline_config(WT, dict(FLAT), dict(SECTION)) + report( + "toml 模式：插件配置为 toml 值", + got3.get("mediamtx_addr") == "rtsp://toml:8554", + f"mediamtx_addr={got3.get('mediamtx_addr')}", + ) + report( + "toml 模式：启动时用 toml 覆盖缓存（前端与运行值一致）", + cache3.get(key) == SECTION, + f"cache={cache3.get(key)}", + ) + + # ── 4. 边界：无缓存后端 / 无段（旧 config/workers/config.toml 兼容路径）── + cache4 = {key: dict(CACHED)} + m4 = manager(cache4, "cache") + m4.cache = None + report( + "无缓存后端：原样返回（经典 toml 行为）", + m4.align_inline_config(WT, dict(FLAT), dict(SECTION)) == FLAT + and cache4[key] == CACHED, + ) + cache5 = {} + got5 = manager(cache5, "toml").align_inline_config(WT, dict(FLAT), None) + report("无 [workers.config] 段（旧写法）：不写缓存、原样返回", got5 == FLAT and not cache5) + + # ── 5. 与 SDK 侧一致：缓存键必须相同；SDK 为本地版本时才比对解析结果 ── + # 注意：venv 里的 chongming-worker 可能是索引里的发布版（uv sync 会把可编辑安装 + # 换回锁文件版本），新版才有 resolve_startup_config / 正确的 toml 分支语义。 + sys.path.insert(0, os.path.join(REPO, "utils", "worker", "src")) + import chongming_worker.runtime as r # noqa: E402 + + report( + "缓存键与 SDK 一致（两端必须读写同一键）", + r._configs_cache_key(WT) == _CONFIGS_KEY.format(type=WT), + f"sdk={r._configs_cache_key(WT)}", + ) + local_src = os.path.join(REPO, "utils", "worker", "src") in os.path.abspath(r.__file__) + if not local_src: + print(f"⚠️ 跳过解析结果比对：当前 SDK 为 {r.__file__}（非本地源码，缺本次改动）") + else: + c = {r._configs_cache_key(WT): dict(CACHED)} + sdk = r.resolve_runtime_config(WT, dict(FLAT), c, "cache") + report( + "cache 模式：网关侧兜底与 SDK 解析结果一致", + sdk.get("mediamtx_addr") == CACHED["mediamtx_addr"], + f"sdk={sdk.get('mediamtx_addr')}", + ) + + # ── 6. 端到端：网关对齐 + 插件初始化（部署组合：网关新、worker 包为发布版）── + # 该断言即线上现象的复现与验证：插件 on_init 拿到的必须是网关对齐后的缓存值， + # 与 venv 里 chongming-worker 是否为含本次改动的版本无关。 + import asyncio + + from chongming_worker.plugin import Plugin # noqa: E402 + + seen = {} + + class DemoPlugin(Plugin): + type = WT + auto_bind = False + + async def _bootstrap(self): # 隔离数据库/对象存储引导，只看配置来源 + return None + + async def on_init(self): + seen["on_init"] = self.config.get("mediamtx_addr") + + def _create_workers_impl(self, session, config, count=1, worker_manager=None): + seen["impl"] = (config or {}).get("mediamtx_addr") + return [] + + aligned = manager({key: dict(CACHED)}, "cache").align_inline_config( + WT, dict(FLAT), dict(SECTION) + ) + asyncio.run(DemoPlugin().create_workers(None, aligned, 1)) + report( + "端到端：网关对齐后插件 on_init 读到缓存值（不依赖 worker 包版本）", + seen.get("on_init") == CACHED["mediamtx_addr"] == seen.get("impl"), + f"on_init={seen.get('on_init')}, impl={seen.get('impl')}, " + f"sdk={os.path.abspath(__import__('chongming_worker').__file__)}", + ) + + print("\n" + "=" * 56) + print(f"测试结果: ✅ {PASS} 通过 ❌ {FAIL} 失败") + return 0 if FAIL == 0 else 1 + + +if __name__ == "__main__": + sys.exit(run())