From 1995b2d170508cba094147d2da765a81661073a0 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 14 Aug 2026 18:04:57 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20worker=20=E8=BF=90=E8=A1=8C=E6=97=B6?= =?UTF-8?q?=E4=B8=8B=E6=B2=89=20chongming-worker=20SDK=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E6=8F=92=E4=BB=B6=E4=B8=8E=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E9=9B=B6=E5=AF=BC=E5=85=A5=E5=85=B3=E7=B3=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 7 + pyproject.toml | 2 +- src/chongming/__init__.py | 120 ------------- src/chongming/app/core/contracts.py | 205 ++-------------------- src/chongming/app/core/route_generator.py | 90 +--------- src/chongming/app/core/route_publisher.py | 90 ---------- src/chongming/app/core/route_registry.py | 5 +- test_route_registry.py | 2 +- test_worker_isolation.py | 91 ++++++++++ uv.lock | 3 +- 10 files changed, 120 insertions(+), 495 deletions(-) delete mode 100644 src/chongming/app/core/route_publisher.py create mode 100644 test_worker_isolation.py diff --git a/README.md b/README.md index e600a3b..20b2617 100644 --- a/README.md +++ b/README.md @@ -384,6 +384,13 @@ uv pip install -e utils/ # 安装/更新工具包 进程入口 `uv run worker_serve --type `），经 Zenoh 在 `chongming/routes/**` 注册路由，网关动态挂载 HTTP 端点 - 两种模式可同时存在（混合部署） +- **进程插件与网关完全隔离**：进程插件只依赖 `chongming-worker` SDK + zenoh， + 与网关（`chongming` 包）零导入关系，仅通过 Zenoh（`chongming/routes/**` 注册路由、 + `worker//` 服务发现）通讯。`workers/` 目录仅为开发便利， + **生产部署时进程插件与网关分别打包到不同容器**： + - 网关容器：`chongming`（含 `src/chongming/`）+ 依赖 + - 进程插件容器：插件代码（`workers//`）+ `chongming-worker` + 依赖， + 入口 `chongming-worker --type `（或 `uv run worker_serve --type `） - **Zenoh 配置**：`config/zenoh_config.json5` - **插件注册**：`config/workers//config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容） - **路由定义**：`config/workers//*.toml`，每个端点一个文件 diff --git a/pyproject.toml b/pyproject.toml index 56d347c..9b483f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,4 +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" +worker_serve = "chongming_worker.process:worker_serve" diff --git a/src/chongming/__init__.py b/src/chongming/__init__.py index f89b63c..991f8c2 100644 --- a/src/chongming/__init__.py +++ b/src/chongming/__init__.py @@ -23,126 +23,6 @@ def serve(): 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() - - # 控制台脚本启动时 sys.path 不含项目根目录，显式加入以保证 - # plugins.（inline 插件）与 workers.（进程插件）均可导入 - project_root = os.path.abspath(os.getcwd()) - if project_root not in sys.path: - sys.path.insert(0, project_root) - - 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) - 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 会话 - 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/core/contracts.py b/src/chongming/app/core/contracts.py index 18d64c7..8714938 100644 --- a/src/chongming/app/core/contracts.py +++ b/src/chongming/app/core/contracts.py @@ -1,166 +1,31 @@ """ -启动契约校验（config/workers TOML ↔ @rpc_method 注册表 ↔ gen_models 生成模型） +启动契约校验（网关 inline 模式） -挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入 -（@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。 +校验逻辑的唯一实现在 chongming_worker.contracts（网关与 worker 进程共享）， +本模块只做网关侧的编排：对 inline 插件批量校验、日志汇总、硬失败终止启动。 -策略: - - 硬失败（汇总后 raise，启动终止）: - HTTP 路由（GET/POST/PUT/DELETE/PATCH 且 internal != true）在注册表中 - 缺少对应 subject#METHOD 的 @rpc_method → 该路由运行时必然 503，提前失败。 - - 软警告（仅日志汇总，不阻断启动）: - [MQTT] / internal 路由缺少注册、已注册方法无 TOML 路由、 - TOML 声明 params/response_model 但生成模型缺失（gen_models 过期）、孤儿模型。 +external 插件在自己进程内做同样校验（validate_worker_contracts），网关侧跳过。 """ -import importlib -from typing import Any, Dict, List, Optional, Set +from typing import Any, Dict, List from chongming_worker import get_all_rpc_methods +from chongming_worker.contracts import check_plugin, load_plugin_models +from chongming_worker.route_config import load_worker_routes_from_dir -from .route_generator import load_worker_routes_from_dir from .logger import get_logger logger = get_logger("contracts") -# 参与 HTTP 路由注册的方法（与 route_generator 的注册行为一致） -HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH") - - -def _model_name(base: str, method: str, suffix: str) -> str: - """与 gen_models._model_name 保持一致的模型命名: """ - method_title = "Mqtt" if method == "MQTT" else method.title() - return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}" - - -def _load_plugin_models(worker_type: str) -> Optional[Set[str]]: - """ - 加载插件 models.pydantic 中所有 Pydantic 模型的类名集合。 - - 返回 None 表示模块不存在（插件无生成模型，跳过模型相关校验）。 - 注: gen_models 生成的 `Response` 联合类型——多模型时为 UnionType - （自动被 isinstance(type) 过滤），单模型时是类的别名（与真实模型同一对象， - 按对象去重排除），两者都不会被误报为孤儿模型。 - """ - from pydantic import BaseModel - - for mod_name in ( - f"plugins.{worker_type}.models.pydantic", - f"src.plugins.{worker_type}.models.pydantic", - f"workers.{worker_type}.models.pydantic", - ): - try: - mod = importlib.import_module(mod_name) - except ImportError: - continue - names: Set[str] = set() - seen_objs: Set[int] = set() - for name in dir(mod): - obj = getattr(mod, name) - if ( - isinstance(obj, type) - and issubclass(obj, BaseModel) - and obj is not BaseModel - and id(obj) not in seen_objs - ): - seen_objs.add(id(obj)) - # 用 obj.__name__ 而非 dir() 名称：gen_models 生成的 - # Response 别名与真实模型同一对象，别名保留原类名， - # 按对象去重后自动收敛到真实模型名（避免别名抢占排序位） - names.add(obj.__name__) - return names - 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 模式，启动时）。 + 对每个发现的 inline 插件执行三方一致性校验。 + + 挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入 + （@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。 Args: - plugins: _discover_plugins() 返回的插件配置列表 + plugins: _discover_plugins() 返回的 inline 插件配置列表 Raises: RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时 @@ -176,8 +41,8 @@ def validate_contracts(plugins: List[Dict[str, Any]]) -> None: toml_routes = load_worker_routes_from_dir(worker_type) registered = registry.get(worker_type, {}) - model_names = _load_plugin_models(worker_type) - hard, warns = _check_plugin(worker_type, toml_routes, registered, model_names) + model_names = load_plugin_models(worker_type) + hard, warns = check_plugin(worker_type, toml_routes, registered, model_names) hard_errors.extend(hard) warnings.extend(warns) @@ -197,45 +62,3 @@ 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/route_generator.py b/src/chongming/app/core/route_generator.py index db7239c..b3ee88f 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -11,6 +11,8 @@ from inspect import Parameter, Signature from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends from pydantic import create_model, Field, BaseModel +from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_from_dir + from .worker_manager import WorkerManager from .logger import get_logger from .auth import store_user_to_request @@ -110,94 +112,6 @@ class RouteConfig: ] -# 默认的 worker 配置目录 -WORKER_CONFIG_DIR = "config/workers" - - -def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]: - """ - 从 config/workers/{worker_type}/ 目录加载路由配置。 - - 目录结构: - config/workers/{worker_type}/ - config.toml # [worker] 分组信息 + registration + defaults - add.toml # 接口名 = 文件名（不含扩展名） - [GET] - summary = ... - params = [...] - response_model = { ... } - [POST] - ... # 同一个接口支持多个请求方法 - [MQTT] - ... # MQTT 专用端点（不注册 HTTP 路由） - - 返回: List[Dict], 每个 dict 包含 flatten 后的路由配置, - 每个方法节(如 GET/POST/MQTT) 生成一个独立的路由条目。 - """ - from pathlib import Path - import tomllib - - base_dir = Path(WORKER_CONFIG_DIR) / worker_type - if not base_dir.is_dir(): - logger.warning(f"Worker 配置目录不存在: {base_dir}") - return [] - - # 1. 读取分组配置 - group_cfg_path = base_dir / "config.toml" - group_defaults: Dict[str, Any] = {} - registration: Dict[str, Any] = {} - router_prefix: str = "" - if group_cfg_path.exists(): - with open(group_cfg_path, "rb") as f: - cfg = tomllib.load(f) - w = cfg.get("worker", {}) - group_defaults = w.get("defaults", {}) - registration = w.get("registration", {}) - router_prefix = registration.get("router_prefix", "") - - # 2. 遍历所有接口 toml 文件 - routes: List[Dict[str, Any]] = [] - for route_file in sorted(base_dir.glob("*.toml")): - if route_file.name == "config.toml": - continue - - # 文件名（不含 .toml）作为路径名 - route_name = route_file.stem # 例如 "add" - base_path = router_prefix + "/" + route_name # 例如 "/calc/add" - - with open(route_file, "rb") as f: - route_data = tomllib.load(f) - - # 每个方法节（GET/POST/PUT/DELETE）生成一个独立路由 - for method, spec in route_data.items(): - if not isinstance(spec, dict): - logger.warning(f"忽略非 dict 的方法节: {route_file.name}[{method}]") - continue - - # 合并: spec 显式字段优先，然后 group_defaults 兜底 - route = dict(spec) - for key, default_val in group_defaults.items(): - if key not in route: - route[key] = default_val - - # 覆盖关键字段 - method 始终从 TOML 节键获取，不被 defaults 覆盖 - route["method"] = method.upper() - route["path"] = route.get("path", base_path) - if "subject" not in route: - # 自动生成 subject: worker_type.route_name (如 "calc.add") - route["subject"] = f"{worker_type}.{route_name}" - # 记录来源文件名（契约校验用其推断 gen_models 模型命名） - route["route_name"] = route_name - - routes.append(route) - logger.debug( - f" 已加载: {route['method']} {route['path']} -> {route['subject']}" - ) - - logger.info(f"从 {base_dir} 加载了 {len(routes)} 个路由") - return routes - - def register_route_group( app: FastAPI, worker_manager: WorkerManager, diff --git a/src/chongming/app/core/route_publisher.py b/src/chongming/app/core/route_publisher.py deleted file mode 100644 index 18f5612..0000000 --- a/src/chongming/app/core/route_publisher.py +++ /dev/null @@ -1,90 +0,0 @@ -""" -路由描述符发布器（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 index cef7087..483ef3f 100644 --- a/src/chongming/app/core/route_registry.py +++ b/src/chongming/app/core/route_registry.py @@ -24,14 +24,13 @@ from typing import Any, Dict, List, Optional import zenoh +from chongming_worker.routes_publisher import META_KEY, ROUTES_PREFIX, SCHEMA_VERSION + 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 diff --git a/test_route_registry.py b/test_route_registry.py index 0ab4c1c..6896eaa 100644 --- a/test_route_registry.py +++ b/test_route_registry.py @@ -17,8 +17,8 @@ import httpx import zenoh from fastapi import FastAPI +from chongming_worker.routes_publisher import RoutePublisher from src.chongming.app.core.route_registry import RouteRegistry -from src.chongming.app.core.route_publisher import RoutePublisher failures = [] diff --git a/test_worker_isolation.py b/test_worker_isolation.py new file mode 100644 index 0000000..48d5909 --- /dev/null +++ b/test_worker_isolation.py @@ -0,0 +1,91 @@ +""" +Worker 进程隔离性测试 + +验证进程插件与网关完全隔离：阻断 chongming（网关包）的两种导入路径后， +worker 侧全链路（SDK 导入 → 插件配置自读 → 插件 import → 契约校验）仍可用。 + +任何对 chongming.* / src.chongming.* 的导入都会直接触发 ImportError。 +""" +import sys +from pathlib import Path + +SRC = Path(__file__).parent +sys.path.insert(0, str(SRC)) + +# 阻断网关包导入（两种访问路径）。chongming_worker / zenoh / pydantic 等不受影响。 +sys.modules["chongming"] = None # type: ignore[assignment] +sys.modules["src.chongming"] = None # type: ignore[assignment] + +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) + + +# ========== 1. worker SDK 导入（若依赖网关包将在此抛 ImportError） ========== +import chongming_worker.contracts as con_mod # noqa: E402 +import chongming_worker.process as proc_mod # noqa: E402 +import chongming_worker.route_config as rc_mod # noqa: E402 +import chongming_worker.routes_publisher as pub_mod # noqa: E402 + +check("worker SDK 导入（阻断 chongming 后成功）", True) + +# ========== 2. 插件配置自读（不依赖网关 _discover_plugins） ========== +wc = proc_mod.load_plugin_config("demo") +check( + "插件配置自读（type/load_mode/module）", + wc["type"] == "demo" + and wc["load_mode"] == "external" + and wc["module"] == "workers.demo", + f"type={wc['type']}, load_mode={wc['load_mode']}, module={wc['module']}", +) + +# ========== 3. 完整 worker 启动前置链路：import 插件 + 契约校验 ========== +import workers.demo # noqa: E402,F401 触发 @rpc_method 注册 + +routes = con_mod.validate_worker_contracts("demo") +check( + "进程内契约校验通过（TOML ↔ 注册表 ↔ 模型）", + len(routes) == 1 and routes[0]["subject"] == "demo.hello", + f"routes={[r['subject'] for r in routes]}", +) + +# ========== 4. 代码级双重保险：SDK 无对网关包的 import 语句（AST 精确检查） ========== +import ast # noqa: E402 +import inspect # noqa: E402 + + +def _imports_gateway(mod) -> list: + """返回模块中引用 chongming（非 chongming_worker）的 import 节点列表""" + found = [] + tree = ast.parse(inspect.getsource(mod)) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "chongming" or alias.name.startswith("chongming."): + found.append(alias.name) + elif isinstance(node, ast.ImportFrom): + if node.module and ( + node.module == "chongming" or node.module.startswith("chongming.") + ): + found.append(node.module) + return found + + +for mod in (proc_mod, con_mod, pub_mod, rc_mod): + bad = _imports_gateway(mod) + if bad: + check(f"{mod.__name__} import 了网关包: {bad}", False) + break +else: + check("worker SDK 无对网关包的 import 语句", True) + +print("\n" + ("=" * 40)) +if failures: + print(f"❌ {len(failures)} 项未通过: {failures}") + sys.exit(1) +print("✅ Worker 进程隔离性全部通过") diff --git a/uv.lock b/uv.lock index 77a42c7..b31149a 100644 --- a/uv.lock +++ b/uv.lock @@ -319,9 +319,10 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git [[package]] name = "chongming-worker" version = "0.1.0" -source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#4908378e597eb200e255c71df92d16e272b4b12e" } +source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#c0a5b45b72a606c2c69850403d435c4883b4ff78" } dependencies = [ { name = "eclipse-zenoh" }, + { name = "pydantic" }, { name = "tomli" }, ]