chongming-utils-worker/src/chongming_worker/process.py

"""
独立 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", "cache"([workers.cache] 段), "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", "cache")}
    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,
        "cache": dict(workers_section.get("cache") or {}),
        "config": runtime_config,
    }


def build_worker_cache(cache_cfg: dict):
    """按插件 [workers.cache] 段建立配置缓存（redis/diskcache），返回 (cache, mode)。

    从配置中剔除内部控制键 launch_worker_count（决定配置优先来源 toml/cache），
    避免其被当作后端连接参数传给 Cache.from_config。
    """
    from chongming_cache import Cache

    mode = str((cache_cfg or {}).get("launch_worker_count", "toml")).lower()
    clean = {k: v for k, v in (cache_cfg or {}).items() if k != "launch_worker_count"}
    return Cache.from_config(clean), mode


def worker_serve():
    """独立 Worker 进程入口（console script: chongming-worker / worker_serve）。"""
    # 独立进程不依赖网关的日志配置，兜底 basicConfig（已有 handler 时不覆盖）
    if not logging.getLogger().handlers:
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
        )

    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] 节）")
    parser.add_argument(
        "--instance", default=None,
        help="外部进程实例标识（多实例全局分摊/逐实例精细扩缩用）。缺省用 "
             "CHONGMING_WORKER_INSTANCE 或 {hostname}:{pid}（同机多实例默认即自动区分）；"
             "跨重启稳定适逐实例缩放请显式指定",
    )
    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)

        # 1.5 配置缓存（[workers.cache] 段，进程插件自行管理）+ 配置优先来源登记
        #     [workers.config] 的单一事实来源解析由 worker 运行时（create_workers）统一完成。
        #     多实例场景下还登记「外部进程实例参与方」，使扩缩容按存活实例数分摊全局 count。
        from chongming_worker.runtime import (
            register_instance,
            run_instance_heartbeat,
            set_current_instance_id,
            set_worker_cache,
            set_worker_mode,
            unregister_instance,
        )
        set_current_instance_id(args.instance)

        cache = None
        mode = "toml"
        hb_task = None  # 实例参与方心跳任务（外部多实例全局分摊用）
        if wc.get("cache"):
            cache, mode = build_worker_cache(wc["cache"])
            set_worker_cache(args.type, cache)
            set_worker_mode(args.type, mode)
            register_instance(args.type, cache, args.instance)
            hb_task = asyncio.create_task(
                run_instance_heartbeat(args.type, cache, instance_id=args.instance)
            )
            logger.info(
                "[%s] 已启用配置缓存（backend=%s, mode=%s）并登记外部实例参与方",
                args.type, getattr(getattr(cache, "backend", None), "name", "?"), mode,
            )
        worker_config = wc["config"]

        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, worker_config, count, worker_manager=None)
            else:
                tasks = factory(session, worker_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)
            if hb_task is not None:
                hb_task.cancel()
                await asyncio.gather(hb_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)
            if cache is not None:
                unregister_instance(args.type, cache, args.instance)
            logger.info("[%s] Worker 进程已退出", args.type)
        finally:
            session.close()
            if cache is not None:
                try:
                    cache.close()
                except Exception:  # noqa: BLE001
                    pass

    try:
        asyncio.run(_run())
    except KeyboardInterrupt:
        pass