191 lines
7.6 KiB
Python
191 lines
7.6 KiB
Python
|
"""
|
|||
|
|
独立 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
|