add: worker 侧运行时下沉（route_config/contracts/routes_publisher/process 入口，零网关依赖）
This commit is contained in:
@@ -0,0 +1,190 @@
|
||||
"""
|
||||
独立 Worker 进程运行时（与网关完全隔离）
|
||||
|
||||
进程插件与网关零导入关系：本模块只依赖 chongming_worker 自身、zenoh 与
|
||||
标准库，不 import chongming（网关包）。进程插件容器 = 插件代码 + chongming-worker。
|
||||
|
||||
进程入口（console script）:
|
||||
chongming-worker --type <name> [--count N] [--config config.toml]
|
||||
|
||||
流程:
|
||||
1. 读全局 config.toml（仅 [zenoh].json5_path）→ 打开 Zenoh 会话
|
||||
2. 自读 config/workers/<type>/config.toml 的 [workers] 节（插件注册 + 运行时配置）
|
||||
3. import 插件模块（plugins.<name> / workers.<name>）→ @rpc_method 注册
|
||||
4. 进程内契约校验（TOML ↔ 注册表 ↔ 生成模型），硬失败则退出
|
||||
5. create_workers 创建副本（count 进程内保留）
|
||||
6. 经 Zenoh 发布路由描述符（chongming/routes/<type>/...），每 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.<name>（inline 插件）与 workers.<name>（进程插件）均可导入
|
||||
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
|
||||
Reference in New Issue
Block a user