From 589c2534d6355e2b0f36c6567d51075d54242f24 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 14 Aug 2026 17:50:08 +0800 Subject: [PATCH] =?UTF-8?q?add:=20worker=20=E4=BE=A7=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E6=97=B6=E4=B8=8B=E6=B2=89=EF=BC=88route=5Fconfig/contracts/ro?= =?UTF-8?q?utes=5Fpublisher/process=20=E5=85=A5=E5=8F=A3=EF=BC=8C=E9=9B=B6?= =?UTF-8?q?=E7=BD=91=E5=85=B3=E4=BE=9D=E8=B5=96=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 4 + src/chongming_worker/contracts.py | 189 ++++++++++++++++++++++ src/chongming_worker/process.py | 190 +++++++++++++++++++++++ src/chongming_worker/route_config.py | 103 ++++++++++++ src/chongming_worker/routes_publisher.py | 89 +++++++++++ 5 files changed, 575 insertions(+) create mode 100644 src/chongming_worker/contracts.py create mode 100644 src/chongming_worker/process.py create mode 100644 src/chongming_worker/route_config.py create mode 100644 src/chongming_worker/routes_publisher.py diff --git a/pyproject.toml b/pyproject.toml index 292f9e3..2e70ac9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,8 +5,12 @@ requires = ["hatchling"] [project] dependencies = [ "eclipse-zenoh>=1.9.0", + "pydantic>=2.0", "tomli>=2.4.1", ] + +[project.scripts] +chongming-worker = "chongming_worker.process:worker_serve" description = "chongming worker" name = "chongming-worker" readme = "README.md" diff --git a/src/chongming_worker/contracts.py b/src/chongming_worker/contracts.py new file mode 100644 index 0000000..fc83181 --- /dev/null +++ b/src/chongming_worker/contracts.py @@ -0,0 +1,189 @@ +""" +Worker 侧契约校验（网关与 worker 进程共享，唯一实现） + +校验三方一致性：config/workers TOML 路由 ↔ @rpc_method 注册表 ↔ gen_models 模型。 +网关 inline 模式启动时对 inline 插件批量校验（见网关 contracts.py 封装）； +worker 进程 external 模式在自己进程内对自身校验（validate_worker_contracts）。 + +策略: + - 硬失败（汇总后 raise）: + HTTP 路由（GET/POST/PUT/DELETE/PATCH 且 internal != true）在注册表中 + 缺少对应 subject#METHOD 的 @rpc_method → 该路由运行时必然 503。 + - 软警告（仅日志汇总，不阻断）: + [MQTT] / internal 路由缺少注册、已注册方法无 TOML 路由、 + TOML 声明 params/response_model 但生成模型缺失（gen_models 过期）、孤儿模型。 +""" +import importlib +import logging +from typing import Any, Dict, List, Optional, Set + +from chongming_worker.decorators import get_all_rpc_methods +from chongming_worker.route_config import load_worker_routes_from_dir, model_name + +logger = logging.getLogger("chongming_worker.contracts") + +# 参与 HTTP 路由注册的方法（与网关 route_generator 的注册行为一致） +HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH") + + +def load_plugin_models(worker_type: str) -> Optional[Set[str]]: + """ + 加载插件 models.pydantic 中所有 Pydantic 模型的类名集合。 + + 按序尝试 inline 插件（src/plugins/）与进程插件（workers/）路径； + 返回 None 表示模块不存在（插件无生成模型，跳过模型相关校验）。 + 注: gen_models 生成的 `Response` 联合类型——多模型时为 UnionType + （自动被 isinstance(type) 过滤），单模型时是类的别名（与真实模型同一对象， + 用 obj.__name__ 收敛到真实模型名，别名不会抢占排序位）。 + """ + 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)) + 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 = "（该插件注册表为空，请检查插件导入日志）" + 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_worker_contracts(worker_type: str) -> List[Dict[str, Any]]: + """ + Worker 进程侧契约校验（external 模式）：校验单个插件并返回可发布的路由描述符。 + + 调用时机：worker 进程 import 插件模块后（@rpc_method 注册表就绪）、 + 发布路由描述符之前。 + + 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_worker/process.py b/src/chongming_worker/process.py new file mode 100644 index 0000000..4bb134f --- /dev/null +++ b/src/chongming_worker/process.py @@ -0,0 +1,190 @@ +""" +独立 Worker 进程运行时（与网关完全隔离） + +进程插件与网关零导入关系：本模块只依赖 chongming_worker 自身、zenoh 与 +标准库，不 import chongming（网关包）。进程插件容器 = 插件代码 + chongming-worker。 + +进程入口（console script）: + chongming-worker --type [--count N] [--config config.toml] + +流程: + 1. 读全局 config.toml（仅 [zenoh].json5_path）→ 打开 Zenoh 会话 + 2. 自读 config/workers//config.toml 的 [workers] 节（插件注册 + 运行时配置） + 3. import 插件模块（plugins. / workers.）→ @rpc_method 注册 + 4. 进程内契约校验（TOML ↔ 注册表 ↔ 生成模型），硬失败则退出 + 5. create_workers 创建副本（count 进程内保留） + 6. 经 Zenoh 发布路由描述符（chongming/routes//...），每 10s 周期重发布自愈 + 7. SIGTERM/SIGINT → 摘除路由 + cleanup + 关闭会话 +""" +import argparse +import asyncio +import inspect +import logging +import os +import signal +import sys +import tomllib + +import zenoh + +from chongming_worker.contracts import validate_worker_contracts +from chongming_worker.routes_publisher import RoutePublisher + +logger = logging.getLogger("chongming_worker.process") + +# 路由描述符周期性重发布间隔（秒）：网关后启动/样本丢失时自愈 +ANNOUNCE_INTERVAL = 10.0 + + +def load_plugin_config(worker_type: str, config_dir: str = "config/workers") -> dict: + """ + 读取单个插件的 config.toml [workers] 节并展平（与网关 _discover_plugins + 的展平规则一致，worker 侧独立实现，不依赖网关）。 + + Returns: + {"type", "module", "factory", "cleanup", "count", "load_mode", + "mqtt_topics", "config": 展平后的运行时配置} + """ + import pathlib + + path = pathlib.Path(config_dir) / worker_type / "config.toml" + with open(path, "rb") as f: + plugin_conf = tomllib.load(f) + + workers_section = plugin_conf.get("workers", {}) + if not workers_section: + raise ValueError(f"{path} 缺少 [workers] 节") + + plugin_type = workers_section.get("type", worker_type) + load_mode = workers_section.get("load_mode", "inline") + + mqtt_topics = [] + for t in workers_section.get("mqtt_topics", []) or []: + if not isinstance(t, dict) or not t.get("topic"): + continue + t = dict(t) + t.setdefault("worker", plugin_type) + mqtt_topics.append(t) + + workers_flat = {k: v for k, v in workers_section.items() + if k not in ("config", "mqtt_topics", "load_mode")} + runtime_config = { + **plugin_conf, + **workers_flat, + **workers_section.get("config", {}), + } + + return { + "type": plugin_type, + "module": workers_section.get("module", f"plugins.{plugin_type}"), + "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, + } + + +def worker_serve(): + """独立 Worker 进程入口（console script: chongming-worker / worker_serve）。""" + parser = argparse.ArgumentParser(description="chongming 独立 Worker 进程") + parser.add_argument("--type", required=True, help="插件类型（如 demo / example）") + parser.add_argument("--count", type=int, default=None, help="副本数（覆盖配置中的 count）") + parser.add_argument("--config", default="config.toml", help="全局配置文件路径（读取 [zenoh] 节）") + args = parser.parse_args() + + # console script 启动时 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) + + try: + wc = load_plugin_config(args.type) + except FileNotFoundError as e: + logger.error("未找到插件配置: %s", e) + sys.exit(1) + except ValueError as e: + logger.error("%s", e) + sys.exit(1) + + if wc.get("load_mode", "inline") != "external": + logger.warning( + "[%s] 插件配置 load_mode 不是 \"external\"：网关也会在进程内加载该插件，" + "外部进程将与网关内副本重复服务。若以独立进程运行，请在插件 " + "config.toml 的 [workers] 节设置 load_mode = \"external\"。", + args.type, + ) + + # 全局配置只读 [zenoh] 节（worker 不依赖网关的配置结构） + with open(args.config, "rb") as f: + global_config = tomllib.load(f) + json5_path = global_config.get("zenoh", {}).get("json5_path", "config/zenoh_config.json5") + + async def _run(): + # 1. Zenoh 会话 + zconfig = zenoh.Config() + zconfig.from_file(json5_path) + 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 副本（跨进程经 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=ANNOUNCE_INTERVAL) + 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 diff --git a/src/chongming_worker/route_config.py b/src/chongming_worker/route_config.py new file mode 100644 index 0000000..076aa87 --- /dev/null +++ b/src/chongming_worker/route_config.py @@ -0,0 +1,103 @@ +""" +Worker 路由 TOML 配置解析（网关与 worker 进程共享，唯一实现） + +从 config/workers// 目录解析路由描述符（TOML 兼容格式）， +网关 inline 模式用于盘扫注册 FastAPI 路由，worker 进程 external 模式用于 +自读配置 + 契约校验 + 发布描述符。两边共用本模块保证格式一致。 +""" +import logging +import tomllib +from pathlib import Path +from typing import Any, Dict, List + +logger = logging.getLogger("chongming_worker.route_config") + +# 默认的 worker 配置目录 +WORKER_CONFIG_DIR = "config/workers" + + +def model_name(base: str, method: str, suffix: str) -> str: + """gen_models 模型命名约定: """ + method_title = "Mqtt" if method == "MQTT" else method.title() + return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}" + + +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) 生成一个独立的路由条目。 + """ + 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 diff --git a/src/chongming_worker/routes_publisher.py b/src/chongming_worker/routes_publisher.py new file mode 100644 index 0000000..98e3100 --- /dev/null +++ b/src/chongming_worker/routes_publisher.py @@ -0,0 +1,89 @@ +""" +路由描述符发布器（worker 进程侧，协议唯一实现） + +独立 worker 进程在完成契约校验后，将路由描述符 put 到 Zenoh 的 +`chongming/routes//...` 命名空间；进程优雅退出时 delete 摘除。 + +网关 RouteRegistry 订阅同名空间并动态挂载 HTTP 路由。 +描述符 schema_version=1，字段与网关 RouteConfig 完全对位（TOML 兼容 dict）。 +""" +import json +import logging +from typing import Any, Dict, List, Optional + +import zenoh + +logger = logging.getLogger("chongming_worker.routes_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()