diff --git a/README.md b/README.md index 193d755..ed47735 100644 --- a/README.md +++ b/README.md @@ -340,6 +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 pip install -e utils/ # 安装/更新工具包 ``` @@ -374,6 +375,10 @@ uv pip install -e utils/ # 安装/更新工具包 ## 配置说明 - **应用配置**:根目录 `config.toml`,包含环境、服务器、安全、MQTT 等全局设置 +- **插件加载模式**:`config.toml` 的 `[.workers].load_mode` + - `"inline"`(默认):插件在网关进程内加载 + - `"process"`:插件以独立进程运行(`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 fef0ed3..b608d78 100644 --- a/config.toml +++ b/config.toml @@ -24,6 +24,11 @@ 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/pyproject.toml b/pyproject.toml index 863fff5..56d347c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,3 +68,4 @@ gen_models = "chongming.scripts.gen_models:gen_models" gen_plugin = "chongming.scripts.gen_plugin:gen_plugin" gunicorn = "chongming:gunicorn_serve" serve = "chongming:serve" +worker_serve = "chongming:worker_serve" diff --git a/src/chongming/__init__.py b/src/chongming/__init__.py index 191de7f..b81e351 100644 --- a/src/chongming/__init__.py +++ b/src/chongming/__init__.py @@ -22,6 +22,114 @@ def serve(): logger.info(f"启动日志: {log_config}") uvicorn.run("chongming.app:app", **server_config) + +def worker_serve(): + """ + 独立 Worker 进程入口(process 模式) + + 流程: 读取全局配置 → 发现本插件的 config/workers//config.toml → + import 插件(触发 @rpc_method 注册)→ 契约校验(硬失败则退出)→ + 创建 Worker 副本(count 进程内保留)→ 经 Zenoh 发布路由描述符 → + 等待 SIGTERM/SIGINT → 摘除路由 + cleanup + 关闭会话。 + + 用法: uv run worker_serve --type example [--count 3] [--config config.toml] + 需在项目根目录运行(配置与插件均按相对路径解析)。 + """ + import argparse + import asyncio + import inspect + import os + import signal + import sys + + import zenoh + + from .app.core.contracts import validate_worker_contracts + from .app.core.lifespan_setup import _discover_plugins + from .app.core.route_publisher import RoutePublisher + + parser = argparse.ArgumentParser(description="chongming 独立 Worker 进程") + parser.add_argument("--type", required=True, help="插件类型(如 example / auth)") + parser.add_argument("--count", type=int, default=None, help="副本数(覆盖配置中的 count)") + parser.add_argument("--config", default="config.toml", help="全局配置文件路径") + args = parser.parse_args() + + config = load_config() + logger = get_logger("app") + + plugins = _discover_plugins() + wc = next((p for p in plugins if p.get("type") == args.type), None) + if wc is None: + logger.error("未找到插件配置: %s(请检查 config/workers/%s/config.toml)", args.type, args.type) + sys.exit(1) + + async def _run(): + # 1. Zenoh 会话 + zconfig = zenoh.Config() + zconfig.from_file(config.get("zenoh", {}).get("json5_path", "config/zenoh_config.json5")) + session = zenoh.open(zconfig) + logger.info("[%s] Zenoh 会话已建立", args.type) + + try: + # 2. import 插件(触发 @rpc_method 注册),然后契约校验 + module = __import__(wc["module"], fromlist=[wc["factory"]]) + routes = validate_worker_contracts(args.type) + + # 3. 创建 Worker 副本(worker_manager=None:跨进程经 liveliness 发现) + factory = getattr(module, wc["factory"]) + count = args.count or wc.get("count", 1) + if inspect.iscoroutinefunction(factory): + tasks = await factory(session, wc["config"], count, worker_manager=None) + else: + tasks = factory(session, wc["config"], count, worker_manager=None) + + # 4. 发布路由描述符(网关 RouteRegistry 订阅后动态挂载 HTTP 路由) + publisher = RoutePublisher(session, args.type) + mqtt_topics = wc.get("mqtt_topics") or [] + publisher.publish_routes(routes, mqtt_topics) + + # 5. 等待退出信号;期间周期性重发布(网关后启动/样本丢失时自愈) + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stop_event.set) + + async def _announce(): + while not stop_event.is_set(): + try: + await asyncio.wait_for(stop_event.wait(), timeout=10.0) + return + except asyncio.TimeoutError: + publisher.publish_routes(routes, mqtt_topics) + + announce_task = asyncio.create_task(_announce()) + logger.info("[%s] Worker 进程就绪,等待退出信号...", args.type) + await stop_event.wait() + announce_task.cancel() + await asyncio.gather(announce_task, return_exceptions=True) + + # 6. 优雅退出:摘除路由 → cleanup → 取消副本任务 → 关闭会话 + publisher.unpublish() + cleanup = getattr(module, wc.get("cleanup", ""), None) + if cleanup is not None: + worker_id = f"worker/{args.type}/worker-i-{os.getpid()}" + if inspect.iscoroutinefunction(cleanup): + await cleanup(worker_id) + else: + cleanup(worker_id) + for t in tasks: + t.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + logger.info("[%s] Worker 进程已退出", args.type) + finally: + session.close() + + try: + asyncio.run(_run()) + except KeyboardInterrupt: + pass + def gunicorn_serve(): """ 使用 gunicorn 启动服务 diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 3e036d2..3dce1c3 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -47,14 +47,23 @@ async def lifespan(app: FastAPI): # 4. WorkerManager manager = lifespan_setup.setup_worker_manager(app, session) - # 5. 加载 workers(同时自动发现插件自定义路由) - worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager) + # 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.5 契约校验(TOML 路由 ↔ @rpc_method ↔ 生成模型),失败则启动终止 - lifespan_setup.validate_contracts(app) + # 5.5 契约校验(TOML 路由 ↔ @rpc_method ↔ 生成模型),失败则启动终止 + lifespan_setup.validate_contracts(app) - # 6. 注册路由(含 TOML 动态路由 + 插件自定义路由) - await lifespan_setup.register_routes(app, manager, custom_routers) + # 6. 注册路由(含 TOML 动态路由 + 插件自定义路由) + await lifespan_setup.register_routes(app, manager, custom_routers) # 7. 启动 MQTT Bridge(内嵌 Broker + Worker 桥接) await lifespan_setup.setup_mqtt(app, manager) @@ -78,8 +87,14 @@ async def lifespan(app: FastAPI): # 1. 停止 MQTT Bridge await lifespan_setup.shutdown_mqtt(app) - # 2. 清理 workers(调用 cleanup) - await lifespan_setup.cleanup_workers(app) + # 1.5 process 模式:停止路由注册表订阅 + 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) # 3. 取消刷新任务 refresh_task = getattr(app.state, "refresh_task", None) diff --git a/src/chongming/app/core/contracts.py b/src/chongming/app/core/contracts.py index 98e0797..16a1e5e 100644 --- a/src/chongming/app/core/contracts.py +++ b/src/chongming/app/core/contracts.py @@ -67,9 +67,93 @@ def _load_plugin_models(worker_type: str) -> Optional[Set[str]]: return None +def _check_plugin( + worker_type: str, + toml_routes: List[Dict[str, Any]], + registered: Dict[str, Any], + model_names: Optional[Set[str]], +) -> tuple[List[str], List[str]]: + """ + 对单个插件执行 TOML ↔ 注册表 ↔ 生成模型一致性校验。 + + Args: + worker_type: 插件类型 + toml_routes: load_worker_routes_from_dir() 输出的路由列表 + registered: get_all_rpc_methods().get(worker_type, {}) 注册表 + model_names: 插件 models.pydantic 的模型名集合(None 表示无生成模型) + + Returns: + (hard_errors, warnings): 硬错误与软警告消息列表 + """ + hard_errors: List[str] = [] + warnings: List[str] = [] + toml_keys = {f"{r['subject']}#{r['method']}" for r in toml_routes} + + for r in toml_routes: + key = f"{r['subject']}#{r['method']}" + route_desc = f"{r['method']} {r['path']} -> subject '{r['subject']}'" + + if key in registered: + continue + + if r["method"] in HTTP_METHODS and not r.get("internal", False): + # 硬失败:运行时必然 503 + hint = "" + if not registered: + hint = "(该插件注册表为空,请检查启动日志中的 '加载 Worker 插件失败' 错误)" + hard_errors.append( + f"[{worker_type}] {route_desc} 缺少对应的 @rpc_method 注册{hint}" + ) + else: + # 软警告:MQTT / internal 端点缺失注册 + warnings.append( + f"[{worker_type}] {route_desc} 无对应的 @rpc_method(method='{r['method']}') 注册" + ) + + # 已注册方法无 TOML 路由(反向漂移检测) + for key in registered: + if key not in toml_keys: + warnings.append( + f"[{worker_type}] 已注册 '{key}' 但无对应 TOML 路由定义" + ) + + # 模型一致性(仅当插件存在生成模型时校验) + if model_names is not None: + referenced: Set[str] = set() + for r in toml_routes: + route_name = r.get("route_name", "") + if not route_name: + continue + method = r["method"] + if r.get("params"): + expected = _model_name(route_name, method, "Request") + referenced.add(expected) + if expected not in model_names: + warnings.append( + f"[{worker_type}] '{r['subject']}' [{method}] 声明了 params," + f"但模型 {expected} 不存在(可能未运行 gen_models)" + ) + if r.get("response_model"): + expected = _model_name(route_name, method, "Response") + referenced.add(expected) + if expected not in model_names: + warnings.append( + f"[{worker_type}] '{r['subject']}' [{method}] 声明了 response_model," + f"但模型 {expected} 不存在(可能未运行 gen_models)" + ) + + for name in sorted(model_names - referenced): + warnings.append( + f"[{worker_type}] 模型 {name} 未被任何 TOML 路由引用" + f"(孤儿模型,可重新运行 gen_models 清理)" + ) + + return hard_errors, warnings + + def validate_contracts(plugins: List[Dict[str, Any]]) -> None: """ - 对每个发现的插件执行三方一致性校验。 + 对每个发现的插件执行三方一致性校验(网关 inline 模式,启动时)。 Args: plugins: _discover_plugins() 返回的插件配置列表 @@ -88,67 +172,10 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None: toml_routes = load_worker_routes_from_dir(worker_type) registered = registry.get(worker_type, {}) - toml_keys = {f"{r['subject']}#{r['method']}" for r in toml_routes} model_names = _load_plugin_models(worker_type) - - for r in toml_routes: - key = f"{r['subject']}#{r['method']}" - route_desc = f"{r['method']} {r['path']} -> subject '{r['subject']}'" - - if key in registered: - continue - - if r["method"] in HTTP_METHODS and not r.get("internal", False): - # 硬失败:运行时必然 503 - hint = "" - if not registered: - hint = "(该插件注册表为空,请检查启动日志中的 '加载 Worker 插件失败' 错误)" - hard_errors.append( - f"[{worker_type}] {route_desc} 缺少对应的 @rpc_method 注册{hint}" - ) - else: - # 软警告:MQTT / internal 端点缺失注册 - warnings.append( - f"[{worker_type}] {route_desc} 无对应的 @rpc_method(method='{r['method']}') 注册" - ) - - # 已注册方法无 TOML 路由(反向漂移检测) - for key in registered: - if key not in toml_keys: - warnings.append( - f"[{worker_type}] 已注册 '{key}' 但无对应 TOML 路由定义" - ) - - # 模型一致性(仅当插件存在生成模型时校验) - if model_names is not None: - referenced: Set[str] = set() - for r in toml_routes: - route_name = r.get("route_name", "") - if not route_name: - continue - method = r["method"] - if r.get("params"): - expected = _model_name(route_name, method, "Request") - referenced.add(expected) - if expected not in model_names: - warnings.append( - f"[{worker_type}] '{r['subject']}' [{method}] 声明了 params," - f"但模型 {expected} 不存在(可能未运行 gen_models)" - ) - if r.get("response_model"): - expected = _model_name(route_name, method, "Response") - referenced.add(expected) - if expected not in model_names: - warnings.append( - f"[{worker_type}] '{r['subject']}' [{method}] 声明了 response_model," - f"但模型 {expected} 不存在(可能未运行 gen_models)" - ) - - for name in sorted(model_names - referenced): - warnings.append( - f"[{worker_type}] 模型 {name} 未被任何 TOML 路由引用" - f"(孤儿模型,可重新运行 gen_models 清理)" - ) + hard, warns = _check_plugin(worker_type, toml_routes, registered, model_names) + hard_errors.extend(hard) + warnings.extend(warns) # 输出结果 if warnings: @@ -166,3 +193,45 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None: ) logger.info("契约校验通过(TOML 路由 ↔ @rpc_method 注册表一致)") + + +def validate_worker_contracts(worker_type: str) -> List[Dict[str, Any]]: + """ + Worker 进程侧契约校验(process 模式):校验单个插件并返回可发布的路由描述符。 + + 调用时机:worker 进程 import 插件模块后(@rpc_method 注册表就绪)、 + 发布路由描述符之前。校验逻辑与网关 validate_contracts 共用 _check_plugin。 + + Args: + worker_type: 插件类型(与 config/workers/ 目录一致) + + Returns: + 通过校验的 TOML 路由描述符列表(直接用于 RoutePublisher 发布) + + Raises: + RuntimeError: 存在硬失败项(HTTP 路由缺失 @rpc_method 注册)时, + 调用方应终止 worker 进程,避免发布残缺路由。 + """ + registered = get_all_rpc_methods().get(worker_type, {}) + toml_routes = load_worker_routes_from_dir(worker_type) + model_names = _load_plugin_models(worker_type) + + hard_errors, warnings = _check_plugin( + worker_type, toml_routes, registered, model_names + ) + + for w in warnings: + logger.warning(" ⚠ %s", w) + + if hard_errors: + logger.error("Worker 契约校验失败,发现 %d 个硬性错误:", len(hard_errors)) + for e in hard_errors: + logger.error(" ✗ %s", e) + raise RuntimeError( + f"[{worker_type}] 契约校验失败: " + "; ".join(hard_errors) + ) + + logger.info( + "[%s] 契约校验通过,可发布 %d 个路由描述符", worker_type, len(toml_routes) + ) + return toml_routes diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 60f6730..5d5beb4 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -222,6 +222,34 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager: app.state.refresh_task = refresh_task return manager +def get_load_mode() -> str: + """ + 读取插件加载模式(config.toml [.workers].load_mode)。 + + - "inline"(默认): 网关进程内加载插件(现状行为) + - "process": 独立 worker 进程经 Zenoh 注册路由,网关不再 import 插件 + """ + env_config = config.get(env, {}) + return env_config.get("workers", {}).get("load_mode", "inline") + + +async def setup_route_registry(app: FastAPI, session: zenoh.Session, manager: WorkerManager): + """ + process 模式:启动动态路由注册表(订阅 chongming/routes/**)。 + + 注意:process 模式下插件自定义 APIRouter(如 multipart 上传端点)无法由 + 网关发现(网关不再 import 插件),此类端点需在 P2 通过专门通道注册。 + """ + from ..core.route_registry import RouteRegistry + + prefix = config["default"]["prefix"] + registry = RouteRegistry(app, manager, prefix=prefix) + await registry.start(session) + app.state.route_registry = registry + logger.info(f"路由注册表已启动(load_mode=process, prefix={prefix})") + return registry + + def _discover_plugins() -> list[dict]: """ 发现所有插件配置。 @@ -487,6 +515,10 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge ) app.state.mqtt_bridge = bridge await bridge.start() + # process 模式:把 Bridge 挂接给路由注册表(动态主题 + 补推已到达的元信息) + registry = getattr(app.state, "route_registry", None) + if registry is not None: + registry.attach_bridge(bridge) logger.info(f"MQTT Bridge 已启动 ({host}:{port})") return bridge diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index c0e2187..5c6fc0a 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/src/chongming/app/core/mqtt_bridge.py @@ -50,6 +50,7 @@ class MQTTBridge: (ct, _compile_topic(ct.get("topic", ""))) for ct in self._custom_topics if ct.get("topic") ] + self._subscribed_custom: set = set() async def start(self): if self._running: @@ -74,10 +75,40 @@ class MQTTBridge: topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)] topics += [(ct["topic"], QOS_1) for ct, _ in self._compiled] await self._client.subscribe(topics) + self._subscribed_custom = {ct["topic"] for ct, _ in self._compiled} logger.info(f"已订阅 {MQTT_SUBSCRIBE_TOPIC}" + (f" 及 {len(self._compiled)} 个自定义主题" if self._compiled else "")) self._lt = asyncio.create_task(self._loop()) + async def update_custom_topics(self, custom_topics: list): + """ + 动态更新自定义订阅主题(process 模式:随 worker 路由表变化推送)。 + + 重建编译表并增量重订阅(退订已移除主题、订阅新增主题), + 运行中调用即生效。 + """ + 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") + ] + new_topics = {ct["topic"] for ct, _ in self._compiled} + if not (self._running and self._client): + self._subscribed_custom = new_topics + return + removed = self._subscribed_custom - new_topics + added = new_topics - self._subscribed_custom + if removed: + await self._client.unsubscribe(list(removed)) + if added: + await self._client.subscribe([(t, QOS_1) for t in added]) + self._subscribed_custom = new_topics + if added or removed: + logger.info( + f"自定义订阅主题更新: +{len(added)} -{len(removed)}" + f"(当前 {len(new_topics)} 个)" + ) + async def _sc_stop(self): # 先断开连接,让 deliver_message() 解锁 if self._client: diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py index 31a89fc..41bcf13 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -198,6 +198,46 @@ def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]: return routes +def register_route_group( + app: FastAPI, + worker_manager: WorkerManager, + worker_type: str, + routes: List[Dict[str, Any]], + prefix: str = "", + extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, +) -> List[Any]: + """ + 为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。 + + 与 register_dynamic_routes(盘扫路径)共用同一注册逻辑, + process 模式下 RouteRegistry 也调用本函数做动态挂载。 + + Args: + app: FastAPI 应用 + worker_manager: WorkerManager 实例 + worker_type: worker 类型(OpenAPI 分组标签) + routes: 路由描述符列表(TOML 兼容格式,与 load_worker_routes_from_dir 输出一致) + prefix: 全局路由前缀(如 /api/v1) + extra_schemas: 额外 schema 收集 dict(GET 请求模型注入 OpenAPI 用) + + Returns: + List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表 + (调用方可记录引用,用于动态移除) + """ + router = APIRouter(prefix=prefix) + for route_conf in routes: + _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas) + + before = set(id(r) for r in app.router.routes) + app.include_router(router) + added = [r for r in app.router.routes if id(r) not in before] + logger.info( + f"已挂载 worker '{worker_type}' 的路由组: {len(routes)} 个描述符 -> " + f"{len(added)} 条 HTTP 路由 (prefix={prefix or '/'})" + ) + return added + + def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: str = ""): """根据 config/workers/ 目录结构注册所有动态路由""" from pathlib import Path @@ -207,7 +247,6 @@ def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: logger.info(f"Worker 配置目录不存在: {WORKER_CONFIG_DIR}") return - router = APIRouter(prefix=prefix) extra_schemas: Dict[str, type[BaseModel]] = {} # 扫描所有 worker 子目录(每个子目录名 = worker_type) @@ -219,24 +258,26 @@ def register_dynamic_routes(app: FastAPI, worker_manager: WorkerManager, prefix: if not routes: continue - for route_conf in routes: - _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas) - - app.include_router(router) + register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas) # 将额外 schema 注册到 OpenAPI 的 components/schemas if extra_schemas: - _inject_schemas_to_openapi(app, extra_schemas) + _inject_schemas_to_openapi(app, lambda: extra_schemas) logger.info(f"动态路由注册完成,worker 配置目录: {WORKER_CONFIG_DIR}") -def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]]): +def _inject_schemas_to_openapi(app: FastAPI, schemas_provider): """ 将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。 通过重写 app.openapi() 方法注入,因为 FastAPI 在生命周期内只生成一次 OpenAPI schema,之后会被缓存。 + + Args: + schemas_provider: 无参 callable,返回 {name: BaseModel} 字典。 + 盘扫路径传静态 dict;process 模式传 RouteRegistry 的动态 dict, + 使路由表变化后重新生成的 OpenAPI 能反映最新 schema。 """ original_openapi = app.openapi @@ -248,7 +289,7 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas: Dict[str, type[BaseModel]] return {"openapi": "3.1.0", "info": {"title": "Chongming", "version": "1.0.0"}, "paths": {}} # 注入我们的 schema schemas_dict = schema.setdefault("components", {}).setdefault("schemas", {}) - for name, model in schemas.items(): + for name, model in schemas_provider().items(): if name not in schemas_dict: schemas_dict[name] = model.model_json_schema() app.openapi_schema = schema diff --git a/src/chongming/app/core/route_publisher.py b/src/chongming/app/core/route_publisher.py new file mode 100644 index 0000000..18f5612 --- /dev/null +++ b/src/chongming/app/core/route_publisher.py @@ -0,0 +1,90 @@ +""" +路由描述符发布器(worker 进程侧) + +独立 worker 进程在完成契约校验后,将路由描述符 put 到 Zenoh 的 +`chongming/routes//...` 命名空间;进程优雅退出时 delete 摘除。 + +描述符 schema_version=1,字段与网关 RouteConfig 完全对位(TOML 兼容 dict), +网关 RouteRegistry 订阅同名空间并动态挂载 HTTP 路由。 +""" +import json +from typing import Any, Dict, List, Optional + +import zenoh + +from .logger import get_logger + +logger = get_logger("route_publisher") + +ROUTES_PREFIX = "chongming/routes" +META_KEY = "__meta__" +SCHEMA_VERSION = 1 + + +def route_id(subject: str, method: str) -> str: + """route_id = subject 与 method 的编码(避免 keyexpr 特殊字符歧义)""" + return f"{subject.replace('.', '_')}__{method.upper()}" + + +class RoutePublisher: + """发布/摘除某 worker_type 的路由描述符。""" + + def __init__(self, session: zenoh.Session, worker_type: str): + self._session = session + self._worker_type = worker_type + self._keys: List[str] = [] + + def publish_routes( + self, + routes: List[Dict[str, Any]], + mqtt_topics: Optional[List[Dict[str, Any]]] = None, + ): + """ + 发布路由描述符。 + + Args: + routes: 契约校验通过的路由描述符列表(TOML 兼容格式) + mqtt_topics: 插件声明的 [[workers.mqtt_topics]](可空) + """ + for r in routes: + subject = r.get("subject", "") + method = r.get("method", "GET") + if not subject: + logger.warning("跳过无 subject 的路由描述符: %s", r) + continue + descriptor = dict(r) + descriptor["schema_version"] = SCHEMA_VERSION + descriptor["type"] = self._worker_type + key = f"{ROUTES_PREFIX}/{self._worker_type}/{route_id(subject, method)}" + self._session.put(key, json.dumps(descriptor, ensure_ascii=False).encode("utf-8")) + self._keys.append(key) + logger.info( + "已发布路由: %s %s (subject=%s)", + method.upper(), descriptor.get("path", "?"), subject, + ) + + if mqtt_topics: + meta_key = f"{ROUTES_PREFIX}/{self._worker_type}/{META_KEY}" + meta = {"schema_version": SCHEMA_VERSION, "mqtt_topics": mqtt_topics} + self._session.put( + meta_key, json.dumps(meta, ensure_ascii=False).encode("utf-8") + ) + self._keys.append(meta_key) + logger.info("已发布插件元信息: %d 个自定义 MQTT 主题", len(mqtt_topics)) + + # 去重(周期性重发布会重复调用本方法) + self._keys = list(dict.fromkeys(self._keys)) + logger.info( + "[%s] 路由发布完成: %d 端点 + %s", self._worker_type, + len(self._keys), "元信息" if mqtt_topics else "无元信息", + ) + + def unpublish(self): + """摘除全部已发布的路由描述符(优雅退出时调用)。""" + for key in self._keys: + try: + self._session.delete(key) + except Exception as e: # pragma: no cover + logger.warning("摘除路由失败 [%s]: %s", key, e) + logger.info("[%s] 已摘除 %d 个路由描述符", self._worker_type, len(self._keys)) + self._keys.clear() diff --git a/src/chongming/app/core/route_registry.py b/src/chongming/app/core/route_registry.py new file mode 100644 index 0000000..907d255 --- /dev/null +++ b/src/chongming/app/core/route_registry.py @@ -0,0 +1,252 @@ +""" +动态路由注册表(process 模式) + +网关订阅 `chongming/routes/**` 命名空间,独立 worker 进程上线时 put 路由描述符、 +下线时 delete,网关据此动态挂载/摘除 FastAPI 路由并同步 MQTT 自定义订阅主题。 + +协议(描述符 schema_version=1): + chongming/routes//__meta__ 插件级描述: {"schema_version":1,"mqtt_topics":[...]} + chongming/routes// 端点描述(TOML 兼容 dict,与 RouteConfig 对位): + {"schema_version":1, "type","subject","method","path","params", + "response_model","summary","timeout","cacheable","auth_required","internal"} + +特性: + - SampleKind.PUT -> upsert;DELETE -> remove + - 同一 route_id 内容相同(hash)视为多副本重复发布,忽略;不同 -> 告警并保留先到者 + - type 级 2 秒防抖:worker 重启风暴合并为一次路由重建 + - 迟到一致性:worker 侧周期性重发布描述符(RoutePublisher 由调用方周期调用), + 网关后启动或样本丢失时,路由在一个发布周期内自动补齐 +""" +import asyncio +import hashlib +import json +from typing import Any, Dict, List, Optional + +import zenoh + +from .logger import get_logger +from .route_generator import register_route_group, _inject_schemas_to_openapi + +logger = get_logger("route_registry") + +ROUTES_PREFIX = "chongming/routes" +META_KEY = "__meta__" +SCHEMA_VERSION = 1 +# worker 重启风暴防抖窗口(秒) +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() + + +class RouteRegistry: + """process 模式下的网关侧动态路由注册表。""" + + def __init__( + self, + app, + worker_manager, + prefix: str = "", + ): + self._app = app + self._wm = worker_manager + self._prefix = prefix + self._loop: Optional[asyncio.AbstractEventLoop] = None + + # worker_type -> route_id -> descriptor + self._groups: Dict[str, Dict[str, Dict[str, Any]]] = {} + # worker_type -> 插件级描述(mqtt_topics 等) + self._meta: Dict[str, Dict[str, Any]] = {} + # worker_type -> 已挂载的 APIRoute 对象(用于动态摘除) + self._route_objs: Dict[str, List[Any]] = {} + # OpenAPI 额外 schema(GET 请求模型),随路由表变化 + self._schemas: Dict[str, Any] = {} + # worker_type -> 防抖任务 + self._debounce: Dict[str, asyncio.Task] = {} + + self._subscriber: Optional[zenoh.Subscriber] = None + self._mqtt_bridge = None + + # 将动态 schema 注入 OpenAPI(provider 实时读取本注册表) + _inject_schemas_to_openapi(app, lambda: self._schemas) + + async def start(self, session: zenoh.Session): + """订阅 chongming/routes/**,开始接收路由描述符。""" + self._loop = asyncio.get_running_loop() + self._subscriber = session.declare_subscriber( + f"{ROUTES_PREFIX}/**", self._on_sample + ) + logger.info(f"路由注册表已就绪: 订阅 {ROUTES_PREFIX}/**") + + def stop(self): + """停止订阅(shutdown 时调用)。""" + if self._subscriber is not None: + self._subscriber.undeclare() + self._subscriber = None + for task in self._debounce.values(): + task.cancel() + + def attach_bridge(self, mqtt_bridge): + """挂接 MQTT Bridge:后续路由表变化会推送自定义订阅主题。""" + self._mqtt_bridge = mqtt_bridge + # Bridge 晚于路由到达时,立即补推当前已收集的主题 + topics = self._collect_mqtt_topics() + if topics: + self._loop.create_task(mqtt_bridge.update_custom_topics(topics)) + + @property + def route_types(self) -> List[str]: + """当前已注册路由的 worker 类型列表(观测用)。""" + return sorted(self._groups.keys()) + + def ready_summary(self) -> str: + """路由就绪状态摘要(日志/观测用)。""" + parts = [] + for wtype in sorted(self._groups): + http = sum( + 1 for d in self._groups[wtype].values() + if d.get("method") not in ("MQTT",) and not d.get("internal") + ) + parts.append(f"{wtype}={len(self._groups[wtype])} 端点/{http} HTTP") + return ", ".join(parts) if parts else "(空)" + + # ---- zenoh 回调线程 -> 事件循环 ---- + + def _on_sample(self, sample: zenoh.Sample): + """zenoh 后台线程回调:转交事件循环处理。""" + if self._loop is None: + return + key = str(sample.key_expr) + kind = sample.kind + payload = bytes(sample.payload) if sample.payload else b"" + asyncio.run_coroutine_threadsafe( + self._handle_sample(key, kind, payload), self._loop + ) + + async def _handle_sample(self, key: str, kind: zenoh.SampleKind, payload: bytes): + """解析样本并更新内存路由表(在事件循环内执行)。""" + parts = key.split("/") + if len(parts) < 4: + return + wtype, route_id = parts[2], parts[3] + + if kind == zenoh.SampleKind.DELETE: + self._remove(wtype, route_id) + return + + try: + descriptor = json.loads(payload.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError): + logger.warning("丢弃非法路由描述符 [%s]: JSON 解析失败", key) + return + if not isinstance(descriptor, dict): + logger.warning("丢弃非法路由描述符 [%s]: 非 dict", key) + return + + if route_id == META_KEY: + self._upsert_meta(wtype, descriptor) + else: + self._upsert_route(wtype, route_id, descriptor) + + def _upsert_meta(self, wtype: str, descriptor: Dict[str, Any]): + old = self._meta.get(wtype) + new_hash = _descriptor_hash(descriptor) + if old is not None and _descriptor_hash(old) == new_hash: + return + self._meta[wtype] = descriptor + self._schedule_rebuild(wtype) + + def _upsert_route(self, wtype: str, route_id: str, descriptor: Dict[str, Any]): + group = self._groups.setdefault(wtype, {}) + existing = group.get(route_id) + if existing is not None: + if _descriptor_hash(existing) == _descriptor_hash(descriptor): + # 多副本重复发布同一描述符,幂等忽略 + return + logger.warning( + "路由描述符冲突 [%s/%s]: 已存在版本保留,忽略新版本", + wtype, route_id, + ) + return + # 补全描述符字段(发布端缺省时的兜底) + descriptor.setdefault("type", wtype) + descriptor.setdefault("schema_version", SCHEMA_VERSION) + group[route_id] = descriptor + logger.info( + "收到路由注册: %s %s %s (subject=%s)", + descriptor.get("method", "?"), + descriptor.get("path", "?"), + f"[{wtype}]", + descriptor.get("subject", "?"), + ) + self._schedule_rebuild(wtype) + + def _remove(self, wtype: str, route_id: str): + if route_id == META_KEY: + if self._meta.pop(wtype, None) is not None: + self._schedule_rebuild(wtype) + return + group = self._groups.get(wtype) + if group and group.pop(route_id, None) is not None: + logger.info("收到路由摘除: [%s] %s", wtype, route_id) + self._schedule_rebuild(wtype) + + def _schedule_rebuild(self, wtype: str): + """type 级防抖:合并短时间内的多次变更。""" + prev = self._debounce.get(wtype) + if prev is not None and not prev.done(): + prev.cancel() + self._debounce[wtype] = asyncio.create_task(self._debounced_rebuild(wtype)) + + async def _debounced_rebuild(self, wtype: str): + try: + await asyncio.sleep(DEBOUNCE_SECONDS) + except asyncio.CancelledError: + return + self._debounce.pop(wtype, None) + await self._rebuild(wtype) + + async def _rebuild(self, wtype: str): + """整组重建该 worker_type 的 HTTP 路由 + 同步 MQTT 自定义主题。""" + routes = list(self._groups.get(wtype, {}).values()) + routes.sort(key=lambda r: (r.get("path", ""), r.get("method", ""))) + + # 1. 摘除旧路由(按对象引用从 router.routes 过滤) + for r in self._route_objs.pop(wtype, []): + if r in self._app.router.routes: + self._app.router.routes.remove(r) + + # 2. 挂载新路由(MQTT / internal 端点不注册 HTTP,但保留在路由表中) + http_routes = [ + r for r in routes + if r.get("method") not in ("MQTT",) and not r.get("internal") + ] + if http_routes: + self._route_objs[wtype] = register_route_group( + self._app, self._wm, wtype, http_routes, self._prefix, self._schemas + ) + else: + logger.info("[%s] 无 HTTP 路由可挂载(%d 个描述符,全部为 MQTT/internal)", + wtype, len(routes)) + + # 路由表变化后 OpenAPI 缓存失效,下一次访问时重建 + self._app.openapi_schema = None + + # 3. 同步 MQTT 自定义订阅主题 + if self._mqtt_bridge is not None: + await self._mqtt_bridge.update_custom_topics(self._collect_mqtt_topics()) + + def _collect_mqtt_topics(self) -> List[Dict[str, Any]]: + """汇总所有插件 __meta__ 声明的自定义订阅主题。""" + topics: List[Dict[str, Any]] = [] + for wtype in sorted(self._meta): + meta = self._meta[wtype] + for t in meta.get("mqtt_topics") or []: + if isinstance(t, dict) and t.get("topic"): + topic = dict(t) + topic.setdefault("worker", wtype) + topics.append(topic) + return topics diff --git a/test_process_mode_e2e.py b/test_process_mode_e2e.py new file mode 100644 index 0000000..340cee8 --- /dev/null +++ b/test_process_mode_e2e.py @@ -0,0 +1,107 @@ +""" +process 模式端到端测试 + +前置条件: + 1. 网关已以 load_mode = "process" 启动(config.toml [development.workers]) + uv run serve + 2. 本脚本自己拉起/停止独立 worker 进程(uv run worker_serve --type example) + +验证: + 1. worker 启动后 → HTTP 路由动态出现且跨进程 RPC 可调 + 2. worker 优雅退出 → 路由动态摘除(404) + 3. worker 重启 → 路由恢复 +""" +import subprocess +import sys +import time +import urllib.request +from pathlib import Path + +GATEWAY_URL = "http://localhost:8000" +TEST_URL = f"{GATEWAY_URL}/api/v1/calc/add?a=3&b=4" +POLL_TIMEOUT = 30.0 + +# 优先使用 venv 控制台脚本(uv sync 后生成),不存在时退回 uv run +_WORKER_CMD = [str(Path(sys.executable).parent / "worker_serve"), "--type", "example"] +if not Path(_WORKER_CMD[0]).exists(): + _WORKER_CMD = ["uv", "run", "worker_serve", "--type", "example"] + +PASS = 0 +FAIL = 0 + + +def http_get(url): + try: + with urllib.request.urlopen(url, timeout=3) as resp: + return resp.status, resp.read().decode("utf-8") + except urllib.error.HTTPError as e: + return e.code, "" + except Exception as e: + return None, str(e) + + +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 wait_http(target_status=200, timeout=POLL_TIMEOUT): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + status, body = http_get(TEST_URL) + last = (status, body) + if status == target_status: + return True, body + time.sleep(0.5) + return False, last + + +def run(): + # 1. 拉起独立 worker 进程 + proc = subprocess.Popen( + _WORKER_CMD, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + try: + ok, body = wait_http(target_status=200) + report("worker 启动后路由动态注册 + 跨进程 RPC", ok, f"body={body}") + + # 2. 优雅退出 → 路由摘除(404) + proc.terminate() + proc.wait(timeout=15) + ok, _ = wait_http(target_status=404) + report("worker 退出后路由动态摘除 (404)", ok) + + # 3. 重启 worker → 路由恢复 + proc2 = subprocess.Popen( + _WORKER_CMD, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + try: + ok, body = wait_http(target_status=200) + report("worker 重启后路由恢复", ok, f"body={body}") + finally: + proc2.terminate() + proc2.wait(timeout=15) + finally: + if proc.poll() is None: + proc.terminate() + + print("\n" + "=" * 50) + print(f"测试结果: ✅ {PASS} 通过 ❌ {FAIL} 失败") + return 0 if FAIL == 0 else 1 + + +if __name__ == "__main__": + print("提示: 请先以 load_mode=process 启动网关 (uv run serve)\n") + sys.exit(run()) diff --git a/test_route_registry.py b/test_route_registry.py new file mode 100644 index 0000000..e2d1535 --- /dev/null +++ b/test_route_registry.py @@ -0,0 +1,160 @@ +""" +RouteRegistry 单元测试(loopback Zenoh 会话 + httpx ASGI 功能断言) + +验证 process 模式核心链路: + RoutePublisher.put 描述符 → 网关 RouteRegistry 动态挂载 FastAPI 路由(命中返回 503,worker 桩无实例) + RoutePublisher.delete → 动态摘除路由(404) + __meta__ 中的 mqtt_topics → 推送给 MQTT Bridge(update_custom_topics) + +注:FastAPI 0.139 的 include_router 将路由包装为 _IncludedRouter(惰性挂载), + 不能靠遍历 app.router.routes 的 path 判断,故用 ASGI 请求状态码做功能断言。 +""" +import asyncio +import time + +import httpx +import zenoh +from fastapi import FastAPI + +from src.chongming.app.core.route_registry import RouteRegistry +from src.chongming.app.core.route_publisher import RoutePublisher + +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: + """桩:无可用 worker,命中路由后 call 抛 RuntimeError(网关转 503)。""" + + async def call(self, **kwargs): + raise RuntimeError("无可用 worker(测试桩)") + + +class FakeBridge: + """记录 update_custom_topics 的调用结果。""" + + def __init__(self): + self.topics = None + + async def update_custom_topics(self, topics): + self.topics = list(topics) + + +async def wait_until(cond, timeout=10.0, interval=0.2) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if cond(): + return True + await asyncio.sleep(interval) + return False + + +HTTP_DESCRIPTOR = { + "subject": "example.add", + "method": "GET", + "path": "/calc/add", + "params": ["a: float", "b: float"], + "response_model": {"result": {"type": "float", "required": True}}, + "summary": "加法运算", + "timeout": 30.0, + "cacheable": False, + "auth_required": False, + "internal": False, +} + +MQTT_DESCRIPTOR = { + "subject": "example.add", + "method": "MQTT", + "path": "/calc/add_mqtt", + "params": ["a: float", "b: float"], + "summary": "MQTT 加法", + "timeout": 30.0, + "cacheable": False, + "auth_required": False, + "internal": False, +} + +MQTT_TOPICS = [{"topic": "box/aibox_sophon/board_ping"}] +# registry 会为每个主题补 worker 字段 +MQTT_TOPICS_EXPECTED = [ + {"topic": "box/aibox_sophon/board_ping", "worker": "example"} +] + + +async def main(): + zconfig = zenoh.Config() + try: + zconfig.from_file("config/zenoh_config.json5") + except Exception: + pass # 回退默认配置 + + gw_session = zenoh.open(zconfig) + pub_session = zenoh.open(zconfig) + try: + app = FastAPI() + registry = RouteRegistry(app, FakeWorkerManager(), prefix="/api/v1") + await registry.start(gw_session) + await asyncio.sleep(0.5) # 等待订阅生效 + + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + # 1. 发布 → HTTP 路由动态挂载(命中 503:路由存在但无 worker) + publisher = RoutePublisher(pub_session, "example") + publisher.publish_routes( + [HTTP_DESCRIPTOR, MQTT_DESCRIPTOR], mqtt_topics=MQTT_TOPICS + ) + + async def mounted(): + resp = await client.get("/api/v1/calc/add?a=1&b=2") + return resp.status_code == 503 + + ok = await wait_until(mounted) + check("put 后 HTTP 路由动态挂载(503 命中)", ok) + + resp = await client.get("/api/v1/calc/add_mqtt?a=1&b=2") + check("MQTT 端点不挂载 HTTP 路由(404)", resp.status_code == 404, + f"status={resp.status_code}") + + # 2. __meta__ 主题推送给 Bridge + bridge = FakeBridge() + registry.attach_bridge(bridge) + ok = await wait_until(lambda: bridge.topics == MQTT_TOPICS_EXPECTED) + check("mqtt_topics 推送给 Bridge", ok, f"bridge.topics={bridge.topics}") + + # 3. 重复发布相同描述符 → 幂等(路由不重复、仍可命中) + publisher.publish_routes([HTTP_DESCRIPTOR]) + await asyncio.sleep(3.5) # 覆盖防抖窗口 + resp = await client.get("/api/v1/calc/add?a=1&b=2") + check("重复发布幂等(仍正常命中)", resp.status_code == 503, + f"status={resp.status_code}") + + # 4. 摘除 → 路由动态移除(404) + publisher.unpublish() + + async def unmounted(): + resp = await client.get("/api/v1/calc/add?a=1&b=2") + return resp.status_code == 404 + + ok = await wait_until(unmounted) + check("delete 后 HTTP 路由动态摘除(404)", ok) + ok = await wait_until(lambda: bridge.topics == []) + check("元信息摘除后 Bridge 主题清空", ok, f"bridge.topics={bridge.topics}") + finally: + registry.stop() + gw_session.close() + pub_session.close() + + +if __name__ == "__main__": + asyncio.run(main()) + print("\n" + ("=" * 40)) + if failures: + print(f"❌ {len(failures)} 项未通过: {failures}") + raise SystemExit(1) + print("✅ RouteRegistry 全部通过")