385 lines
15 KiB
Markdown
385 lines
15 KiB
Markdown
# 进程插件开发指南(external 模式)
|
||||
|
|
|
|||
|
|
> 进程插件(Process Plugin)是运行在**独立进程**中的 Worker 插件(`load_mode = "external"`),
|
|||
|
|
> 与网关零导入关系、仅经 Zenoh 通讯,可独立打包部署到不同容器。
|
|||
|
|
> 本文档覆盖从**创建代码 → 配置 → 调试 → 验证 → 打包 → 部署**的完整开发流程。
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
|
|||
|
|
## 1. 什么是进程插件
|
|||
|
|
|
|||
|
|
网关支持两种插件运行模式,**按插件粒度共存**(混合部署):
|
|||
|
|
|
|||
|
|
| 模式 | 运行位置 | 目录 | 路由来源 | 契约校验位置 |
|
|||
|
|
| ---- | -------- | ---- | -------- | ------------ |
|
|||
|
|
| **inline**(默认) | 网关进程内 import | `src/plugins/<name>/`(chongming-plugins 仓库) | 网关盘扫 TOML 生成 | 网关启动时 |
|
|||
|
|
| **external(进程插件)** | 独立进程 | `workers/<name>/`(本地开发目录,**不入库**) | Worker 经 Zenoh 发布路由描述符,网关动态挂载 | **Worker 进程内** |
|
|||
|
|
|
|||
|
|
进程插件的特点:
|
|||
|
|
|
|||
|
|
- **零导入关系**:只依赖 `chongming-worker` SDK + zenoh,禁止 import `chongming`(网关包)
|
|||
|
|
- **独立生命周期**:由 Docker/K8s 编排拉起,进程入口 `worker_serve --type <name>`(或 `chongming-worker --type <name>`)
|
|||
|
|
- **动态路由**:启动时把路由描述符发布到 Zenoh `chongming/routes/<type>/...`,网关 RouteRegistry 常驻订阅后动态挂载 HTTP 端点;退出时摘除
|
|||
|
|
- **独立打包**:`uv run build_plugin <name>` 一键打包为可部署 zip,与网关分别部署到不同容器
|
|||
|
|
- **代码结构一致**:与 inline 插件完全相同(Plugin 基类 + router/ + models/)
|
|||
|
|
|
|||
|
|
**适用场景**:需要独立扩缩容、强隔离(第三方依赖冲突、资源占用大)、或与网关发布节奏解耦的业务。
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
|
|||
|
|
## 2. 运行原理
|
|||
|
|
|
|||
|
|
### 2.1 架构
|
|||
|
|
|
|||
|
|
```mermaid
|
|||
|
|
graph TB
|
|||
|
|
HTTP[HTTP 客户端] -->|REST API| API[FastAPI 动态路由]
|
|||
|
|
REG[网关 RouteRegistry<br/>订阅 chongming/routes/**] -->|动态挂载/摘除| API
|
|||
|
|
API --> WM[WorkerManager<br/>负载均衡 + 超时重试]
|
|||
|
|
|
|||
|
|
subgraph GW[网关容器 chongming]
|
|||
|
|
API
|
|||
|
|
REG
|
|||
|
|
WM
|
|||
|
|
end
|
|||
|
|
|
|||
|
|
subgraph PW[进程插件容器 workers/demo]
|
|||
|
|
WS[worker_serve 进程]
|
|||
|
|
WS -->|① put 路由描述符<br/>chongming/routes/demo/**| REG
|
|||
|
|
WS -->|② liveliness 注册<br/>worker/demo/<id>| 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.<name> / workers.<name>)→ 触发 @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/<type>/<route_id>`,其中
|
|||
|
|
`route_id = <subject 的 . 替换为 _>__<METHOD>`(如 `demo_hello__GET`),
|
|||
|
|
描述符为 TOML 兼容 dict + `schema_version=1` + `type=<插件名>`
|
|||
|
|
- **元信息**:插件声明了 `[[workers.mqtt_topics]]` 时,额外发布
|
|||
|
|
`chongming/routes/<type>/__meta__`(含 mqtt_topics),网关据此动态订阅外部主题
|
|||
|
|
- **自愈**:每 10s 重发布一次全部描述符(`ANNOUNCE_INTERVAL`),网关后启动或样本丢失可恢复
|
|||
|
|
- **摘除**:优雅退出时 delete 全部已发布 key,网关 2s 防抖后摘除 HTTP 端点(请求变 404)
|
|||
|
|
- **网关侧**:RouteRegistry 常驻订阅 `chongming/routes/**`,与盘扫解耦——
|
|||
|
|
**网关容器无需放置 external 插件的配置文件**
|
|||
|
|
|
|||
|
|
### 2.4 服务发现
|
|||
|
|
|
|||
|
|
Worker 副本经 liveliness 注册 `worker/<type>/<id>`(与 inline 副本同一机制),
|
|||
|
|
WorkerManager 据此做负载均衡,网关侧无需任何改动。
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
|
|||
|
|
## 3. 目录结构约定
|
|||
|
|
|
|||
|
|
进程插件代码放在本地开发目录 `workers/<name>/`(**gitignore,不入库**):
|
|||
|
|
|
|||
|
|
```text
|
|||
|
|
workers/<name>/
|
|||
|
|
├── __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.<name>` 时,`sys.path` 需包含项目根目录——
|
|||
|
|
> `worker_serve` 已自动处理(console script 启动时会显式把 CWD 加入 `sys.path`)。
|
|||
|
|
|
|||
|
|
### 第 2 步:插件注册配置
|
|||
|
|
|
|||
|
|
创建 `config/workers/<name>/config.toml`,`[workers]` 节注册插件
|
|||
|
|
(**关键:`load_mode = "external"`**):
|
|||
|
|
|
|||
|
|
```toml
|
|||
|
|
# config/workers/demo/config.toml
|
|||
|
|
[workers]
|
|||
|
|
type = "demo" # 插件类型(worker type)
|
|||
|
|
module = "workers.demo" # Python 模块路径(进程插件用 workers.<name>)
|
|||
|
|
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/<name>/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/<name>/models/` 下生成 `pydantic.py` 与 `__init__.py`。
|
|||
|
|
|
|||
|
|
### 第 5 步:实现 RPC 端点
|
|||
|
|
|
|||
|
|
`workers/<name>/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/<name>/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/<name>/` 插件代码目录存在
|
|||
|
|
- `config/workers/<name>/config.toml` 存在且含 `[workers]` 节
|
|||
|
|
- `config/zenoh_config.json5` 存在
|
|||
|
|
- `load_mode` 非 `"external"` 时**告警**(不阻断,但会造成重复服务)
|
|||
|
|
|
|||
|
|
产物布局(**zip 根即容器工作目录**):
|
|||
|
|
|
|||
|
|
```text
|
|||
|
|
workers/<name>/ 插件代码(模块 workers.<name>)
|
|||
|
|
config/workers/<name>/ 插件注册 + 端点配置
|
|||
|
|
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 <name>
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
扩展依赖:`workers/<name>/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/<name>/` 仅部署在**插件容器**(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/<name>/config.toml` `[workers]` 节设置 `load_mode = "external"`,否则网关内副本与外部进程重复服务 |
|
|||
|
|
| 契约校验失败、进程退出 | 检查 TOML 端点(subject/方法节)与 `@rpc_method(subject=..., method=...)` 是否一致;模型缺失时重跑 `uv run gen_models config/workers/<name> workers/<name>` |
|
|||
|
|
| `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/<name>/config.toml`,`--config` 可指定全局配置路径) |
|
|||
|
|
| 容器内 import `workers.demo` 失败 | 确认在容器工作目录(zip 根)启动,`start.sh` 内含 `cd "$(dirname "$0")"`,worker_serve 会把 CWD 加入 sys.path |
|
|||
|
|
|
|||
|
|
---
|
|||
|
|
|
|||
|
|
## 8. 相关命令速查
|
|||
|
|
|
|||
|
|
```bash
|
|||
|
|
uv run worker_serve --type <name> # 本地调试独立 Worker 进程
|
|||
|
|
chongming-worker --type <name> # 容器入口(console script)
|
|||
|
|
uv run build_plugin <name> [--output <dir>] # 进程插件打包 → dist/<name>.zip
|
|||
|
|
uv run gen_models config/workers/<name> workers/<name> # 生成模型到进程插件目录
|
|||
|
|
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`(打包器)
|