diff --git a/README.md b/README.md index ed47735..2fda4da 100644 --- a/README.md +++ b/README.md @@ -340,7 +340,7 @@ uv run serve # 启动开发服务 uv run gen_plugin # 生成统一风格插件骨架 uv run python -m chongming.scripts.gen_models # 生成 Pydantic 模型 uv run python -m chongming.scripts.build # 构建打包（PyInstaller/Nuitka） -uv run worker_serve --type # 以独立进程运行 Worker（load_mode=process） +uv run worker_serve --type # 以独立进程运行 Worker（load_mode=external） uv pip install -e utils/ # 安装/更新工具包 ``` @@ -375,10 +375,12 @@ uv pip install -e utils/ # 安装/更新工具包 ## 配置说明 - **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置 -- **插件加载模式**：`config.toml` 的 `[.workers].load_mode` - - `"inline"`（默认）：插件在网关进程内加载 - - `"process"`：插件以独立进程运行（`uv run worker_serve --type `）， - 经 Zenoh 在 `chongming/routes/**` 注册路由，网关动态挂载 HTTP 端点 +- **插件加载模式**：按插件粒度声明在 `config/workers//config.toml` 的 `[workers]` 节 + - `load_mode = "inline"`（默认）：插件在网关进程内加载 + - `load_mode = "external"`：插件以独立进程运行（进程编排交给 Docker/K8s， + 进程入口 `uv run worker_serve --type `），经 Zenoh 在 `chongming/routes/**` + 注册路由，网关动态挂载 HTTP 端点 + - 两种模式可同时存在（混合部署） - **Zenoh 配置**：`config/zenoh_config.json5` - **插件注册**：`config/workers//config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容） - **路由定义**：`config/workers//*.toml`，每个端点一个文件 diff --git a/config.toml b/config.toml index b608d78..fef0ed3 100644 --- a/config.toml +++ b/config.toml @@ -24,11 +24,6 @@ secret_key = "your-secret-key-change-in-production" debug = true env = "development" -[development.workers] -# load_mode: "inline" = 网关进程内加载插件（现状）； -# "process" = 独立 worker 进程经 Zenoh 注册路由（uv run worker_serve --type ） -load_mode = "inline" - [development.server] access_log = false host = "0.0.0.0" diff --git a/src/chongming/__init__.py b/src/chongming/__init__.py index b81e351..7082f47 100644 --- a/src/chongming/__init__.py +++ b/src/chongming/__init__.py @@ -62,6 +62,13 @@ def worker_serve(): if wc is None: logger.error("未找到插件配置: %s（请检查 config/workers/%s/config.toml）", args.type, args.type) sys.exit(1) + if wc.get("load_mode", "inline") != "external": + logger.warning( + "[%s] 插件配置 load_mode 不是 \"external\"：网关也会在进程内加载该插件，" + "外部进程将与网关内副本重复服务。若以独立进程运行，请在插件 " + "config.toml 的 [workers] 节设置 load_mode = \"external\"。", + args.type, + ) async def _run(): # 1. Zenoh 会话 diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 3dce1c3..ee287f7 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -47,23 +47,24 @@ async def lifespan(app: FastAPI): # 4. WorkerManager manager = lifespan_setup.setup_worker_manager(app, session) - # 5. 加载 workers（inline 进程内加载 / process 独立进程 Zenoh 注册路由） - load_mode = lifespan_setup.get_load_mode() - worker_tasks = [] - custom_routers = [] - if load_mode == "process": - # process 模式：网关不 import 插件、不扫盘注册路由， - # 路由由独立 worker 进程经 chongming/routes/** 动态注册 - await lifespan_setup.setup_route_registry(app, session, manager) - else: - # inline 模式（默认）：插件在网关进程内加载，自动发现自定义路由 - worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager) + # 5. 插件加载（混合模式：inline 进程内加载；external 由独立进程运行， + # 其路由经 Zenoh 动态注册——进程编排交给 Docker/K8s，网关不拉起进程） + worker_tasks, custom_routers, inline_types, external_types = ( + await lifespan_setup.load_workers(app, session, manager) + ) + if external_types: + # 注册表为 external 插件服务；inline 类型集合防止误报重复挂载 + await lifespan_setup.setup_route_registry( + app, session, manager, inline_types=inline_types + ) - # 5.5 契约校验（TOML 路由 ↔ @rpc_method ↔ 生成模型），失败则启动终止 - lifespan_setup.validate_contracts(app) + # 5.5 契约校验（仅 inline 插件；external 插件在自己进程内校验），失败则启动终止 + lifespan_setup.validate_contracts(app) - # 6. 注册路由（含 TOML 动态路由 + 插件自定义路由） - await lifespan_setup.register_routes(app, manager, custom_routers) + # 6. 注册路由（inline 插件的 TOML 动态路由 + 插件自定义路由） + await lifespan_setup.register_routes( + app, manager, custom_routers, worker_types=inline_types + ) # 7. 启动 MQTT Bridge（内嵌 Broker + Worker 桥接） await lifespan_setup.setup_mqtt(app, manager) @@ -87,14 +88,13 @@ async def lifespan(app: FastAPI): # 1. 停止 MQTT Bridge await lifespan_setup.shutdown_mqtt(app) - # 1.5 process 模式：停止路由注册表订阅 + # 1.5 停止路由注册表订阅（存在 external 插件时） registry = getattr(app.state, "route_registry", None) if registry is not None: registry.stop() - # 2. 清理 workers（仅 inline 模式；process 模式下网关未加载插件） - if load_mode != "process": - await lifespan_setup.cleanup_workers(app) + # 2. 清理 inline workers（cleanup_workers 内部按 load_mode 过滤） + await lifespan_setup.cleanup_workers(app) # 3. 取消刷新任务 refresh_task = getattr(app.state, "refresh_task", None) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 5d5beb4..23ff15a 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -222,31 +222,36 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager: app.state.refresh_task = refresh_task return manager -def get_load_mode() -> str: +async def setup_route_registry( + app: FastAPI, + session: zenoh.Session, + manager: WorkerManager, + inline_types: list[str] | None = None, +): """ - 读取插件加载模式（config.toml [.workers].load_mode）。 + 启动动态路由注册表（订阅 chongming/routes/**）。 - - "inline"（默认）: 网关进程内加载插件（现状行为） - - "process": 独立 worker 进程经 Zenoh 注册路由，网关不再 import 插件 - """ - env_config = config.get(env, {}) - return env_config.get("workers", {}).get("load_mode", "inline") + 存在 load_mode="external" 的插件时启动：external 插件以独立进程运行 + （由 Docker/K8s 等编排拉起，网关不负责进程管理），其路由描述符经 Zenoh + 动态注册；inline 插件仍在网关进程内加载，两者同时存在（混合模式）。 + inline_types: 网关进程内加载的插件类型，注册表忽略其描述符以防重复挂载。 -async def setup_route_registry(app: FastAPI, session: zenoh.Session, manager: WorkerManager): - """ - process 模式：启动动态路由注册表（订阅 chongming/routes/**）。 - - 注意：process 模式下插件自定义 APIRouter（如 multipart 上传端点）无法由 - 网关发现（网关不再 import 插件），此类端点需在 P2 通过专门通道注册。 + 注意：external 插件的自定义 APIRouter（如 multipart 上传端点）无法由 + 网关发现（网关不 import 该插件），此类端点需保持 inline 或等 P2 专门通道。 """ from ..core.route_registry import RouteRegistry prefix = config["default"]["prefix"] registry = RouteRegistry(app, manager, prefix=prefix) + if inline_types: + registry.set_inline_types(set(inline_types)) await registry.start(session) app.state.route_registry = registry - logger.info(f"路由注册表已启动（load_mode=process, prefix={prefix}）") + logger.info( + f"路由注册表已启动（prefix={prefix}, 外部插件路由动态注册, " + f"inline 插件: {sorted(inline_types or []) or '无'}）" + ) return registry @@ -278,6 +283,10 @@ def _discover_plugins() -> list[dict]: continue plugin_type = workers_section.get("type", plugin_type) + # 插件运行模式（按插件粒度，混合共存）: + # "inline"（默认） = 网关进程内加载 + # "external" = 独立进程运行（Docker/K8s 编排），经 Zenoh 注册路由 + load_mode = workers_section.get("load_mode", "inline") # 插件声明的 MQTT Bridge 额外订阅主题（[[workers.mqtt_topics]]） # 网关级路由配置，不传给插件运行时；worker 目标即本插件类型。 mqtt_topics = [] @@ -294,7 +303,7 @@ def _discover_plugins() -> list[dict]: # — 其中 type 被 chongming_worker.runtime.create_workers 用于匹配 RPC 方法 # 3. [workers.config] 展平到顶层 (api_key, data_dir, heartbeat 等) workers_flat = {k: v for k, v in workers_section.items() - if k not in ("config", "mqtt_topics")} + if k not in ("config", "mqtt_topics", "load_mode")} runtime_config = { **plugin_conf, **workers_flat, @@ -307,6 +316,7 @@ def _discover_plugins() -> list[dict]: "factory": workers_section.get("factory", "create_workers"), "cleanup": workers_section.get("cleanup", "cleanup"), "count": workers_section.get("count", 1), + "load_mode": load_mode, "mqtt_topics": mqtt_topics, "config": runtime_config, }) @@ -334,29 +344,43 @@ def _discover_plugins() -> list[dict]: async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerManager): """ - 加载配置中的 worker 插件，启动对应的 worker 任务。 + 加载 inline 插件，启动对应的 worker 任务（混合模式）。 - 自动从 config/workers//config.toml 发现并加载插件。 - 每个插件的 [workers] 节定义加载参数，[workers.config] 节定义运行时配置。 + 自动从 config/workers//config.toml 发现插件；load_mode="external" + 的插件不在网关内加载（由独立进程运行，其路由经 RouteRegistry 动态注册）。 - 同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的 APIRouter 实例）， - 收集后供 register_custom_routes() 统一注册，实现网关与插件的解耦。 + 同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的 + APIRouter 实例），收集后供 register_custom_routes() 统一注册。 - 返回: (worker_tasks, custom_routers) + 返回: (worker_tasks, custom_routers, inline_types, external_types) """ workers_conf = _discover_plugins() - logger.info(f"发现 {len(workers_conf)} 个 worker 插件") + inline_plugins = [wc for wc in workers_conf if wc.get("load_mode", "inline") != "external"] + external_plugins = [wc for wc in workers_conf if wc.get("load_mode", "inline") == "external"] + inline_types = sorted({wc["type"] for wc in inline_plugins}) + external_types = sorted({wc["type"] for wc in external_plugins}) + logger.info( + f"发现 {len(workers_conf)} 个 worker 插件: " + f"inline={inline_types or '无'}, external={external_types or '无'}" + ) - # 汇总各插件声明的 MQTT Bridge 额外订阅主题，供 setup_mqtt 使用 + for wc in external_plugins: + logger.info( + "插件 %s 为 external 模式，跳过进程内加载（由独立进程经 Zenoh 注册路由）", + wc["type"], + ) + + # 汇总 inline 插件声明的 MQTT Bridge 额外订阅主题，供 setup_mqtt 使用 + # （external 插件的主题由 RouteRegistry 经 __meta__ 动态推送） app.state.plugin_mqtt_topics = [ topic - for wc in workers_conf + for wc in inline_plugins for topic in (wc.get("mqtt_topics") or []) ] worker_tasks = [] custom_routers = [] - for wc in workers_conf: + for wc in inline_plugins: module_name = wc["module"] factory_name = wc["factory"] count = wc["count"] @@ -397,12 +421,23 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana ) app.state.worker_tasks = worker_tasks - return worker_tasks, custom_routers + app.state.inline_types = inline_types + app.state.external_types = external_types + return worker_tasks, custom_routers, inline_types, external_types -async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers: list | None = None): - """注册动态路由和插件自定义路由。""" +async def register_routes( + app: FastAPI, + manager: WorkerManager, + custom_routers: list | None = None, + worker_types: list[str] | None = None, +): + """注册动态路由和插件自定义路由。 + + worker_types: 仅注册这些插件类型的 TOML 动态路由（混合模式下只注册 + inline 插件；external 插件路由由 RouteRegistry 动态挂载）。 + """ prefix = config["default"]["prefix"] - register_dynamic_routes(app, manager, prefix=prefix) + register_dynamic_routes(app, manager, prefix=prefix, worker_types=worker_types) # 注册插件自定义路由（如 FORM-DATA 文件上传等 TOML 无法描述的端点） # 统一加上全局 prefix（如 /api/v1），与 TOML 动态路由保持一致 if custom_routers: @@ -413,15 +448,19 @@ async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers: def validate_contracts(app: FastAPI | None = None): """ - 契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ gen_models 生成模型）。 + 契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ 生成模型），仅覆盖 inline 插件。 在 load_workers 之后、register_routes 之前调用：此时插件模块已导入 （注册表就绪）。HTTP 路由缺失对应 @rpc_method 时抛 RuntimeError 终止启动， 其余不一致以警告日志汇总。 + + external 插件在自己进程内做同样校验（validate_worker_contracts），网关侧跳过。 """ from ..core.contracts import validate_contracts as _validate - _validate(_discover_plugins()) + plugins = _discover_plugins() + inline_plugins = [p for p in plugins if p.get("load_mode", "inline") != "external"] + _validate(inline_plugins) def setup_mqtt_broker(app: FastAPI): """启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）""" @@ -534,10 +573,12 @@ async def shutdown_mqtt(app: FastAPI): async def cleanup_workers(app: FastAPI): - """调用每个 worker 配置中定义的 cleanup 函数。""" + """调用每个 inline worker 配置中定义的 cleanup 函数。""" workers_conf = _discover_plugins() for wc in workers_conf: + if wc.get("load_mode", "inline") == "external": + continue module_name = wc.get("module", "") cleanup_name = wc.get("cleanup", "") worker_type = wc.get("type", "") diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index 5c6fc0a..66179a1 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/src/chongming/app/core/mqtt_bridge.py @@ -45,12 +45,22 @@ class MQTTBridge: self._lt: Optional[asyncio.Task] = None self._running = False # custom_topics: [{"topic","worker","subject"?,"response_topic"?}, ...] - self._custom_topics = custom_topics or [] + # 混合模式双来源： + # _static_topics — inline 插件，构造时由网关从配置收集 + # _dynamic_topics — external 插件，运行时由 RouteRegistry 推送 + self._static_topics = custom_topics or [] + self._dynamic_topics: list = [] + self._compiled: list = [] + self._rebuild_compiled() + self._subscribed_custom: set = set() + + def _rebuild_compiled(self): + """合并静态/动态主题并重编译匹配正则。""" + merged = self._static_topics + self._dynamic_topics self._compiled = [ (ct, _compile_topic(ct.get("topic", ""))) - for ct in self._custom_topics if ct.get("topic") + for ct in merged if ct.get("topic") ] - self._subscribed_custom: set = set() async def start(self): if self._running: @@ -80,18 +90,15 @@ class MQTTBridge: + (f" 及 {len(self._compiled)} 个自定义主题" if self._compiled else "")) self._lt = asyncio.create_task(self._loop()) - async def update_custom_topics(self, custom_topics: list): + async def update_dynamic_topics(self, custom_topics: list): """ - 动态更新自定义订阅主题（process 模式：随 worker 路由表变化推送）。 + 动态更新 external 插件的自定义订阅主题（RouteRegistry 推送）。 - 重建编译表并增量重订阅（退订已移除主题、订阅新增主题）， + 与构造时的 inline 静态主题合并后增量重订阅（退订已移除主题、订阅新增主题）， 运行中调用即生效。 """ - self._custom_topics = list(custom_topics or []) - self._compiled = [ - (ct, _compile_topic(ct.get("topic", ""))) - for ct in self._custom_topics if ct.get("topic") - ] + self._dynamic_topics = list(custom_topics or []) + self._rebuild_compiled() new_topics = {ct["topic"] for ct, _ in self._compiled} if not (self._running and self._client): self._subscribed_custom = new_topics diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py index 41bcf13..db7239c 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -238,8 +238,18 @@ def register_route_group( return added -def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: str = ""): - """根据 config/workers/ 目录结构注册所有动态路由""" +def register_dynamic_routes( + app: FastAPI, + worker_manager: WorkerManager, + prefix: str = "", + worker_types: Optional[List[str]] = None, +): + """ + 根据 config/workers/ 目录结构注册所有动态路由。 + + worker_types: 白名单（混合模式下只注册 inline 插件； + None = 全部注册，兼容纯 inline 部署）。 + """ from pathlib import Path workers_base = Path(WORKER_CONFIG_DIR) @@ -254,6 +264,8 @@ def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: if not worker_dir.is_dir(): continue worker_type = worker_dir.name + if worker_types is not None and worker_type not in worker_types: + continue routes = load_worker_routes_from_dir(worker_type) if not routes: continue diff --git a/src/chongming/app/core/route_registry.py b/src/chongming/app/core/route_registry.py index 907d255..cef7087 100644 --- a/src/chongming/app/core/route_registry.py +++ b/src/chongming/app/core/route_registry.py @@ -69,10 +69,16 @@ class RouteRegistry: self._subscriber: Optional[zenoh.Subscriber] = None self._mqtt_bridge = None + # 网关进程内加载（inline）的插件类型：忽略其描述符，防止与盘扫注册重复挂载 + self._inline_types: set = set() # 将动态 schema 注入 OpenAPI（provider 实时读取本注册表） _inject_schemas_to_openapi(app, lambda: self._schemas) + def set_inline_types(self, types: set): + """设置 inline 插件类型集合（混合模式防双注册）。""" + self._inline_types = set(types) + async def start(self, session: zenoh.Session): """订阅 chongming/routes/**，开始接收路由描述符。""" self._loop = asyncio.get_running_loop() @@ -90,12 +96,12 @@ class RouteRegistry: task.cancel() def attach_bridge(self, mqtt_bridge): - """挂接 MQTT Bridge：后续路由表变化会推送自定义订阅主题。""" + """挂接 MQTT Bridge：后续路由表变化会推送 external 插件的动态订阅主题。""" self._mqtt_bridge = mqtt_bridge # Bridge 晚于路由到达时，立即补推当前已收集的主题 topics = self._collect_mqtt_topics() if topics: - self._loop.create_task(mqtt_bridge.update_custom_topics(topics)) + self._loop.create_task(mqtt_bridge.update_dynamic_topics(topics)) @property def route_types(self) -> List[str]: @@ -133,6 +139,10 @@ class RouteRegistry: return wtype, route_id = parts[2], parts[3] + if wtype in self._inline_types: + # 该插件由网关进程内加载（盘扫注册路由），忽略外部描述符防双注册 + return + if kind == zenoh.SampleKind.DELETE: self._remove(wtype, route_id) return @@ -235,9 +245,9 @@ class RouteRegistry: # 路由表变化后 OpenAPI 缓存失效，下一次访问时重建 self._app.openapi_schema = None - # 3. 同步 MQTT 自定义订阅主题 + # 3. 同步 external 插件的 MQTT 动态订阅主题 if self._mqtt_bridge is not None: - await self._mqtt_bridge.update_custom_topics(self._collect_mqtt_topics()) + await self._mqtt_bridge.update_dynamic_topics(self._collect_mqtt_topics()) def _collect_mqtt_topics(self) -> List[Dict[str, Any]]: """汇总所有插件 __meta__ 声明的自定义订阅主题。""" diff --git a/test_process_mode_e2e.py b/test_process_mode_e2e.py index 340cee8..300c857 100644 --- a/test_process_mode_e2e.py +++ b/test_process_mode_e2e.py @@ -1,9 +1,9 @@ """ -process 模式端到端测试 +混合模式（external 插件）端到端测试 前置条件: - 1. 网关已以 load_mode = "process" 启动（config.toml [development.workers]） - uv run serve + 1. 网关正常运行（uv run serve）；example 插件配置为 external： + config/workers/example/config.toml [workers] 节 load_mode = "external" 2. 本脚本自己拉起/停止独立 worker 进程（uv run worker_serve --type example） 验证: diff --git a/test_route_registry.py b/test_route_registry.py index e2d1535..0ab4c1c 100644 --- a/test_route_registry.py +++ b/test_route_registry.py @@ -1,10 +1,11 @@ """ RouteRegistry 单元测试（loopback Zenoh 会话 + httpx ASGI 功能断言） -验证 process 模式核心链路： +验证 external 插件路由动态注册核心链路（含混合模式防双注册）： RoutePublisher.put 描述符 → 网关 RouteRegistry 动态挂载 FastAPI 路由（命中返回 503，worker 桩无实例） RoutePublisher.delete → 动态摘除路由（404） - __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_custom_topics） + __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge（update_dynamic_topics） + set_inline_types → 忽略 inline 插件的描述符（防双注册） 注：FastAPI 0.139 的 include_router 将路由包装为 _IncludedRouter（惰性挂载）， 不能靠遍历 app.router.routes 的 path 判断，故用 ASGI 请求状态码做功能断言。 @@ -37,12 +38,12 @@ class FakeWorkerManager: class FakeBridge: - """记录 update_custom_topics 的调用结果。""" + """记录 update_dynamic_topics 的调用结果。""" def __init__(self): self.topics = None - async def update_custom_topics(self, topics): + async def update_dynamic_topics(self, topics): self.topics = list(topics) @@ -145,6 +146,26 @@ async def main(): check("delete 后 HTTP 路由动态摘除（404）", ok) ok = await wait_until(lambda: bridge.topics == []) check("元信息摘除后 Bridge 主题清空", ok, f"bridge.topics={bridge.topics}") + + # 5. 混合模式：inline 类型的描述符被忽略（防与盘扫注册重复挂载） + registry.set_inline_types({"example"}) + publisher.publish_routes([HTTP_DESCRIPTOR]) + await asyncio.sleep(3.5) # 覆盖防抖窗口 + resp = await client.get("/api/v1/calc/add?a=1&b=2") + check("inline 类型描述符被忽略（防双注册，404）", resp.status_code == 404, + f"status={resp.status_code}") + + # 清除 inline 集合后恢复动态挂载 + registry.set_inline_types(set()) + publisher.publish_routes([HTTP_DESCRIPTOR]) + + async def remounted(): + resp = await client.get("/api/v1/calc/add?a=1&b=2") + return resp.status_code == 503 + + ok = await wait_until(remounted) + check("清除 inline 集合后恢复动态挂载", ok) + publisher.unpublish() finally: registry.stop() gw_session.close()