From c0b9d7e75fcace443bfdfd1b6911ddf99fcdb34e Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 21 Aug 2026 14:55:31 +0800 Subject: [PATCH] =?UTF-8?q?modify=20=E8=BF=9B=E7=A8=8B=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E5=BC=80=E5=8F=91=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1 + docs/process-plugin.md | 384 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 385 insertions(+) create mode 100644 docs/process-plugin.md diff --git a/README.md b/README.md index 866ac60..70e5bca 100644 --- a/README.md +++ b/README.md @@ -676,5 +676,6 @@ uv pip install -e utils/ # 安装/更新工具包子包 - [配置说明](config/README.md) - [插件开发指南](src/plugins/README.md) +- [进程插件开发指南（external 模式，从开发到部署）](docs/process-plugin.md) - [前端开发指南](src/chongming-web/README.md) - [Worker 框架](utils/worker/README.md) diff --git a/docs/process-plugin.md b/docs/process-plugin.md new file mode 100644 index 0000000..3ff5108 --- /dev/null +++ b/docs/process-plugin.md @@ -0,0 +1,384 @@ +# 进程插件开发指南（external 模式） + +> 进程插件（Process Plugin）是运行在**独立进程**中的 Worker 插件（`load_mode = "external"`）， +> 与网关零导入关系、仅经 Zenoh 通讯，可独立打包部署到不同容器。 +> 本文档覆盖从**创建代码 → 配置 → 调试 → 验证 → 打包 → 部署**的完整开发流程。 + +--- + +## 1. 什么是进程插件 + +网关支持两种插件运行模式，**按插件粒度共存**（混合部署）： + +| 模式 | 运行位置 | 目录 | 路由来源 | 契约校验位置 | +| ---- | -------- | ---- | -------- | ------------ | +| **inline**（默认） | 网关进程内 import | `src/plugins//`（chongming-plugins 仓库） | 网关盘扫 TOML 生成 | 网关启动时 | +| **external（进程插件）** | 独立进程 | `workers//`（本地开发目录，**不入库**） | Worker 经 Zenoh 发布路由描述符，网关动态挂载 | **Worker 进程内** | + +进程插件的特点： + +- **零导入关系**：只依赖 `chongming-worker` SDK + zenoh，禁止 import `chongming`（网关包） +- **独立生命周期**：由 Docker/K8s 编排拉起，进程入口 `worker_serve --type `（或 `chongming-worker --type `） +- **动态路由**：启动时把路由描述符发布到 Zenoh `chongming/routes//...`，网关 RouteRegistry 常驻订阅后动态挂载 HTTP 端点；退出时摘除 +- **独立打包**：`uv run build_plugin ` 一键打包为可部署 zip，与网关分别部署到不同容器 +- **代码结构一致**：与 inline 插件完全相同（Plugin 基类 + router/ + models/） + +**适用场景**：需要独立扩缩容、强隔离（第三方依赖冲突、资源占用大）、或与网关发布节奏解耦的业务。 + +--- + +## 2. 运行原理 + +### 2.1 架构 + +```mermaid +graph TB + HTTP[HTTP 客户端] -->|REST API| API[FastAPI 动态路由] + REG[网关 RouteRegistry
订阅 chongming/routes/**] -->|动态挂载/摘除| API + API --> WM[WorkerManager
负载均衡 + 超时重试] + + subgraph GW[网关容器 chongming] + API + REG + WM + end + + subgraph PW[进程插件容器 workers/demo] + WS[worker_serve 进程] + WS -->|① put 路由描述符
chongming/routes/demo/**| REG + WS -->|② liveliness 注册
worker/demo/| WM + end + + WM -->|Zenoh Query| WS + WS -.->|每 10s 重发布（自愈）| REG + WS -.->|SIGTERM 时 delete 摘除| REG +``` + +### 2.2 Worker 进程启动流程（`chongming_worker.process.worker_serve`） + +```text +worker_serve --type demo + 1. 读全局 config.toml（仅 [zenoh] 节）→ 打开 Zenoh 会话 + 2. 自读 config/workers/demo/config.toml 的 [workers] 节（load_plugin_config 展平） + 3. import 插件模块（plugins. / workers.）→ 触发 @rpc_method 注册 + 4. 进程内契约校验（TOML ↔ 注册表 ↔ 生成模型），硬失败则退出（退出码 1） + 5. create_workers 创建 Worker 副本（默认 count 个，--count 可覆盖） + 6. RoutePublisher 发布路由描述符 → 网关动态挂载 HTTP 路由 + 7. 等待退出信号；期间每 10s 周期性重发布（网关后启动 / 样本丢失时自愈） + 8. SIGTERM/SIGINT → 摘除路由 → cleanup → 取消副本任务 → 关闭会话 +``` + +### 2.3 路由注册与摘除协议 + +- **发布**：路由描述符 put 到 `chongming/routes//`，其中 + `route_id = __`（如 `demo_hello__GET`）， + 描述符为 TOML 兼容 dict + `schema_version=1` + `type=<插件名>` +- **元信息**：插件声明了 `[[workers.mqtt_topics]]` 时，额外发布 + `chongming/routes//__meta__`（含 mqtt_topics），网关据此动态订阅外部主题 +- **自愈**：每 10s 重发布一次全部描述符（`ANNOUNCE_INTERVAL`），网关后启动或样本丢失可恢复 +- **摘除**：优雅退出时 delete 全部已发布 key，网关 2s 防抖后摘除 HTTP 端点（请求变 404） +- **网关侧**：RouteRegistry 常驻订阅 `chongming/routes/**`，与盘扫解耦—— + **网关容器无需放置 external 插件的配置文件** + +### 2.4 服务发现 + +Worker 副本经 liveliness 注册 `worker//`（与 inline 副本同一机制）， +WorkerManager 据此做负载均衡，网关侧无需任何改动。 + +--- + +## 3. 目录结构约定 + +进程插件代码放在本地开发目录 `workers//`（**gitignore，不入库**）： + +```text +workers// +├── __init__.py # Plugin 子类 + create_workers / cleanup 模块级入口 +├── global_var.py # 模块级共享状态（Database、API client 等） +├── router/ # RPC 端点实现 +│ ├── __init__.py # 导入所有端点模块 +│ └── <操作名>.py # 每个端点一个文件，@rpc_method 装饰器 +├── models/ # 数据模型（gen_models 自动生成，勿手动修改） +│ ├── __init__.py +│ └── pydantic.py # 请求/响应 Pydantic 模型 +├── requirements.txt # 扩展依赖（可选，每行一个包，打包时并入 pyproject.toml） +└── README.md # 插件说明（可选） +``` + +> 参考示例：`config/workers/demo/` 中的注册与端点配置（`demo` / `hello.toml`）， +> 以及 `workers/demo/`（本地目录，配合 `test_worker_isolation.py` 使用）。 + +--- + +## 4. 开发流程 + +### 第 1 步：创建插件代码 + +```bash +mkdir -p workers/demo/router +mkdir -p workers/demo/models +touch workers/demo/__init__.py +touch workers/demo/global_var.py +touch workers/demo/router/__init__.py +``` + +`workers/demo/__init__.py`（与 inline 插件完全一致，推荐 Plugin 基类风格）： + +```python +import logging + +from chongming_worker.plugin import Plugin + +from . import router # noqa: F401 触发 @rpc_method 注册 + +logger = logging.getLogger(__name__) + + +class DemoPlugin(Plugin): + type = "demo" + db_name = None # 不需要数据库则置 None + auto_bind = True + + async def on_init(self): + """首次 create_workers 时调用一次：初始化资源。""" + # self.set_var("db", self.db) + pass + + async def on_cleanup(self): + """cleanup 时调用：关闭自定义客户端等。""" + + +_plugin = DemoPlugin() +create_workers = _plugin.create_workers +cleanup = _plugin.cleanup +``` + +> 注意：`module` 注册为 `workers.` 时，`sys.path` 需包含项目根目录—— +> `worker_serve` 已自动处理（console script 启动时会显式把 CWD 加入 `sys.path`）。 + +### 第 2 步：插件注册配置 + +创建 `config/workers//config.toml`，`[workers]` 节注册插件 +（**关键：`load_mode = "external"`**）： + +```toml +# config/workers/demo/config.toml +[workers] +type = "demo" # 插件类型（worker type） +module = "workers.demo" # Python 模块路径（进程插件用 workers.） +factory = "create_workers" # 工厂函数名（Plugin 基类绑定的方法） +cleanup = "cleanup" # 清理函数名 +count = 2 # Worker 副本数 +load_mode = "external" # ★ 进程插件：独立进程运行，经 Zenoh 注册路由 + +[worker] +name = "demo" +version = "0.1.0" +description = "进程插件示例" +defaults = { method = "GET", timeout = 30.0, auth_required = false, cacheable = false, internal = false } + +[worker.registration] +router_prefix = "/demo" # HTTP 路由前缀 +tags = ["demo", "process-plugin"] + +# Worker 运行时配置（展平后传给 create_workers 的 config 参数） +[workers.config] +greeting_prefix = "Hello" +``` + +### 第 3 步：端点 TOML 定义 + +每个端点一个文件，如 `config/workers//hello.toml`： + +```toml +[GET] +summary = "进程插件示例：问候" +params = ["name: str"] +auth_required = false + +[GET.response_model] +greeting = { type = "str", required = true } +worker_id = { type = "str", required = false, default = "" } +timestamp = { type = "float", required = true } +``` + +流式端点（SSE）在方法节追加 `stream = true` 并配大 `timeout`，详见主 README「流式端点」。 + +### 第 4 步：生成 Pydantic 模型 + +```bash +uv run gen_models config/workers/demo workers/demo +``` + +在 `workers//models/` 下生成 `pydantic.py` 与 `__init__.py`。 + +### 第 5 步：实现 RPC 端点 + +`workers//router/hello.py`： + +```python +import time + +from chongming_worker.decorators import rpc_method + +from ..models.pydantic import HelloGetRequest, HelloGetResponse + + +@rpc_method(subject="demo.hello", method="GET", response_model=HelloGetResponse) +async def hello(request: HelloGetRequest) -> HelloGetResponse: + return HelloGetResponse( + greeting=f"Hello, {request.name}!", + timestamp=time.time(), + ) +``` + +`workers//router/__init__.py`： + +```python +from . import hello + +__all__ = ["hello"] +``` + +### 第 6 步：本地调试 + +```bash +uv run worker_serve --type demo # 本地调试（开发环境） +uv run worker_serve --type demo --count 1 # 覆盖副本数 +chongming-worker --type demo # 容器入口（console script） +``` + +启动后预期日志： + +```text +[ demo ] 路由发布完成: 1 端点 + 无元信息 +[ demo ] Worker 进程就绪，等待退出信号... +``` + +> 若配置了 `load_mode` 非 `"external"`，会输出告警：网关也会在进程内加载该插件， +> 外部进程将与网关内副本**重复服务**。 + +### 第 7 步：验证 + +```bash +# 1. 进程隔离性（阻断 chongming 导入 + AST 检查 import 语句） +uv run python test_worker_isolation.py + +# 2. 端到端（需先启动网关 uv run serve，且被测插件 load_mode = "external"）： +# 路由动态注册 → 跨进程 RPC → 退出摘除(404) → 重启恢复 +uv run python test_process_mode_e2e.py +``` + +### 第 8 步：打包 + +```bash +uv run build_plugin demo # → dist/demo.zip +uv run build_plugin demo --output /tmp/pkg # 指定输出目录 +``` + +打包前校验（任一不满足即失败退出）： + +- `workers//` 插件代码目录存在 +- `config/workers//config.toml` 存在且含 `[workers]` 节 +- `config/zenoh_config.json5` 存在 +- `load_mode` 非 `"external"` 时**告警**（不阻断，但会造成重复服务） + +产物布局（**zip 根即容器工作目录**）： + +```text +workers// 插件代码（模块 workers.） +config/workers// 插件注册 + 端点配置 +config/zenoh_config.json5 Zenoh 会话配置 +config.toml 最小根配置（自动生成，仅 [zenoh] 节，不泄露网关配置） +pyproject.toml uv 依赖清单（chongming-worker 内网 git 源 + orjson + 扩展依赖） +start.sh 容器入口：uv sync --no-dev + uv run worker_serve --type +``` + +扩展依赖：`workers//requirements.txt` 中逐行声明（`#` 开头为注释）， +打包时自动合并进 `pyproject.toml` 的 dependencies。 + +### 第 9 步：容器部署 + +```bash +# 镜像需预装 uv（https://docs.astral.sh/uv/） +unzip demo.zip -d /app +cd /app +./start.sh +``` + +- `start.sh` 自动执行 `uv sync --no-dev`（按 pyproject 从内网 git 源拉取 chongming-worker）并启动 worker +- 生产部署时进程插件与网关**分别打包到不同容器**： + - 网关容器：`chongming` + 依赖 + - 进程插件容器：插件代码 + `chongming-worker` + 依赖 +- `config/workers//` 仅部署在**插件容器**（worker 自读）；网关侧无需放置 + （RouteRegistry 常驻订阅 Zenoh 路由描述符，与盘扫解耦） + +--- + +## 5. 隔离性约束 + +进程插件**禁止 import `chongming`（网关包）**，只允许依赖： + +- `chongming-worker`（worker SDK：`chongming_worker.*`） +- `zenoh` / `pydantic` 等第三方库 +- 插件自身的扩展依赖（requirements.txt） + +由 `test_worker_isolation.py` 双重保证： + +1. **运行时阻断**：`sys.modules["chongming"] = None` 后走完整链路 + （SDK 导入 → 插件配置自读 → 插件 import → 契约校验），任何网关依赖立即 ImportError +2. **静态检查**：AST 解析 SDK 各模块源码，确认无 `chongming` / `chongming.*` import 语句 + +--- + +## 6. 进程内契约校验 + +Worker 进程启动时自读 TOML 并**在自己进程内**做三方一致性校验 +（`chongming_worker.contracts.validate_worker_contracts`），校验失败直接退出： + +| 检查项 | 策略 | +| ---------------------------- | --------------- | +| TOML HTTP 路由缺 `@rpc_method` | **硬失败，进程退出** | +| `[MQTT]` / `internal` 路由缺注册 | 警告日志 | +| 已注册方法无 TOML 路由 | 警告日志（反向漂移检测） | +| params/response_model 缺生成模型 | 警告日志（gen_models 过期） | +| 生成模型未被任何路由引用 | 警告日志（孤儿模型） | + +--- + +## 7. 常见问题排查 + +| 现象 | 原因 / 处理 | +| ---- | ----------- | +| 启动告警「load_mode 不是 external」 | 进程插件必须在 `config/workers//config.toml` `[workers]` 节设置 `load_mode = "external"`，否则网关内副本与外部进程重复服务 | +| 契约校验失败、进程退出 | 检查 TOML 端点（subject/方法节）与 `@rpc_method(subject=..., method=...)` 是否一致；模型缺失时重跑 `uv run gen_models config/workers/ workers/` | +| `TypeError: rpc_method() got an unexpected keyword argument 'stream'` | venv 里是旧版 chongming-worker（git 源拉取），本地未 push 的框架改动不生效；联调时 `uv pip install -e utils/worker` | +| 路由没出现 | worker 是否就绪（日志「路由发布完成」）；网关 RouteRegistry 是否订阅（启动流程 5.1）；每 10s 自愈重发布，稍等或重启 worker | +| 路由一直 404 | worker 是否已退出（优雅退出会摘除路由）；进程被 SIGKILL 时描述符残留，重启 worker 后重发布自愈覆盖 | +| 找不到插件配置 | `worker_serve` 须在项目根目录下运行（读 `config/workers//config.toml`，`--config` 可指定全局配置路径） | +| 容器内 import `workers.demo` 失败 | 确认在容器工作目录（zip 根）启动，`start.sh` 内含 `cd "$(dirname "$0")"`，worker_serve 会把 CWD 加入 sys.path | + +--- + +## 8. 相关命令速查 + +```bash +uv run worker_serve --type # 本地调试独立 Worker 进程 +chongming-worker --type # 容器入口（console script） +uv run build_plugin [--output ] # 进程插件打包 → dist/.zip +uv run gen_models config/workers/ workers/ # 生成模型到进程插件目录 +uv run python test_worker_isolation.py # 进程隔离性测试 +uv run python test_process_mode_e2e.py # 端到端测试（先启动网关） +``` + +--- + +## 9. 参考 + +- [主 README：进程插件与混合部署](../README.md) +- [Worker 插件开发指南（inline，结构约定与装饰器细节）](../src/plugins/README.md) +- [Worker 框架（chongming-worker）](../utils/worker/README.md) +- [chongming-cli（build_plugin / gen_models 等命令）](../utils/cli/README.md) +- 代码：`utils/worker/src/chongming_worker/process.py`（进程入口）、 + `routes_publisher.py`（路由发布协议）、`contracts.py`（契约校验）、 + `utils/cli/src/chongming_cli/build_plugin.py`（打包器）