From fe96fc9b6cade4f1cdd709693652dad97617f3f1 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Tue, 13 Oct 2026 17:29:38 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E8=AF=B4=E6=98=8E=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E9=87=8D=E6=95=B4=EF=BC=88README=20=E6=94=B9=E4=BD=BF=E7=94=A8?= =?UTF-8?q?=E5=AF=BC=E5=90=91=EF=BC=8C=E8=BF=90=E7=BB=B4=E7=BB=86=E8=8A=82?= =?UTF-8?q?=E7=A7=BB=E5=85=A5=20docs/worker-management.md=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1079 ++++++++-------------------- deploy/README.md | 5 +- docs/multi-language-plugin-plan.md | 14 +- docs/plugin-wire-protocol.md | 8 +- docs/process-plugin.md | 70 +- docs/rust-plugin.md | 18 +- docs/worker-management.md | 163 +++++ 7 files changed, 512 insertions(+), 845 deletions(-) create mode 100644 docs/worker-management.md diff --git a/README.md b/README.md index 2aad3df..e8422e9 100644 --- a/README.md +++ b/README.md @@ -1,104 +1,37 @@ # Chongming-Zenoh -基于 **FastAPI + Zenoh + MQTT + Vue 3** 的**配置驱动微服务网关平台**。 +基于 **FastAPI + Zenoh + MQTT** 的**配置驱动微服务网关平台**。 -通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 和 **MQTT** 双通道。底层基于 Zenoh 分布式消息协议实现 RPC 通信，Worker 插件可独立开发、按副本数水平扩展。 - ---- +用 TOML 定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 与 +**MQTT** 双通道；底层经 Zenoh 做 Worker 服务发现与 RPC 调用。插件（Worker）可独立开发、 +按副本数水平扩展，既可跑在网关进程内（inline），也可作为独立进程/容器部署（external）。 ## 核心特性 -| 特性 | 说明 | -| --------------------- | -------------------------------------------------------------------- | -| **FastAPI + Uvicorn** | 高性能异步 Web 框架，自动生成 OpenAPI 文档 | -| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 实例定向 + 超时重试 | -| **MQTT 集成** | amqtt Broker（内嵌或外部），MQTT ↔ Worker 双向桥接 | -| **动态路由生成** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由 | -| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 | -| **启动契约校验** | TOML 路由 ↔ `@rpc_method` ↔ 生成模型三方一致性校验，fail fast | -| **混合部署** | inline（网关进程内）与 external（独立进程）两种插件模式按插件粒度共存 | -| **动态路由注册** | external 插件经 Zenoh 注册路由，网关动态挂载/摘除 HTTP 端点 | -| **进程插件隔离** | 进程插件与网关零导入关系，仅经 Zenoh 通讯，可独立打包到不同容器 | -| **自动化模型生成** | `gen_models` 根据 TOML 配置生成请求/响应模型 | -| **插件脚手架** | `gen_plugin` 一键生成风格统一的 Worker 插件骨架 | -| **JWT 鉴权** | Rust 实现，路由级 `auth_required` 控制 | -| **Argon2id 密码哈希** | Rust 实现的高性能密码安全方案 | -| **异步 SQLite** | aiosqlite + SQLModel，多进程并发安全 | -| **本地文件存储** | 内置文件上传/下载端点 | -| **流媒体代理** | ffmpeg + MediaMTX RTSP 流转发 | -| **构建打包** | PyInstaller / Nuitka 一键打包为独立可执行文件 | -| **Vue 3 前端** | Element Plus UI、Pinia 状态管理、类型安全的 API 客户端 | -| **Worker 运维管理** | 前端经 Zenoh 向单个副本下发 restart/stop/改配置指令（超管），前端可视化列表 + 分页管理 | +| 特性 | 说明 | +| ---- | ---- | +| **双通道接入** | HTTP（FastAPI，自动 OpenAPI 文档）与 MQTT（内嵌或外部 Broker） | +| **配置驱动路由** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由与 Pydantic 模型 | +| **Zenoh RPC** | 基于 liveliness 的 Worker 发现与调用：负载均衡、超时重试、实例定向 | +| **混合部署** | inline（网关进程内）与 external（独立进程）按插件粒度共存 | +| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类生命周期 + 启动契约校验 | +| **流式端点** | `@rpc_method` 返回生成器即得 SSE（`text/event-stream`）逐块输出 | +| **Worker 运维** | REST/前端管理：副本重启、配置热更新、扩缩容（全局对称 / 逐实例） | +| **鉴权与安全** | 网关内置登录与 JWT 签发（Rust 实现）；Rust Argon2id 密码哈希 | +| **插件脚手架** | `gen_plugin` 生成统一骨架，`gen_models` 生成模型，`build_plugin` 打包进程插件 | +| **交付打包** | PyInstaller / Nuitka 一键打独立可执行文件；`.mbank` 模块银行支持插件加密与热替换 | +| **Vue 3 前端** | 管理台（Worker 管理 + 登录）与自动生成的 API 客户端 | ---- +## 快速开始 -## 项目结构 +### 环境要求 -```text -chongming-zenoh/ -├── config.toml # 应用主配置 -├── pyproject.toml # 项目依赖与脚本入口 -├── config/ # 配置文件目录 -│ ├── zenoh_config.json5 # Zenoh 会话配置 -│ └── workers/ # Worker 路由定义（TOML） -│ ├── config.toml # 旧式全局插件注册（仅作兜底兼容） -│ ├── auth/ # 鉴权插件（config.toml = 插件注册 + 运行时配置） -│ ├── doubao/ # 豆包视觉理解插件 -│ ├── aibox_sophon_adapter/ # AI Box 设备心跳监控插件 -│ └── example/ # 示例插件（四则运算 / MQTT 端点） -├── src/ -│ ├── chongming/ # 核心应用 -│ │ ├── app/ -│ │ │ ├── core/ # WorkerManager, RouteGenerator, MQTTBridge, contracts(契约校验), auth -│ │ │ └── routes/ # 内置路由（文件上传等） -│ ├── chongming-web/ # Vue 3 前端 -│ │ └── src/ -│ │ ├── api/generated/ # 自动生成 API 客户端 -│ │ ├── views/ # 页面组件 -│ │ └── router/ # Vue Router 配置 -│ └── plugins/ # Worker 插件源码（inline 模式） -│ ├── auth/ # JWT 鉴权 -│ ├── doubao/ # 豆包 AI（支持 AICC 安全通道） -│ ├── aibox_sophon_adapter/ # 设备心跳 + 告警 -│ └── example/ # 演示示例 -├── workers/ # 进程插件（external 模式，本地开发目录，不入库； -│ └── demo/ # 生产在独立容器部署，示例见本地目录） -├── utils/ # 独立工具包 -│ ├── bytedance/ # 字节跳动安全通信 SDK -│ ├── cli/ # 命令脚本集合（build / gen_models / gen_plugin，chongming-cli） -│ ├── config/ # TOML 配置加载 -│ ├── database/ # 异步 SQLite 封装 -│ ├── jwt/ # Rust JWT 编解码 -│ ├── password/ # Rust Argon2id 哈希 -│ ├── request/ # Zenoh RPC 客户端 -│ ├── stream_manager/ # ffmpeg RTSP 流管理 -│ └── worker/ # Worker 框架核心 -├── data/ -│ ├── storage/chongming/ # SQLite 数据库文件 -│ └── uploads/ # 文件上传目录 -├── logs/ # 日志目录 -├── resources/ # 模块银行打包资源 -├── test_mqtt_e2e.py # MQTT 端到端测试 -├── publisher.py / subscriber.py # MQTT 收发示例 -└── README.md -``` +- Python >= 3.12、[uv](https://docs.astral.sh/uv/) +- Rust 工具链（编译 `jwt` / `password` 扩展） +- Node.js >= 22（前端开发） +- ffmpeg（可选，`stream_manager` 需要） ---- - -## 多仓库结构与开发工作流 - -本仓库（**chongming-core**）只跟踪网关核心代码，其余资产分布在独立仓库中： - -| 目录 / 包 | 独立仓库 | 说明 | -| ---------------- | ----------------------------------------------- | -------------------------- | -| `src/chongming/` | 本仓库（chongming-core） | 网关核心（FastAPI + Zenoh）| -| `utils//` | `chongming-utils-*`（9 个，见「工具包列表」） | 工具包，由 `uv sync` 按 git 源安装 | -| `utils/cli/` | `chongming-utils-cli`（规划中，暂以 path 源安装）| 命令脚本集合（build / build_plugin / gen_models / gen_plugin） | -| `src/plugins/` | `chongming-plugins` | Worker 插件源码 | -| `config/` | `chongming-config` | 应用与 Worker 路由配置 | -| `src/chongming-web/` | `chongming-web` | Vue 3 前端 | - -### 首次克隆（开发环境引导） +### 首次克隆 ```bash git clone http://124.71.68.6:3000/chongming/chongming-core.git chongming-zenoh @@ -106,75 +39,26 @@ cd chongming-zenoh git clone http://124.71.68.6:3000/chongming/chongming-plugins.git src/plugins git clone http://124.71.68.6:3000/chongming/chongming-config.git config git clone http://124.71.68.6:3000/chongming/chongming-web.git src/chongming-web -uv sync # utils/* 由 uv 按 pyproject 的 git 源自动安装，无需手工 clone +uv sync # 网关（chongming-gateway）与 utils/* 工具包按 pyproject 源自动安装 ``` -### 开发与发布检查清单 - -**修改 `utils/`（工具包）**： - -1. 在 `utils/` 内提交并 push 到对应远程（并发布新版本，见下） -2. 回主仓库执行 `uv lock --upgrade-package <包名>` 刷新版本 pin -3. `uv sync` 后运行相关测试 -4. 主仓库提交 `uv.lock` - -> ⚠️ **打包陷阱（"改了 utils/* 但产物行为没变"几乎都是它）**：`uv run build` 会先按 -> 锁文件同步环境，**`uv pip install -e utils/` 的可编辑安装会被替换回锁文件里的 -> 发布版**。因此本地改了工具包后，必须走上面 1-4 步（push + 发版 → 刷新 pin → sync）， -> 否则 `build/` 产物里仍是旧版代码。仅本地联调时可临时用可编辑安装，但打包前要确认 -> venv 指向的是准备发布的版本： -> -> ```bash -> uv pip install -e utils/worker --no-deps # 仅联调：让 venv 用本地源码 -> python -c "import chongming_worker, inspect; print(inspect.getsourcefile(chongming_worker))" -> ``` - -**修改 `src/plugins/` / `config/` / `src/chongming-web/`**： - -1. 在对应子仓库内提交并 push -2. 回主仓库启动验证（`uv run serve` + 相关测试） - -**发布前总检查**： - -- [ ] 所有子仓库 `git status` 干净且已 push -- [ ] `uv.lock` 与各 git 源最新提交一致 -- [ ] 打包前确认 venv 里 `chongming-worker` 等工具包为目标版本（改了 `utils/*` 却仍是旧版时， - 产物行为不会变，见上一节打包陷阱） -- [ ] 配置中无硬编码凭据——密钥一律走环境变量： - `VOLC_ACCESSKEY`（豆包 Ark Key）、`AIBOX_MQTT_URI`（AI Box 外部 Broker URI） -- [ ] `uv run python test_mqtt_e2e.py` 等测试通过 - ---- - -## 快速开始 - -### 环境要求 - -- Python >= 3.12 -- Rust 工具链（用于编译 `jwt` / `password` 扩展） -- Node.js >= 22（前端开发） -- [uv](https://docs.astral.sh/uv/) 包管理器 -- ffmpeg（可选，stream_manager 需要） - ### 安装与启动 -> 首次克隆：先按「多仓库结构与开发工作流」拉取 `src/plugins` / `config` / `src/chongming-web` 子仓库，再执行以下步骤。 - ```bash -# 同步依赖（自动编译 Rust 扩展 + 安装 Python 包） -uv sync - -# 启动服务（开发模式，热重载） -uv run serve +uv sync # 同步依赖（自动编译 Rust 扩展） +uv run serve # 启动服务（开发模式，热重载） ``` -服务启动后： +启动后： -| 通道 | 地址 | 说明 | -| ------- | ---------------------------- | ---------------- | -| HTTP | `http://localhost:8000` | REST API | -| Swagger | `http://localhost:8000/docs` | OpenAPI 交互文档 | -| MQTT | `mqtt://localhost:1883` | 内嵌 Broker | +| 通道 | 地址 | 说明 | +| ---- | ---- | ---- | +| HTTP | `http://localhost:8000` | REST API | +| Swagger | `http://localhost:8000/docs` | OpenAPI 交互文档（仅调试模式开启） | +| MQTT | `mqtt://localhost:1883` | 内嵌 Broker（`use_embedded_broker` 控制） | + +服务器可切换（`uv run gunicorn` / `uv run granian`，或设 `APP_SERVER` 环境变量）， +启动参数统一读 `config.toml` 的 `[.server]` 段。 ### 前端开发 @@ -184,234 +68,66 @@ npm install npm run dev # http://localhost:5173 ``` ---- - -## 架构与数据流 - -```mermaid -graph TB - HTTP[HTTP 客户端] -->|REST API| API[FastAPI 动态路由] - MQTT[MQTT 客户端] -->|chongming/+/+| BRIDGE[MQTT Bridge
纯客户端模式] - - REG[RouteRegistry
订阅 chongming/routes/**] -->|动态挂载/摘除 HTTP 路由| API - API --> WM[WorkerManager
随机负载均衡 / 实例定向 + 超时重试] - BRIDGE --> WM - - subgraph GW[网关进程 chongming] - API - REG - BRIDGE - WM - end - - WM -->|Zenoh Query| IW[inline Worker 副本
网关进程内加载] - WM -->|Zenoh Query| EW[external Worker 进程
独立容器，零网关依赖] - EW -.->|发布路由描述符
chongming/routes/**| REG - - IW --> DB[(SQLite)] - EW --> DB - EW -->|AI 调用| AI[豆包 API] -``` - -**启动流程**： - -``` -uv run serve → FastAPI lifespan: - 1. setup_storage() — 本地文件系统存储 - 2. setup_zenoh() — Zenoh peer 会话 - 3. setup_mqtt_broker() — 内嵌 amqtt Broker（子进程 + 文件锁单例； - `use_embedded_broker=false` 时跳过，改用外部 Broker） - 4. setup_worker_manager() — WorkerManager + 定期 liveliness 刷新 + 网关登记（register_gateway，参与全局分摊）+ 该插件副本数经 resolved_count 按 launch_worker_count 从 cache/toml 解析 - 5. load_workers() — 按插件 [workers].load_mode 分派: - inline → 进程内 import 插件 → 启动 Worker 协程 - external → 跳过（独立进程运行，盘扫条目仅观测） - 5.1 setup_route_registry() — 常驻订阅 chongming/routes/**，external 插件路由动态挂载/摘除 - 6. validate_contracts() — inline 插件三方契约校验，失败终止启动 - 7. register_routes() — inline 插件盘扫生成 HTTP 路由 - 8. setup_mqtt() — MQTT Bridge（inline 主题静态收集 + external 主题动态推送；基于共享缓存的分布式锁保证全局单实例） -``` - ---- - -## Worker 管理系统 - -通过 Zenoh 向 `worker///__admin__` 管理端点下发指令，**精细化控制单个 Worker 副本**（inline 与 external 进程插件共用同一管理协议）。提供 REST 管理端点 + Vue 前端页面，操作需超级管理员鉴权。 - -### 管理协议 - -每个用 Worker 框架（`chongming_worker.runtime.create_workers`）创建的副本都会注册一个管理 Queryable： +## 项目结构 ```text -worker///__admin__ +chongming-zenoh/ +├── config.toml # 应用主配置（环境、服务器、MQTT、缓存等） +├── pyproject.toml # 宿主项目依赖与脚本入口 +├── plugins_deps.toml # 内置插件专属依赖（打包收集用） +├── config/ # 配置子仓库（chongming-config） +│ ├── zenoh_config.json5 # Zenoh 会话配置 +│ └── workers/ # 插件注册 + 路由定义（TOML） +│ ├── /config.toml # 插件注册与运行时配置 +│ └── /.toml # 端点定义（每个端点一个文件） +├── src/ +│ ├── plugins/ # 内置插件源码（chongming-plugins 子仓库） +│ └── chongming-web/ # Vue 3 前端（chongming-web 子仓库） +├── utils/ # 独立工具包（各自独立仓库，uv 安装） +│ ├── gateway/ # 网关本体（chongming-gateway，可独立发包） +│ ├── cli/ # chongming-cli：build / gen_plugin / build_plugin ... +│ ├── worker/ # Worker 框架（chongming-worker） +│ └── ... # config / database / jwt / password / request / ... +├── workers/ # 进程插件本地开发目录（不入库；生产在独立容器） +├── public/ # 可执行文件打包骨架（main.py / server.py / 打包资源） +├── resources/ # 模块银行产物（plugins.mbank 等） +├── data/ logs/ # 数据库、上传文件、日志 +└── docs/ # 专题文档（见文末「更多文档」） ``` -| 指令 | 说明 | -| ----------- | ---------------------------------------------------------------------- | -| `status` | 查询副本状态 | -| `restart` | 重启该副本，副本槽位（index）由管理循环驱动，重新生成新 `worker_id`（`worker---`）并经 liveliness 更新 | -| `stop` | 停止该副本 | -| `drain` | **优雅排空**：摘除 liveliness/业务 Queryable 隔绝新请求，等待正在处理的在途请求完成后再退出，管理循环不重建（数量-1） | -| `rescale` | 按 `data.scale` 把**当前进程内**该类型副本数调整为指定值（`data:{"scale":N}`）；**两种口径**：不带 `instance_id` 时为「全局对称分摊」（写回全局 count + 各实例均分），带 `instance_id + absolute=true` 且命中本实例时为「逐实例精细扩缩」（只改该实例绝对副本数），target 非本实例则安全忽略 | -| `config_set`| 更新 `[workers.config]` 运行时配置（路由配置除外）；成功则写回配置并重启该类型全部对应副本，任一失败返回失败信息 | +## 使用指南 -### REST 管理端点（仅超管） +### 插件：两种运行模式 -| 方法 | 路径 | 说明 | -| ------ | --------------------------------------- | -------------------------------------- | -| `GET` | `/api/v1/admin/workers` | 盘点全部副本（inline 本地注册 + external liveliness 发现） | -| `POST` | `/api/v1/admin/workers/restart` | 重启单个副本（`worker_type` / `worker_id` 查询参数） | -| `POST` | `/api/v1/admin/workers//stop` | 停止 single 副本 / 某插件全部副本 | -| `GET` | `/api/v1/admin/workers//config` | 读取该插件的现有 `[workers.config]`（展示当前值） | -| `POST` | `/api/v1/admin/workers//config` | 更新配置（body `{updates:{k:v}}`），下发并重启 | -| `POST` | `/api/v1/admin/workers//scale` | 调整该类型副本数（body `{count:<全局总量>}`）：写全局 count → 清除逐实例定制回均分 → 广播全部存活实例，**全局对称分摊** -| `GET` | `/api/v1/admin/workers//instances` | 按外部进程实例（instance_id）分组读取该类型副本（供前端逐实例精细扩缩展示） | -| `POST` | `/api/v1/admin/workers//scale_instance` | **逐实例精细扩缩**（body `{instance_id, count}`）：只把指定实例的副本数调到 N，其余实例不动，持久化到 `instcounts` 覆盖表 | +按插件粒度在 `config/workers//config.toml` 的 `[workers]` 节声明： -> 旧版运行时未声明 `__admin__` 管理能力时，管理指令会超时（HTTP 502，见日志提示）。 - -### 配置热更新（路由配置除外） - -- worker 回写插件 `config.toml` 的 `[workers]` 段（保留其他段），**拒绝写保留键**（路由/加载控制键如 `type` / `module` / `count` 等）。 -- `Plugin.reload(new_config)` 在互斥锁下先 `on_cleanup()` 释放旧资源、再 `on_init()` 用新配置重建客户端（如 AI/DB client），使配置**立即生效、无需重启网关**。 -- 仅 index0 副本落盘避免并发写冲突，其余副本同样应用新配置；任一失败则返回失败明细且不触发重启方式的成功确认。 - -### 副本数量管理（扩缩容） - -同类 worker 的副本数可随时调整，并**持久化到配置**（重启网关后保持）： - -- **扩容**：新增副本槽位（`WorkerPool` 新增 `_managed_worker` 管理循环，新副本注册 liveliness 后可被负载均衡）。 -- **缩容**：随机挑选多余副本触发 **`drain`（优雅排空）**——先摘除 liveliness/业务 Queryable **隔绝新请求**，等待正在处理的**在途请求完成**后再关闭该副本（`in-flight` 计数实现），避免中断业务；管理循环不重建。 -- **目标数量持久化**：`rescale` 把 `[workers].count` 写回 `config.toml`，网关/进程重启后按新数量加载。 -- **统一协议**：inline 与 external 共用。网关对 inline 直接调整本进程 `WorkerPool`；对 external 经 Zenoh **向该类型全部存活实例广播 `rescale`**（不只是挑一个副本），由各实例按其 quota 扩缩并写回自身配置（无需重启整个进程/容器）。 -- **逐实例 vs 全局**：`/scale` 是对某个类型的**全局对称扩缩**；若外部进程有多个（多机/多容器），`/scale` 会把全局总量均分到每个存活实例。要单独调整**某一个**外部实例的副本数，见下文「外部进程实例粒度扩缩」。 - -### 全局副本数协调（多网关） - -多个网关共享同一份 `chongming-cache`（根 `config.toml` 的 `[cache]` 段）时，副本数「count」语义为**该类 worker 的全局目标总量**，由 chongming-cache 作为**单一事实来源**，各网关按在线数量自动分摊。这样只在任意一台网关执行扩缩容，即可全局生效。 - -- **单一事实来源**：`count` 存于 cache 键 `worker:cm:counts:`，在线网关 id 列表存 `worker:cm:gateways`，各网关分摊配额存 `worker:cm:quota::`（可观测）。 -- **网关登记/注销 + 心跳租约**：网关启动时 `register_gateway()` 加入在线列表，并每 15s `heartbeat_gateway()` 刷新自身时间戳；超过租约（60s）未心跳的死网关会被自动清理（crash/kill -9 也不残留，避免配额失真）。当前由 `CHONGMING_GATEWAY_ID` 环境变量指定（多实例需唯一，尤其**同宿主多网关必须显式指定**；缺省回退 `hostname`，稳定且重启不变，不残留）；关闭时 `unregister_gateway()` 移除，其余网关实时重算分摊。 -- **分摊算法**：`own_quota()` 将全局总量按在线网关 sorted(id) 均分，余量按序补前若干。`/admin/workers//scale`（body `{count:<全局总量>}`）写全局 count 后，各网关把自己的 session 副本调到自己的 quota。 -- **共享方式与后端**：单实例本机 `sqlite` 即可；**多实例必须共享同一 cache 后端**（同一共享卷的 sqlite 或 redis），否则各写各的 cache 无法收敛成全局量。 -- **配置优先来源模式**：根 `config.toml` 的 `[cache].launch_worker_count` 决定启动时 **`[workers.config]` 与 `count` 从哪一侧取值**（两种模式下"前端显示 / 插件实际使用 / 重启后行为"三处都自洽）： - - ```toml - [cache] - backend = "sqlite" # "sqlite"（默认，本机共享） | "redis"（跨主机） - path = "./cache_data/cache.db" - launch_worker_count = "toml" # "toml"（默认）=以文件为准；"cache"=以缓存为准 - ``` - - | 环节 | `"toml"`（默认） | `"cache"` | - | --- | --- | --- | - | 启动时 | 用 toml 的 `[workers.config]` 与 `count` **覆盖写入缓存**（前端读缓存，因此随之显示 toml 值） | 以缓存为准；插件初始化即读缓存值，缓存缺失才用 toml 引导并写回 | - | 插件实际使用 | toml 值 | 缓存值 | - | 管理接口改动的持久性 | 仅本次运行有效，重启网关被文件覆盖 | 跨重启保持 | - - 两种模式都不回写 `config.toml`：管理接口的改动只进缓存。`cache` 模式下启动日志会打印 toml 与缓存的差异键（以缓存为准）；如需让 toml 的改动生效，先删除 `worker:cm:configs:`（运行时配置）或 `worker:cm:counts:`（副本数）缓存键再重启，或改用 `"toml"` 模式。`GET /admin/workers//config` 的响应带 `source`（cache/toml）与 `mode`，前端据此标注展示值的来路。 - -- **对齐时机（两层，规则一致且幂等）**：inline 插件由**网关**在创建副本之前对齐——`WorkerManager.align_inline_config()` 在收到展平配置与 `[workers.config]` 原始段后按模式取缓存/写缓存，再交给插件的 `create_workers`；worker SDK 的 `Plugin.create_workers` 在 `_bootstrap()`/`on_init()` 之前用 `resolve_startup_config()` 再解析一次。网关侧这层是为了让网关与 worker 包版本解耦：worker 包仍是旧版时 cache 模式同样生效（否则旧版 `on_init` 只拿到 toml 展平配置，表现为"前端显示缓存值、实际跑 toml"）。**external 插件**没有网关这层兜底，由插件进程内的 SDK 解析，需要 worker 包为含该改动的版本。 - -- **进程插件（external）实例粒度参与分摊**：网关只对 **inline** 插件用 cache 解析启动副本数（见「启动流程」第 5 步 `load_workers`），external 插件的启动副本数由外部编排按自身配置决定；但**运行时的缩放**上，external 并不退化为“每进程全额”，而是同样在**共享 cache 上登记实例参与方**来分摊，避免多实例把副本翻倍： - - **实例登记 + 心跳租约**：每个 external 进程启动时向 `worker:cm:instances:` 登记自身参与方（`[instance_id, ts]`），并每 15s 心跳续约；超过租约（60s）未心跳的死实例会被清理。`divmod` 按存活参与方 N 均分全局 count（余量补靠前实例），`N==1` 时退化为全额（单部署零影响）。 - - **实例标识默认唯一**：`--instance` / 环境变量 `CHONGMING_WORKER_INSTANCE` 显式指定（跨重启稳定、逐实例精细扩缩可持续命中）；缺省回退 `{hostname}:{pid}`——**同一机器并行多个未指定 `--instance` 的进程也能靠 pid 区分**，不会因为 id 相同被并成一个参与方导致副本翻倍或前端只显示一个实例。注意缺省 id 随进程重启变化。 - - **广播 + 幂等收敛**：`/scale` 写全局 count 后向该类型**全部存活实例**广播 `rescale`，各实例读同一全局 count + 同一份参与方列表算出的 quota 一致，重复 rescale 也幂等，因此任意地收敛到同一预期状态。 - - **前提**：external 需共用共享 `[cache]`（多实例 redis 或共享卷 sqlite）；`cache is None` 时退化为全额（仅单实例安全）。 - -### 外部进程实例粒度扩缩（逐实例精细扩缩） - -当同一个类型跑多个外部进程（多机/多容器）时，除了 `scale` 的**全局对称扩缩**，还可以只对**某一个实例**精确扩缩： - -- **语义**：把目标实例的副本数**绝对值**设为 N，其它实例不动；与全局 `scale` 共存——全局对称是默认/一键复位，逐实例是定制覆盖。 -- **覆盖表**：每个实例的绝对目标存 cache 键 `worker:cm:instcounts:` = `{instance_id: 绝对副本数}`。有记录的实例按绝对数执行；没记录的实例回退到全局均分。一旦用 `scale_instance` 定制了某实例，之后 `scale`（全局）会**清除整张覆盖表**回到均分（显式“复位”），因此两者不会互相干扰。 -- **命中判断**：rescale 的下发必须命中**本实例**才执行（`instance_id` 与 `register_instance` 一致），目标是其它实例则安全忽略，避免误缩。 -- **REST / UI**：`POST /admin/workers//scale_instance`（body `{instance_id, count}`）下发；`GET /admin/workers//instances` 供前端按实例分组展示当前副本数。前端用「调整数量」（全局对称）和「逐实例精细扩缩」表格（实例 id / 模式 / 当前副本数 / 目标 + 应用）两个入口。 -- **协调前提**：与外部广播缩放一致，需共享 `[cache]`，且外部进程用稳定的 `--instance` / `CHONGMING_WORKER_INSTANCE`（缺省 `{hostname}:{pid}` 虽唯一但重启会变，跨重启持续命中请显式指定）。 - -### 实例定向调用（把请求固定到某个进程实例） - -默认情况下，同 type 的多个副本（含 inline 与 external 进程）逐请求**随机负载均衡**（`random.choice`），这是绝大多数无状态插件想要的。若某插件持有**只存在于某个进程本地**的状态（本地 SQLite/文件、串口或其他设备连接），需要请求固定打到那个进程上，可用**实例定向**： - -- **指定方式（三种入口，语义一致）**： - - HTTP：请求头 `X-Chongming-Instance: <实例id>` - - MQTT：信封载荷顶层加控制字段 `target_instance`（网关读后**摘除**，业务 side 看不到该字段） - - 程序内调用：`worker_manager.call(...)` / `call_raw(...)` / `call_stream(...)` 的 `instance_id=` 参数（`utils/request` 的独立 Client 不支持定向） -- **实例 id 取值**：worker 侧按 `--instance` → `CHONGMING_WORKER_INSTANCE` → `CHONGMING_GATEWAY_ID` → `{hostname}:{pid}` 的顺序解析（`current_instance_id()`）；网关侧定向寻址 inline 副本时用网关自身实例 id（`CHONGMING_GATEWAY_ID`，缺省 hostname）。两侧缺省值在容器环境（K8s 给网关进程注入 `CHONGMING_GATEWAY_ID`）下**自动一致**，无需额外配置；裸机同一台跑多个网关时请显式指定，否则默认实例 id 里的 pid 不同会导致对不上。人可读的取值列表见 `GET /api/v1/admin/workers//instances`。 -- **粒度**：定向到**进程实例**；若该实例内有多个副本，请求仍在这几个副本之间随机均衡（即“实例固定、副本仍均衡”）。 -- **实例离线时严格报错**：返回 HTTP 503 并附已识别实例列表，**不会静默回退到其它实例**——回退会让粘滞状态的插件跑在错误副本上，产生难查的数据错乱。MQTT 同理（响应主题收到 `{"error": ...}`）。 -- **重试不跨实例**：调用失败重试只在同实例内换副本；实例内已无可换副本时立即失败，不重复打同一个副本。流式端点本身不重试，实例离线即失败。 -- **缓存按实例隔离**：定向且 `cacheable=true` 时缓存键带实例前缀，避免取到其它实例的结果。 -- **身份来源与开销**：副本的实例身份取自其 admin status 回复的 `instance_id`（Python/Rust 两种运行时都会返回），网关**按需查询并缓存**——未使用定向功能时零额外开销，不定向的请求路径与行为完全不变；首次定向到某实例时会对该类型的新副本各发一次轻量 status 查询（每个副本只查一次，inline 副本无需查询）。副本重启后 id 变化，会重新查询一次。 -- **版本要求**：worker 端需运行带 `instance_id` 上报的运行时（当前 Python/Rust SDK 均满足）；读不出身份的副本会被排除在定向之外（不定向的调用不受影响）。 - -```bash -# 等价于把这次请求固定到实例 edge-01 的那个进程 -curl -H 'X-Chongming-Instance: edge-01' http://127.0.0.1:8000/api/v1/demo/state - -# MQTT：载荷顶层带 target_instance（网关转发前会摘掉它） -# topic: chongming/demo/state payload: {"meta":{}, "data":{}, "target_instance":"edge-01"} +```toml +load_mode = "inline" # 默认：插件在网关进程内加载 +load_mode = "external" # 独立进程运行，经 Zenoh 注册路由 ``` -### inline worker 宿主级领导选举（gunicorn 多 worker 兼容） +| 模式 | 源码位置 | 加载方式 | 适用 | +| ---- | -------- | -------- | ---- | +| **inline** | `src/plugins//` | 网关进程内 import，TOML 盘扫生成路由 | 与网关同生命周期的业务插件 | +| **external** | `workers//`（自包含项目） | 独立进程，经 Zenoh 注册路由 | 独立扩缩、独立依赖/语言、跨容器部署 | -inline worker 的生命周期交给**基于分布式锁的宿主级领导选举**（取代“每 gunicorn worker 各自拉起”），key `inline_owner:` 按网关隔离： +- external 插件与网关**零导入关系**：只依赖 `chongming-worker` SDK + zenoh，代码结构与 + inline 完全一致；本地调试 `cd workers/ && uv run chongming-worker --type `。 +- 两种模式可同时存在（混合部署），网关对二者的路由挂载、服务发现与运维指令一致。 -- **多 worker（同网关）**：同一时刻只有一个网关进程托管 inline worker（副本数精确等于 `count`，不再是 `worker 数 × count`）；其余进程只做 HTTP，经 Zenoh liveliness 发现并调用 inline worker。HTTP 层仍由所有 gunicorn worker 分担吞吐。 -- **多实例（不同网关）**：各网关按自己的 `gateway_id` 各自选举，互不抢占。 -- **持有者崩溃自愈**：非持有者进程经 `refresh_workers` 周期性巡检（`_maybe_heal_inline`）——若发现某 inline 类型既无线活副本又无本地副本，就经分布式锁抢得领导权并拉起这些 worker（防 gunicorn 复活死 slot 却因租约未过期而跳过的问题）。接管同样 `set_nx` 唯一胜出，无双持有。 -- 前提：同一共享 `[cache]`（见「全局副本数协调」）；单实例本机 `sqlite` 即可，多实例需共享后端。 +开发指南：[内置插件开发](src/plugins/README.md) ｜ +[进程插件开发（external，从开发到部署）](docs/process-plugin.md) ｜ +[Rust 插件](docs/rust-plugin.md) -### 多 worker 的边界（转发 vs 处理） +### 端点定义与路由 -在 gunicorn 多 worker 模式下，结论是：**多 worker 提升的是网关"HTTP 接入/转发"能力，不提升"worker 处理"能力**——网关角色是代理，真正的 RPC 处理在 worker 端。 +| 节类型 | HTTP | OpenAPI | MQTT | 模型命名 | +| ------ | ---- | ------- | ---- | -------- | +| `[GET]` | ✅ | ✅ | ✅ | `XxxGetRequest / XxxGetResponse` | +| `[POST]` | ✅ | ✅ | ✅ | `XxxPostRequest / XxxPostResponse` | +| `[MQTT]` | ❌ | ❌ | ✅ | `XxxMqttRequest` | -| 瓶颈所在 | 多 gunicorn worker 是否有效 | 正确扩法 | -| -------- | --------------------------- | -------- | -| HTTP 接入/转发（高 QPS 转发、并发慢客户端、鉴权/路由） | ✅ 有效（上限≈核数） | 加 gunicorn `-w` | -| inline 插件实际计算 | ⚠️ 有限（多 worker 只是更快把请求塞给同一个 inline worker，inline 仍单点） | 加 worker 副本数 `count` / 改跑 external 进程 | -| external 进程插件处理 | ✅ 有效（多 worker 可并行转发给独立进程） | 加 gunicorn `-w` + 对等扩 external | - -### 前端管理页 - -`Worker 管理`（`/workers`，深色主题）： - -- **登录**：内置网关端点 `POST /auth/login`（auth 域拆分后登录/签发不依赖 auth 插件，插件缩容或故障时登录与 admin API 仍可用），领取管理 Token（无需手填 JWT），从 JWT 解析当前用户与超管标识。 -- **鉴权**：仅 `is_superuser=true` 可管理；权限不足时前端/后端弹出对应提示（403）。 -- **列表分页**：展示副本（插件类型 / 副本 ID / 模式 inline|external / 状态），内置分页。 -- **查询与筛选**：列表顶部统一筛选栏——关键字搜索（插件类型 / 副本 ID / 实例 ID）、模式筛选（全部 / inline / external）、副本数 ≥ N 过滤；**同时作用于下方「逐实例精细扩缩」分组表**（副本数过滤仅对分组表生效），筛选条件变化自动回到第一页。 -- **操作**：单副本重启、批量重启（选中）、插件级重启全部副本、更新插件配置（弹窗展示现有配置值，点行带入编辑；支持对象/数组等嵌套值的**结构化 JSON 编辑与校验/格式化**）、**调整副本数量**（当前数 + `el-input-number` 设目标 + 应用，可扩缩容）。 -- **逐实例精细扩缩**：按外部进程实例（instance_id）**分组**展示（实例 id / 模式 / 当前副本数），每行可设目标副本数并「应用」——仅调整该实例，其余实例不动；全局「调整数量」后分组表也会自动刷新。 -- **自动刷新**：页面每 10s 轮询总副本列表**与**逐实例分组表，扩缩容/重启/停止等异步结果无需手动刷新即可反映。 -- **auth 核心插件**：配置不可更改，不在列表中展示（`Worker` 列表自动过滤 `worker_type == "auth"`）。 -- 调小副本数会**触发优雅排空**并在前端确认提示中说明。 - -### external（进程）插件同样可管理 - -external 副本走同一 `__admin__` Queryable，路由发现、指令下发与扩缩容（`rescale`）与 inline 一致。**前提**： - -1. external 容器运行的是含管理能力（`__admin__` + `reload` + `rescale`）的新版 `chongming-worker` 框架； -2. `config_set` / `rescale` 由副本在**其进程 cwd** 下写 `config/workers//config.toml`，external 容器需挂载与网关同款配置目录才能写回网关侧配置（含 `[workers].count` 持久化）； -3. 纯自定义语言（如 Rust `rsdemo`）进程插件需自行实现 `__admin__` Queryable + liveliness + 数量管理，否则只能发现、无法改写配置或扩缩容。 - ---- - -## 端点类型与路由注册 - -### 端点类型 - -TOML 文件中的节类型决定端点暴露方式： - -| 节类型 | HTTP 路由 | OpenAPI 文档 | MQTT | 请求模型命名 | -| -------- | --------- | ------------ | ---- | ------------------------------------- | -| `[GET]` | ✅ | ✅ | ✅ | `XxxGetRequest / XxxGetResponse` | -| `[POST]` | ✅ | ✅ | ✅ | `XxxPostRequest / XxxPostResponse` | -| `[MQTT]` | ❌ | ❌ | ✅ | `XxxMqttRequest`（无 response_model） | - -### 两种路由注册方式 - -#### 方式一：TOML 动态路由（标准 JSON 端点） - -通过 `config/workers/<插件>/*.toml` 定义端点，自动生成 FastAPI 路由。适用于参数和响应可被 Pydantic 模型描述的标准接口。 +**方式一：TOML 动态路由**（标准 JSON 端点）——`config/workers//*.toml`： ```toml # config/workers/example/add.toml @@ -423,9 +139,8 @@ params = ["a: float", "b: float"] result = { type = "float", required = true } ``` -#### 方式二：插件自定义路由（特殊端点） - -当端点无法用 TOML 描述时（如 `multipart/form-data` 上传），插件暴露 `APIRouter` 实例，网关自动发现并注册。 +**方式二：插件自定义路由**（无法用 TOML 描述时，如 multipart 上传）——插件暴露 +`APIRouter`，网关自动发现： ```python # src/plugins//routes.py @@ -438,465 +153,283 @@ async def upload(file: UploadFile = File(...)): ... ``` -> **约定**：插件 `__init__.py` 中导入 `router` 实例（`from .routes import router`），网关通过 `getattr(module, "router", None)` 自动发现。 +> 约定：插件 `__init__.py` 导入该 `router` 实例，网关通过 `getattr(module, "router", None)` 发现。 ### 流式端点（SSE） -基于 Zenoh Query/Reply 原生"多回复"实现流式输出：Worker 的 `@rpc_method` 返回生成器时， -框架逐块 `query.reply()` 推送（见 `utils/worker` README「流式端点」），网关以 SSE -（`text/event-stream`）逐块转发给 HTTP 客户端。**无额外端口、无额外进程、无新依赖**； -未声明 `stream` 的端点行为与之前完全一致。 - -#### TOML 配置 - -在方法节中声明 `stream = true`（示例 `config/workers/example/stream_chat.toml`）： +方法节声明 `stream = true`，`@rpc_method(..., stream=True)` 返回生成器即可逐块输出： ```toml [POST] summary = "模拟 AI 流式对话（SSE，逐 token 输出）" -auth_required = false stream = true -timeout = 300.0 # 流总寿命上限（秒，zenoh query deadline 硬上限） -stream_idle_timeout = 10.0 # 网关侧连续无数据块的最大间隔（秒） +timeout = 300.0 # 流总寿命上限（秒）；流式端点必须配大 +stream_idle_timeout = 10.0 # 连续无数据块的最大间隔（秒） params = ["prompt: str", "tokens: int"] ``` -| 字段 | 默认值 | 说明 | -| --- | --- | --- | -| `stream` | `false` | 流式端点：以 SSE（`text/event-stream`）输出 | -| `timeout` | `3.0` | 流总寿命上限（秒）。**流式端点必须配大**（如 300s），到期后 zenoh 丢弃后续回复 | -| `stream_idle_timeout` | `10.0` | 空闲超时（秒）：超过该间隔无新数据块则网关终止流 | - -> `timeout`（总寿命硬上限）与 `stream_idle_timeout`（空闲超时）是两回事，流式场景两个都要配。 - -#### 请求与响应 - -- 请求与普通端点一致（GET 走 query/path 参数，POST/PUT/PATCH 走 body 模型），鉴权依赖同样生效 -- 响应为 `text/event-stream`，每块一个 `data: {json}\n\n` 事件（SSE 标准分帧） -- 响应头：`Cache-Control: no-cache`、`X-Accel-Buffering: no`（配合反代关闭缓冲） -- OpenAPI 不声明流式端点的响应 schema（逐块 JSON dict，无法用单一模型描述） - ```bash curl -N -X POST localhost:8000/api/v1/calc/stream_chat \ -H 'Content-Type: application/json' -d '{"prompt":"你好","tokens":3}' # data: {"token":"你好·1","index":0} # data: {"token":"你好·2","index":1} -# data: {"token":"你好·3","index":2} # data: {"done":true} ``` -#### 错误语义 +- 请求与普通端点一致（含鉴权）；响应 `text/event-stream`，每块一个 `data: {json}` 事件。 +- 错误语义：无可用 worker 时 HTTP 503；流中异常/空闲超时以 `data: {"error": "..."}` 收尾； + 客户端断开即终止迭代。 +- 经 nginx 反代时需 `proxy_buffering off;`，否则失去逐块即时性。 +- 流式能力由 `chongming-worker` 提供，需 venv 中为支持 `stream` 的版本 + （本地改框架后按「[仓库结构与开发工作流](#仓库结构与开发工作流)」刷新 pin）。 -| 场景 | 表现 | -| --- | --- | -| 无可用 worker（首个数据块前） | HTTP 503（与普通端点一致） | -| Worker 中途异常 / 流式端点初始化失败 | SSE error 事件 `data: {"error":"..."}` 收尾（HTTP 200 已提交，无法再改状态码） | -| 空闲超时 | SSE error 事件 `data: {"error":"流式空闲超时 ..."}` 收尾 | -| 客户端断开 | 网关终止迭代，Worker 侧 reply 失败自动停流，不空转 | +### Worker 管理与扩缩容 -#### 部署注意 +网关提供 REST 管理端点与 Vue 管理页（需超级管理员），inline 与 external 插件共用同一套 +管理协议（Zenoh `worker///__admin__`）。 -生产环境经 nginx 反代时需关闭缓冲，否则 SSE 会被整段缓冲、失去逐块即时性： +| 方法 | 路径 | 说明 | +| ---- | ---- | ---- | +| `GET` | `/api/v1/admin/workers` | 盘点全部副本（inline 本地注册 + external liveliness 发现） | +| `POST` | `/api/v1/admin/workers/restart` | 重启单个副本（`worker_type` / `worker_id`） | +| `POST` | `/api/v1/admin/workers//stop` | 停止单个副本 / 某插件全部副本 | +| `GET` `POST` | `/api/v1/admin/workers//config` | 读取 / 更新 `[workers.config]`（成功即下发并重启副本） | +| `POST` | `/api/v1/admin/workers//scale` | 调整该类型副本数（全局对称分摊） | +| `GET` | `/api/v1/admin/workers//instances` | 按进程实例分组查看副本（供逐实例扩缩） | +| `POST` | `/api/v1/admin/workers//scale_instance` | 逐实例精细扩缩（只调指定实例） | -```nginx -location /api/v1/ { - proxy_buffering off; +**要点**： + +- 扩缩容持久化到插件配置；缩容走**优雅排空**（先摘除服务发现隔绝新请求，等在途请求完成后退出）。 +- 多网关共享同一 `[cache]` 时，副本数按在线网关自动分摊，任一台操作即全局生效（需 `redis` + 或共享卷 `sqlite`；`CHONGMING_GATEWAY_ID` 用于区分同宿主多网关）。 +- 无状态插件默认逐请求随机负载均衡；**需要固定到某个进程实例**的插件（本地状态/设备连接） + 可用实例定向调用：HTTP 头 `X-Chongming-Instance: <实例id>`，或 MQTT 信封顶层 + `target_instance`；实例离线时严格报错，不回退。 + +```bash +curl -H 'X-Chongming-Instance: edge-01' http://127.0.0.1:8000/api/v1/demo/state +``` + +细节（管理指令语义、多实例分摊与自愈、领导选举、逐实例扩缩）见 +[Worker 管理与多实例运维](docs/worker-management.md)。 + +### MQTT 通道 + +| 方向 | Topic | 说明 | +| ---- | ----- | ---- | +| 请求 | `chongming//` | MQTT → Worker | +| 响应 | `chongming///response` | Worker → MQTT（无响应时跳过） | + +```toml +[development.mqtt] +enabled = true +host = "127.0.0.1" # 内嵌 Broker 时为监听地址；外部 Broker 时为对方地址 +port = 1883 +use_embedded_broker = true # true（默认）网关内嵌 amqtt Broker；false 连接外部 Broker +# username = "chongming" # 可选：Broker 认证（内嵌/外部均生效） +# password = "change-me" +``` + +请求负载为**信封结构**（控制字段与业务数据分离），非信封格式会被丢弃并告警： + +```json +{ + "meta": { "method": "GET", "timeout": 30, "cacheable": false }, + "data": { "a": 3, "b": 5 } } ``` -#### 本地联调前注意 - -流式能力在 `chongming-worker` 框架（`utils/worker/`，独立仓库）。**本地未 push 的框架改动 -不会自动生效**——venv 里安装的是 `uv sync` 从 git 源拉取的旧版副本，旧版 `@rpc_method` -不认 `stream` 参数，会报 `TypeError: rpc_method() got an unexpected keyword argument 'stream'` -并连锁导致契约校验失败（启动终止）。本地联调时先以可编辑方式装入 venv： - -```bash -uv pip install -e utils/worker -``` - -正式发布仍走「开发与发布检查清单」流程（工具包 push → `uv lock --upgrade-package -chongming-worker` → `uv sync`，可编辑安装会被锁文件版本替换）。 - -#### 测试 - -``` -uv run python test_stream_worker.py # Worker 框架流式回复单测 -uv run python test_stream_gateway.py # 网关 call_stream 单测 -uv run python test_stream_e2e_zenoh.py # 真实 Zenoh 会话端到端（peer 模式，无需 router） -uv run python test_stream_route.py # 路由层 SSE 输出与兼容性回归（非流式端点不变） -``` - ---- - -## 进程插件与混合部署 - -两种插件运行模式**按插件粒度共存**，在 `config/workers//config.toml` 的 `[workers]` 节声明： - -```toml -load_mode = "inline" # 默认：插件在网关进程内加载 -load_mode = "external" # 独立进程运行，经 Zenoh 注册路由 -``` - -**inline 插件**（`src/plugins/`，chongming-plugins 仓库）： - -- 网关进程内 import；路由由网关盘扫 TOML 生成；契约校验在网关启动时执行 - -**external 进程插件**（本地开发目录 `workers/`，自包含项目，不入库）： - -- 与网关**零导入关系**：只依赖 `chongming-worker` SDK + zenoh，仅经 Zenoh 通讯 - - 路由注册：worker 启动时 put 描述符到 `chongming/routes//...`，退出时 delete；每 10s 周期重发布自愈 - - 服务发现：liveliness `worker//`（与 inline 副本同一机制，网关无需改动） -- worker 进程自读 TOML 并在**自己进程内**做契约校验（`chongming_worker.contracts`）； - 网关 RouteRegistry **常驻订阅**描述符后动态挂载/摘除 HTTP 路由（type 级 2s 防抖）。 - 路由注册与网关侧插件盘扫（`config/workers/*/config.toml`）解耦：**网关容器无需 - 放置 external 插件的配置文件**，盘扫发现的 external 条目仅用于启动日志观测； - inline 插件仍按盘扫在网关进程内加载，不受影响 -- 每个插件是**自包含项目**（`workers//` = 打包 zip 根 = 容器工作目录， - 开发即产物），代码结构与 inline 插件完全一致（Plugin 基类 + router/ + models/）， - 示例见本地 `workers/demo/`；新项目用 `uv run gen_plugin --process` 生成骨架 -- 运行（在插件项目根 `workers//` 下）： - - ```bash - cd workers/demo - uv run chongming-worker --type demo # 本地调试（首次先 uv sync 装依赖） - ./start.sh # 与容器一致的启动方式（uv sync + 启动） - ``` - -- 生产部署：进程插件与网关**分别打包到不同容器** - - 网关容器：`chongming` + 依赖 - - 进程插件容器：插件代码 + `chongming-worker` + 依赖 -- 隔离性由 `test_worker_isolation.py` 保证（阻断 chongming 导入 + AST 检查 import 语句） - -### 进程插件打包 - -`build_plugin` 把插件项目根直接压缩为可部署 zip（开发目录即产物， -插件代码 + 配置 + 依赖清单 + 容器入口都在 `workers//` 内）： - -```bash -uv run build_plugin demo # → dist/demo.zip -uv run build_plugin demo --output /tmp/pkg -``` - -产物布局（= 项目根，zip 根即容器工作目录）： - -```text -workers// 插件代码（模块 workers.） -config/workers// 插件注册 + 端点配置 -config/zenoh_config.json5 Zenoh 会话配置 -config.toml 最小根配置（仅 [zenoh] 节，worker 启动必需） -pyproject.toml uv 依赖清单（chongming-worker 内网 git 源） -start.sh 容器入口：uv sync + chongming-worker --type -``` - -- 打包自动排除 `__pycache__` / `*.pyc` / `.venv` 等，`start.sh` 权限强制 0o755 -- 插件扩展依赖（可选）：直接在项目根 `pyproject.toml` 的 dependencies 中追加 -- 容器内部署：`unzip .zip -d /app && cd /app && ./start.sh`（镜像需预装 uv） -- 打包前校验：`workers//` 项目根、项目内 `config/workers//config.toml` - （含 `[workers]` 节）、`config/zenoh_config.json5` 与根 `config.toml`（含 `[zenoh]`）必须存在， - `load_mode` 非 `external` 时告警 -- 插件配置随 zip 部署在**插件容器**（worker 自读）；网关侧无需放置（RouteRegistry 常驻订阅 Zenoh 路由描述符，与盘扫解耦） - ---- - -## MQTT 协议 - -### Topic 约定 - -| 方向 | Topic | 说明 | -| ---- | -------------------------------- | ------------------------------ | -| 请求 | `chongming//` | MQTT → Worker | -| 响应 | `chongming///response` | Worker → MQTT（`None` 时跳过） | - -### Broker 来源（内嵌 vs 外部） - -网关默认启动内嵌 amqtt Broker（子进程 + **宿主级分布式排他锁**单例），`[.mqtt].use_embedded_broker` 控制 Broker 来源： - -```toml -[development.mqtt] -enabled = true -host = "0.0.0.0" # true 时为内嵌 Broker 监听地址；false 时为外部 Broker 地址 -port = 1883 -use_embedded_broker = true # true（默认）网关启动 amqtt Broker；false 连接外部 Broker -``` - -- `use_embedded_broker = true`（默认）：网关启动内嵌 amqtt Broker 子进程，MQTT Bridge 作为客户端连回自身。 -- `use_embedded_broker = false`：网关**不启动** Broker，MQTT Bridge 直接连接 `host:port` 指向的外部 MQTT Broker（如独立部署的 EMQX/Mosquitto）。`host` 需改为外部 Broker 地址，`enabled = false` 时整个 MQTT 通道关闭。 - -**单例与生命周期控制**（取代旧文件锁）： - -- 内嵌 Broker 用 `chongming-cache` 的 `DistributedLock`（key `mqtt_broker:`，**按宿主隔离**——内嵌 Broker 绑定本机端口），多 worker/多实例下同一宿主机只启动一个 Broker；`cache` 未配置时跳过并告警。 -- 持锁的网关在缓存登记其 Broker 子进程 PID（`mqtt_broker_owner:`）。前任持有者崩溃/被强杀而留下**孤儿 Broker 子进程**时，接管的网关据此 PID 识别并终止之，再启动**由自己托管**的新子进程（可停、可接管）；否则端口被不可托管进程占用时仅复用、不托管。 -- 关停时仅真正托管子进程的网关终止进程、清理登记并释放锁，绝不误伤他人。 - -### Broker 认证（用户名/密码） - -`[.mqtt].username` / `password` 配置 Broker 认证（可选，默认匿名）： - -```toml -[development.mqtt] -enabled = true -host = "0.0.0.0" -port = 1883 -username = "chongming" # 可选；内嵌/外部 Broker 均生效 -password = "change-me" -``` - -- **内嵌 Broker**（`use_embedded_broker=true`）：配置后启用认证（amqtt FileAuthPlugin + argon2 哈希），所有客户端（含网关 Bridge）必须携带该账号密码连接；未配置则保持匿名。 -- **外部 Broker**：网关 Bridge 以该账号密码连接，broker 侧需自行创建对应账号。 -- 客户端连接示例：`mqtt://chongming:change-me@localhost:1883/`（密码含特殊字符时需百分号编码，如 `@` → `%40`；网关 Bridge 内部自动处理）。 -- 注意：密码明文存放于配置，生产环境建议使用外部 Broker + 环境变量注入。 - -### 自定义订阅主题 - -除内置的 `chongming//` 接口外，可将外部 MQTT 主题桥接到任意 Worker。**推荐在插件自己的配置** `config/workers//config.toml` 中通过 `[[workers.mqtt_topics]]` 声明（目标 worker 即该插件本身）： +**自定义订阅主题**：把外部 MQTT 主题桥接到插件，推荐写在插件自己的配置里： ```toml # config/workers/aibox_sophon_adapter/config.toml [[workers.mqtt_topics]] -topic = "box/aibox_sophon/board_ping" # 支持 +（单级）/ #（多级，须在末尾）通配符 -subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应通配符捕获值 -response_topic = "/board_ping/response" # 可选，响应主题；默认 "<实际主题>/response" +topic = "box/aibox_sophon/board_ping" # 支持 + / # 通配符 +subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应捕获值 +response_topic = "/board_ping/response" # 可选，默认 "<实际主题>/response" ``` -- `subject` 缺省为 `<插件类型>.`；`response_topic` 缺省为 `<实际匹配主题>/response`。 -- 网关发现插件时自动收集所有 `[[workers.mqtt_topics]]`，MQTT Bridge 一并订阅并路由。 -- 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。 -- 全局 `[.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。 +- 多条自定义主题按顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。 +- 全局 `[.mqtt].custom_topics` 可作通用兜底（不推荐，优先用插件配置）。 -### MQTT Bridge 单例（多实例免重复响应） +**多实例部署要点**（细节见 [Worker 管理与多实例运维](docs/worker-management.md)）： -MQTT Bridge 是**网关进程内**的 MQTT 客户端组件，负责订阅主题并把请求经 `WorkerManager` 转发给 Worker、回发 `/response`。为了让「同一条请求只被消费一次、只回一份响应」，Bridge 的启动采用**基于共享缓存的分布式锁**（`chongming-cache` 提供，`DistributedLock`），而不再使用本机文件锁。 +- 单机：内嵌 Broker + 本地 `sqlite` 缓存即可。 +- 多实例：**外部 Broker（必须）** + 共享 `[cache]` 后端（`redis` 或共享卷 `sqlite`）。 + 共享缓存是收敛前提：MQTT Bridge 全局单例、副本数分摊、inline 领导选举都跑在它上面。 -- **抢锁机制**：启动时 `set_nx`（SET NX + TTL 租约）原子抢锁，key 为 `mqtt_bridge`；抢到锁的网关才启动 Bridge，其余网关跳过。 -- **租约续约**：持锁网关后台定期 `touch` 续约；若持锁网关宕机/失联，租约到期后锁自动释放，其他网关可重新抢到（重启后生效）。 -- **受保护释放**：关闭时仅当 token 仍匹配（仍是自己持有）才删除，绝不误删其他网关抢到的锁。 -- **持有者崩溃自愈接管**：仅靠 gunicorn 复活死掉 slot 无法 failover（新进程可能在旧锁租约未过期时抢锁失败而照样跳过）。各网关进程内有**周期性巡检**（`mqtt_bridge_supervisor`，周期经 `[.mqtt].bridge_supervise_interval` 配置，默认 15s）：非持有者进程若发现锁已空闲（原持有者租约过期=已死/失联），即经 `set_nx` 抢得唯一持有权并**自动重建 Bridge**；接管时同步按需拉起/接管内嵌 Broker。分布式锁保证任意时刻只有一方胜出，无分裂。 - -```toml -[cache] -backend = "redis" # 跨主机收敛必须用 redis（或共享卷 sqlite） -``` - -- **关键条件**：跨机器收敛依赖**共享缓存后端**（与「全局副本数协调」同一 `[cache]`，单一事实来源）。仅 `redis`（或共享卷的 `sqlite`）后端时，不同物理机上的网关才会看到同一把锁，真正全局单 Bridge。 -- **`use_embedded_broker = true`（各机内嵌独立 Broker）**：各机 Broker 互不相连、本身不产生重复响应；此时全局单 Bridge 只会连到持锁网关那台机器的 Broker。 -- **`use_embedded_broker = false`（共享外部 Broker）**：原先 N 台网关的 N 个 Bridge 会共同订阅同一批主题，导致一条请求被各消费一次、各回一份响应；改用分布式锁后全局只有一个 Bridge 消费，彻底规避重复响应。 - -**部署结论**（内嵌 Broker 单例同样成立，`mqtt_broker:` 按宿主隔离）： - -| 部署形态 | MQTT Broker | `[cache]` 后端 | 状态 | -| -------- | ----------- | ------------- | ---- | -| 单机（含多 worker） | 内嵌 amqtt | 本地 `sqlite` | ✅ 完全够用 | -| 多实例分布式 | **外部 Broker**（必须，内嵌各机不聚合、接管会断设备） | **redis**（或共享卷 `sqlite`） | ✅ 推荐 | - -> 关键：多实例是否收敛，**MQTT 只是其中一环**，真正的收敛前提是**所有网关共享同一 `[cache]` 后端**。Bridge 全局锁、inline 领导选举锁、在线网关清单/quota 全部工作在同一共享 cache 上；若各实例用各自本机 cache，`mqtt_bridge` 锁跨不了主机，每台都能抢到 Bridge，又会退化成 N 个 Bridge 各连各的 Broker → 单请求多响应，问题回归。 -> -> 即：**单机 = 内嵌 Broker + 本地 sqlite；多实例 = 外部 Broker（必须）+ redis/共享卷 sqlite（选一）**。 - -### 请求负载格式（信封协议） - -MQTT 请求负载为信封结构，控制字段与业务数据分离： - -```json -{ - "meta": { - "method": "GET", // 可选，默认 GET；MQTT 专用端点需显式 "MQTT" - "timeout": 30, // 可选，RPC 超时（秒），默认 30 - "cacheable": false // 可选，是否允许结果缓存，默认 false - }, - "data": { "a": 3, "b": 5 } // 业务参数，传给 Worker -} -``` - -- 非信封格式的负载会被网关丢弃并告警（严格协议，设备侧请按此格式发布）。 -- HTTP 通道由网关自动组装信封（注入 method / user_info），无需调用方关心。 - -### 快速测试 +手动联调： ```bash -# 终端 1：启动服务 -uv run serve - -# 终端 2：端到端测试 -python test_mqtt_e2e.py - -# 手动测试 -python subscriber.py # 监听所有响应 -python publisher.py example add '{"a":3,"b":5}' # 调用 add +uv run serve # 终端 1：启动服务 +uv run python test/test_mqtt_e2e.py # 终端 2：端到端 +uv run python test/publisher.py example add '{"a":3,"b":5}' # 单次调用 +uv run python test/subscriber.py # 监听所有响应 ``` ---- +### 配置 -## 开发指南 +- **应用配置**：根 `config.toml`（环境 `[default].env` → `[development]` / `[production]` + 选段；服务器、安全、MQTT、缓存等）。 +- **插件注册**：`config/workers//config.toml` 的 `[workers]` 节——加载模式、 + `count` 副本数、`router_prefix`、`[workers.config]` 运行时配置。 +- **路由定义**：`config/workers//*.toml`，每个端点一个文件。 +- **Zenoh**：`config/zenoh_config.json5`。 -### 添加新 Worker 插件 +字段速查（完整说明见 [配置说明](config/README.md)）： -```text -1. 创建 src/plugins// — __init__.py（Plugin 子类）, global_var.py, router/, models/ -2. 创建 config/workers//config.toml，在 [workers] 节注册插件 - （type / module / factory / cleanup / count + [workers.config] 运行时配置） -3. 创建 config/workers//*.toml 端点定义 -4. 运行 uv run gen_models 生成模型 -5. 实现 router/.py，引用 models/pydantic.py 中的模型 -6. 更新 router/__init__.py 导入 +| 字段 | 位置 | 说明 | +| ---- | ---- | ---- | +| `count` | `<插件>/config.toml` | Worker 副本数（`[workers]` 节） | +| `load_mode` | `<插件>/config.toml` | `inline`（默认）/ `external` | +| `router_prefix` | `<插件>/config.toml` | API 路由前缀 | +| `defaults.timeout` / `defaults.cacheable` | `<插件>/config.toml` | 端点默认超时与是否缓存 | +| `params` | `<操作>.toml` | 参数列表（`"name: type"`） | +| `auth_required` | `<操作>.toml` | 是否需要 JWT 鉴权 | +| `internal` | `<操作>.toml` | 仅限内部 RPC，不注册 HTTP 路由 | +| `cacheable` | `<操作>.toml` | 端点是否可缓存 | +| `stream` / `stream_idle_timeout` | `<操作>.toml` | 流式输出（SSE）与其空闲超时 | +| `launch_worker_count` | 根 `config.toml` `[cache]` | 副本数/运行时配置以 `toml`（默认）还是 `cache` 为准 | + +## 打包与部署 + +### 构建可执行文件 + +```bash +uv run build pyinstaller # PyInstaller 打包 +uv run build nuitka # Nuitka 打包（Linux 需宿主 patchelf，已在 dev 依赖） + +uv run build --encrypt --license # 对外分发：加密模块（PyArmor 混淆）+ 注入授权公钥 +uv run build --only mbank # 只重新生成 plugins.mbank（插件更新只需替换该文件） +uv run build --only mbank,prepare,vue # 只跑指定步骤 ``` -> 推荐用脚手架一步生成骨架：`uv run gen_plugin [operation ...]`。 -> 启动时网关会做**契约校验**：TOML 路由缺少对应 `@rpc_method` 将导致启动失败， -> 模型缺失/孤儿等不一致以警告日志提示。 +构建流程：模块打包（`plugins.mbank`）→ 资源复制 → PyInstaller/Nuitka 打包 → 数据库初始化 +→ 前端编译 → VFS 挂载。`--only` 可选步骤：`mbank` / `prepare` / `obfuscate` / `pack` / +`clean` / `license_guard` / `vue`。 -### 为现有插件添加新端点 +**网关更新与插件更新的路径不同**： -```text -1. 创建 config/workers//.toml -2. 运行 uv run gen_models -3. 创建 router/.py → 引用生成的模型 -4. 更新 router/__init__.py -5. 重启服务 +- 网关本体（`chongming-gateway` 包）是普通依赖，由 PyInstaller/Nuitka 从 venv 收集进 exe + → 改网关必须重新 `uv run build`； +- 插件代码在 `plugins.mbank` 里 → `uv run build --only mbank` 后替换该文件即可热更新。 + +### 打包产物的依赖收集 + +exe 内的依赖来自三处，新增第三方包时对号入座： + +| 依赖归属 | 声明位置 | +| -------- | -------- | +| 网关本体（`fastapi` / `granian` / `chongming_*` 等） | `chongming-gateway` 包自己的 `pyproject.toml`（消费者自动获得，打包沿静态导入收集） | +| 动态导入、静态分析看不到的（granian 子模块、amqtt 插件、`sqlite_vfs.contrib`、`uvloop`）+ 骨架自身 | 根 `config.toml` 的 `[pyinstaller_build_import]` | +| 内置插件专属（如 doubao 的 `openai` / `bytedance`） | 根 `plugins_deps.toml`，按插件分节 | + +新增插件依赖时，venv 安装清单（根 `pyproject.toml`）与 `plugins_deps.toml` 需同步维护； +依赖变更后必须重跑 `pack`（`--only mbank` 带不动第三方依赖）。 + +`--encrypt` / `--license` 默认**都关**（不加参数即内部调试产物：明文模块、不校验授权）； +对外分发必须同时加这两个开关。 + +### 进程插件打包与容器部署 + +```bash +uv run build_plugin demo # → dist/demo.zip（开发目录即产物） +uv run migrate_plugin # 内置插件一键迁移为进程插件 ``` -流式端点（SSE）在此基础上额外两步：TOML 方法节声明 `stream = true` 并配大 `timeout`； -`@rpc_method(..., stream=True)` 函数返回生成器（yield 逐块），详见上文「流式端点（SSE）」。 +zip 根即容器工作目录：插件代码 + `config/` + `pyproject.toml` + `start.sh` +（`uv sync` + `chongming-worker --type `）；容器内 `unzip .zip -d /app && cd /app && ./start.sh`。 +生产部署时进程插件与网关**分别打包到不同容器**：网关容器装 `chongming-gateway`， +插件容器装插件代码 + `chongming-worker`。 + +## 仓库结构与开发工作流 + +本仓库（**chongming-core**）是宿主项目：骨架（`public/`）、打包配置、构建与交付； +业务资产分布在独立仓库中，通过 `uv` 依赖与子目录 clone 组合： + +| 目录 / 包 | 仓库 | 说明 | +| --------- | ---- | ---- | +| `utils/gateway/` | `chongming-gateway` | 网关本体（独立包，发布到私有源；本仓按发布版依赖） | +| `utils/template/` | 项目模板（独立维护） | 新项目模板：`python utils/template/init.py <目录>` 初始化出「骨架 + 插件 + 配置」的网关项目 | +| `utils//` | `chongming-utils-*` | 工具包（config / worker / jwt / cache / ...） | +| `utils/cli/` | `chongming-utils-cli` | `chongming-cli`：build / gen_plugin / build_plugin / migrate_plugin | +| `src/plugins/` | `chongming-plugins` | 内置插件源码 | +| `config/` | `chongming-config` | 应用与 Worker 路由配置 | +| `src/chongming-web/` | `chongming-web` | Vue 3 前端 | + +**修改 `utils/` / `utils/gateway`（可发布包）**： + +1. 在包目录内提交并 push，`uv build` 后用 twine 发布新版本到私有源 +2. 回主仓库 `uv lock --upgrade-package <包名>` 刷新 pin，`uv sync` +3. 主仓库提交 `uv.lock` + +> ⚠️ **打包陷阱**：`uv run build` 会先按锁文件同步环境，本地 `uv pip install -e utils/` +> 的可编辑安装会被替换回发布版——「改了包但产物行为没变」几乎都是它。打包前确认 +> venv 指向目标版本：`python -c "import chongming_worker, inspect; print(inspect.getsourcefile(chongming_worker))"`。 + +**修改 `src/plugins/` / `config/` / `src/chongming-web/`**：在对应子仓库内提交并 push， +回主仓库启动验证。 + +**发布前总检查**： + +- [ ] 所有子仓库 `git status` 干净且已 push；`uv.lock` 与各源一致 +- [ ] 打包前确认 venv 内网关与工具包为目标版本（见上方打包陷阱） +- [ ] 配置中无硬编码凭据——密钥走环境变量： + `VOLC_ACCESSKEY`（豆包 Ark Key）、`AIBOX_MQTT_URI`（AI Box 外部 Broker URI） +- [ ] 相关测试通过（见 `test/`，如 `uv run python test/test_mqtt_e2e.py`） + +## 内置内容一览 + +### 内置插件（chongming-plugins 仓库，inline） + +| 插件 | HTTP 前缀 | MQTT 主题前缀 | 说明 | +| ---- | --------- | ------------- | ---- | +| **auth** | `/auth` | `auth.*` | 用户管理：注册/资料/改密（登录与 token 签发内置网关） | +| **auth_oauth** | `/auth/oauth` | `auth_oauth.*` | 身份源示例：GitHub OAuth + 手机验证码登录 | +| **doubao** | `/doubao` | `doubao.*` | 豆包视觉理解（支持 AICC 安全通道） | +| **aibox_sophon_adapter** | `/aibox` | `aibox_sophon_adapter.*` | AI Box 设备心跳、离线检测、告警、视频上传 | +| **example** | `/calc` | `example.*` | 四则运算 + 心跳演示（含 MQTT 端点示例） | +| **demo**（进程插件示例） | — | — | `workers/demo/`（本地自包含项目，external 模式） | + +### 工具包（uv 依赖，各自独立仓库） + +| 包 | 路径 | 语言 | 说明 | +| -- | ---- | ---- | ---- | +| **chongming-gateway** | `utils/gateway/` | Python | 网关本体（FastAPI + Zenoh） | +| **chongming-auth** | `utils/auth/` | Python | auth 域共享契约：Users 模型、建库/种子、登录校验与 JWT 签发 | +| **chongming-config** | `utils/config/` | Python | TOML 配置加载（TypedDict 类型安全） | +| **chongming-database** | `utils/database/` | Python | aiosqlite + SQLModel 异步数据库封装 | +| **chongming-jwt** | `utils/jwt/` | Rust + PyO3 | JWT 编解码（HS/RS 系列） | +| **chongming-password** | `utils/password/` | Rust + PyO3 | Argon2id 密码哈希与验证 | +| **chongming-request** | `utils/request/` | Python | Zenoh RPC 直连客户端 | +| **chongming-worker** | `utils/worker/` | Python | Worker 框架：装饰器、运行时、契约校验、独立进程入口 | +| **chongming-stream-manager** | `utils/stream_manager/` | Python | ffmpeg + MediaMTX RTSP 流代理管理 | +| **bytedance** | `utils/bytedance/` | Python | 字节跳动 AICC 安全通信 SDK | ### 常用命令 ```bash -uv run serve # 启动开发服务 -uv run gen_plugin # 生成统一风格插件骨架 -uv run gen_models # 生成 Pydantic 模型 -uv run build # 构建打包（PyInstaller/Nuitka） -uv run chongming-worker --type # 独立 Worker 进程（项目根下本地调试，load_mode=external） -chongming-worker --type # 独立 Worker 进程（容器入口） -uv pip install -e utils/ # 安装/更新工具包 +uv run serve # 启动开发服务 +uv run granian # 或 uv run gunicorn（换服务器，参数仍读 config.toml） +uv run gen_plugin # 生成插件骨架（--process 生成进程插件项目） +uv run gen_models # 按 TOML 生成 Pydantic 模型 +uv run build [pyinstaller|nuitka] # 打包可执行文件 +uv run build_plugin # 打包进程插件 → dist/.zip +uv run migrate_plugin # 内置插件迁移为进程插件 +uv run chongming-worker --type # 独立 Worker 进程（external 插件本地调试/容器入口） +cd utils/gateway && uv build # 构建网关包 wheel（发版用） +python utils/template/init.py ../new-proj # 用模板初始化一个新项目 ``` ---- - -## 插件列表 - -| 插件 | 路径 | 功能 | -| ------------------------ | ----------------------------------- | ------------------------------------------------------- | -| **auth** | `src/plugins/auth/` | 用户管理插件：注册/资料/改密（登录与 token 签发已内置网关，Users 模型/种子/校验/签发共享于 `chongming-auth` 包） | -| **auth_oauth** | `src/plugins/auth_oauth/` | 身份源插件示例：GitHub OAuth + 手机验证码登录，基于 `chongming-auth` 契约签发平台 token | -| **doubao** | `src/plugins/doubao/` | 豆包视觉理解，支持 AICC 安全通信通道 | -| **aibox_sophon_adapter** | `src/plugins/aibox_sophon_adapter/` | AI Box 设备心跳监控、离线检测、告警上报与查询、视频上传 | -| **example** | `src/plugins/example/` | 四则运算 + 心跳演示（含 MQTT 端点示例） | -| **demo**（进程插件示例） | `workers/demo/`（本地自包含项目，不入库） | 独立进程运行（load_mode=external），经 Zenoh 注册路由 | - ---- - -## 工具包列表 - -| 包名 | 路径 | 语言 | 说明 | -| ---------------------------- | ----------------------- | ----------- | ---------------------------------------------- | -| **chongming-auth** | `utils/auth/` | Python | auth 域共享契约：Users 模型、建库/种子、登录校验与 JWT 签发（网关内置登录与 auth 插件共用） | -| **chongming-config** | `utils/config/` | Python | TOML 配置加载，TypedDict 类型安全 | -| **chongming-database** | `utils/database/` | Python | aiosqlite + SQLModel 异步数据库封装 | -| **chongming-jwt** | `utils/jwt/` | Rust + PyO3 | JWT 编解码，支持 HS256/384/512 + RS256/384/512 | -| **chongming-password** | `utils/password/` | Rust + PyO3 | Argon2id 密码哈希与验证 | -| **chongming-request** | `utils/request/` | Python | Zenoh RPC 直连客户端 | -| **chongming-worker** | `utils/worker/` | Python | Worker 框架核心：装饰器、运行时、契约校验、独立进程入口（`chongming-worker`） | -| **chongming-stream-manager** | `utils/stream_manager/` | Python | ffmpeg + MediaMTX RTSP 流代理管理 | -| **bytedance** | `utils/bytedance/` | Python | 字节跳动 AICC 安全通信 SDK（Jeddak 等） | - ---- - -## 配置说明 - -- **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置 -- **插件加载模式**：按插件粒度声明在 `config/workers//config.toml` 的 `[workers]` 节 - - `load_mode = "inline"`（默认）：插件在网关进程内加载 - - `load_mode = "external"`：插件以独立进程运行（进程编排交给 Docker/K8s， - 进程入口 `uv run chongming-worker --type `），经 Zenoh 在 `chongming/routes/**` - 注册路由，网关动态挂载 HTTP 端点（RouteRegistry 常驻订阅，无需在网关侧保留插件配置） - - 两种模式可同时存在（混合部署） -- **进程插件与网关完全隔离**：进程插件只依赖 `chongming-worker` SDK + zenoh， - 与网关（`chongming` 包）零导入关系，仅通过 Zenoh（`chongming/routes/**` 注册路由、 - `worker//` 服务发现）通讯。`workers/` 目录仅为开发便利， - **生产部署时进程插件与网关分别打包到不同容器**： - - 网关容器：`chongming`（含 `src/chongming/`）+ 依赖 - - 进程插件容器：插件代码（`workers//`）+ `chongming-worker` + 依赖， - 入口 `chongming-worker --type `）， - 打包: `uv run build_plugin ` → `dist/.zip`（见「进程插件打包」） -- **Zenoh 配置**：`config/zenoh_config.json5` -- **插件注册**：`config/workers//config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容） -- **路由定义**：`config/workers//*.toml`，每个端点一个文件 - -详细配置说明见 [`config/README.md`](config/README.md)。 - ---- - -## 构建部署 - -```bash -# PyInstaller 打包 -uv run build pyinstaller - -# Nuitka 打包 -uv run build nuitka - -# 仅重新打包 .mbank 模块文件（第三方依赖不变时，更新只需替换 mbank 文件） -uv run build --only mbank - -# 只执行指定步骤，跳过其余流程 -uv run build --only mbank,prepare,vue -``` - -构建过程包括：模块加密打包 → 资源复制 → PyInstaller/Nuitka 打包 → 数据库初始化 → 前端编译 → VFS 挂载。 - -`--only ` 可选步骤：`mbank`（模块打包）、`prepare`（准备构建目录）、 -`obfuscate`（PyArmor 混淆，仅 pyinstaller 模式）、`pack`（PyInstaller/Nuitka 打包）、 -`clean`（清理临时文件）、`vue`（Vue 前端构建）；可逗号分隔或重复传多个。 - -详见 [插件开发指南](src/plugins/README.md)。 - -### 常用命令 - -```bash -uv run serve # 启动开发服务器 -uv run gen_plugin # 生成统一风格插件骨架（--process 生成进程插件项目） -uv run gen_models # 生成 Pydantic 模型 -uv run build pyinstaller # PyInstaller 打包 -uv run build nuitka # Nuitka 打包 -uv run build_plugin # 进程插件打包 → dist/.zip -chongming-worker --type # 独立 Worker 进程（进程插件容器入口） -uv pip install -e utils/ # 安装/更新工具包子包 -``` - -### TOML 配置字段速查 - -| 字段 | 位置 | 说明 | -| -------------------- | --------------------- | --------------------------------------- | -| `count` | `<插件>/config.toml` | Worker 副本数（`[workers]` 节） | -| `launch_worker_count`| `根 config.toml` | 配置优先来源模式：`"toml"`=以文件为准（启动用 toml 的 `[workers.config]`/count 覆盖 cache，默认）；`"cache"`=以 cache 为单一事实来源（插件初始化即读缓存，跨重启保持）（`[cache]` 节） | -| `router_prefix` | `<插件>/config.toml` | API 路由前缀 | -| `defaults.timeout` | `<插件>/config.toml` | 默认超时（秒） | -| `defaults.cacheable` | `<插件>/config.toml` | 是否缓存 RPC 结果 | -| `params` | `<操作>.toml` | 参数列表（`"name: type"` 格式） | -| `auth_required` | `<操作>.toml` | 是否需要 JWT 鉴权 | -| `internal` | `<操作>.toml` | `true` 则仅限内部 RPC，不注册 HTTP 路由 | -| `cacheable` | `<操作>.toml` | 端点是否可缓存 | -| `summary` | `<操作>.toml` | 端点描述（用于 OpenAPI / 日志） | -| `stream` | `<操作>.toml` | `true` 则流式输出（SSE），需同时配大 `timeout` | -| `stream_idle_timeout`| `<操作>.toml` | 流式空闲超时（秒，默认 10），超时无新数据块则终止流 | - ---- - -## 内置插件 - -| 插件 | HTTP 前缀 | MQTT 主题前缀 | 说明 | -| ------------------------ | --------- | ------------------------ | ------------------------------------ | -| **auth** | `/auth` | `auth.*` | 注册、个人信息、改密（登录 `POST /auth/login` 内置网关） | -| **doubao** | `/doubao` | `doubao.*` | 火山引擎豆包视觉理解模型 | -| **example** | `/calc` | `example.*` | 四则运算 + 访问计数 + MQTT 心跳示例 | -| **aibox_sophon_adapter** | `/aibox` | `aibox_sophon_adapter.*` | AI Box 设备心跳适配 + 视频上传 | - ---- - ## 更多文档 - [配置说明](config/README.md) -- [插件开发指南](src/plugins/README.md) -- [进程插件开发指南（external 模式，从开发到部署）](docs/process-plugin.md) -- [前端开发指南](src/chongming-web/README.md) +- [内置插件开发指南](src/plugins/README.md) +- [进程插件开发指南（external，从开发到部署）](docs/process-plugin.md) +- [Worker 管理与多实例运维](docs/worker-management.md) +- [Rust 插件编写说明](docs/rust-plugin.md) +- [插件线协议规范（语言无关）](docs/plugin-wire-protocol.md) +- [网关包说明（chongming-gateway）](utils/gateway/README.md) - [Worker 框架](utils/worker/README.md) +- [前端开发指南](src/chongming-web/README.md) diff --git a/deploy/README.md b/deploy/README.md index 5f4f5d1..8230d1d 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -81,7 +81,7 @@ sudo docker compose down # 停止并移除容器（保留数 sudo docker compose down -v # 停止并移除容器 + 数据卷（慎用，会清空数据） ``` -### 验证 +### 健康检查 ```bash sudo docker exec chongming-redis redis-cli ping # 期望 PONG @@ -94,7 +94,8 @@ sudo docker exec chongming-mosquitto mosquitto_sub -p 1883 -t '$SYS/broker/versi **Zenoh Router**：项目 [zenoh_config.json5](../config/zenoh_config.json5) 默认以 `peer` 模式连接 `tcp/127.0.0.1:7447`。本 router 容器把 7447 映射到宿主机，同机部署时应用即可直接连到该 router；若跨机部署，需把 `connect.endpoints` 改为 router 机器地址。 -**Mosquitto**：项目 [config.toml](../config.toml) 默认 `use_embedded_broker = true`（内嵌 amqtt）。要切到本外部 broker，需改为： +**Mosquitto**：网关默认可内嵌 amqtt Broker（`use_embedded_broker = true`）。要改连本目录的 +外部 broker，把 `config.toml` 的 `[.mqtt]` 改为： ```toml [development.mqtt] diff --git a/docs/multi-language-plugin-plan.md b/docs/multi-language-plugin-plan.md index 38777aa..948e146 100644 --- a/docs/multi-language-plugin-plan.md +++ b/docs/multi-language-plugin-plan.md @@ -1,11 +1,13 @@ # 多语言插件演进统一规划（Rust 优先） +> **本文是演进规划记录，不是使用指南**。要开发/使用插件请先看： +> [进程插件开发指南（Python）](./process-plugin.md)、 +> [Rust 插件编写说明](./rust-plugin.md)、 +> [进程插件线协议规范（语言无关）](./plugin-wire-protocol.md)。 + > 背景：团队技术栈杂糅，Python 在部分计算密集场景性能有限。 > 目标：在不破坏现有体系的前提下，开放进程插件多语言实现能力， > **Rust 优先落地**，为后续其他语言（Go / C++ 等）铺路。 -> -> 配套文档：[进程插件线协议规范（语言无关）](./plugin-wire-protocol.md)、 -> [进程插件开发指南（Python）](./process-plugin.md) --- @@ -60,7 +62,7 @@ chongming-worker-rs/ # 独立仓库（或 utils/worker-rs/，与 Pyth │ # 另有 [workers.config] 运行时读写（保留键校验 + 原子落盘） ├── descriptor.rs # 描述符结构体（规范 §2.2 字段表）+ 端点 TOML 解析 ├── publisher.rs # RoutePublisher 等价：PUT / DELETE / 10s 重发布 - ├── serve.rs # worker_serve 等价：liveliness token + Queryable + 生命周期 + ├── serve.rs # 入口等价：chongming-worker（Python 侧）：liveliness token + Queryable + 生命周期 ├── runtime.rs # 托管副本模型：ReplicaController（index/epoch/restart/stop/drain） │ # + WorkerPool（按 target 动态扩缩，优雅排空） ├── admin.rs # __admin__ Queryable：status/restart/stop/drain/get_config/ @@ -159,7 +161,7 @@ Rust 侧用 sqlx 在 `utils/worker-rs` 内提供等价封装（feature `database SQL 占位符遵循各自方言（SQLite `?`、PostgreSQL `$1`），SDK 不做 SQL 重写。 - **TLS**：默认不启用 TLS 后端；PostgreSQL 走 SSL 时需追加 `tls-rustls-ring` 等 feature。 - **参考实现**：`workers/rsdemo` 的 `db_ping` 端点（写 + 读一表做连通性探测）；`gen_plugin --rust` 模板已内置 - 注释版 `[workers.config.database]` 段与 `database` feature（见 [rsdemo README](../../workers/rsdemo/README.md)）。 + 注释版 `[workers.config.database]` 段与 `database` feature（见 [rsdemo README](../workers/rsdemo/README.md)）。 --- @@ -167,7 +169,7 @@ Rust 侧用 sqlx 在 `utils/worker-rs` 内提供等价封装（feature `database ### 4.1 build_plugin 支持非 Python 插件 -- 现状：`build_plugin` 产物为 pyproject.toml + start.sh（`uv run worker_serve --type `），纯 Python 链路 +- 现状：`build_plugin` 产物为 pyproject.toml + start.sh（`uv run chongming-worker --type `），纯 Python 链路 - 目标：支持声明插件语言/入口，允许打包任意可执行产物 + 自定义 start.sh （Rust 插件 = 编译产物 + `--type` 参数约定），`--type` 语义跨语言保持一致 - 约束：打包结构（zip 根 = 容器工作目录）不变 diff --git a/docs/plugin-wire-protocol.md b/docs/plugin-wire-protocol.md index c1a830a..a14a0fd 100644 --- a/docs/plugin-wire-protocol.md +++ b/docs/plugin-wire-protocol.md @@ -237,11 +237,11 @@ worker/// | 协议项 | 唯一实现（文件:行号） | | ---- | ---- | | 描述符 keyexpr / route_id 编码 / 发布摘除 | `utils/worker/src/chongming_worker/routes_publisher.py:23-89` | -| 描述符字段定义与消费 | `src/chongming/app/core/route_generator.py:82-100`（RouteConfig） | +| 描述符字段定义与消费 | `utils/gateway/src/chongming/app/core/route_generator.py:82-100`（RouteConfig） | | 描述符生成约定（TOML → dict） | `utils/worker/src/chongming_worker/route_config.py:25-103` | -| 网关订阅 / 防抖 / 幂等 / 冲突 / 筛选 | `src/chongming/app/core/route_registry.py:44-261` | -| liveliness 发现 / 负载均衡 / 超时重试 / 缓存 | `src/chongming/app/core/worker_manager.py:73-379` | -| 请求信封构造（meta.method / user_info） | `src/chongming/app/core/route_generator.py:241-405` | +| 网关订阅 / 防抖 / 幂等 / 冲突 / 筛选 | `utils/gateway/src/chongming/app/core/route_registry.py:44-261` | +| liveliness 发现 / 负载均衡 / 超时重试 / 缓存 | `utils/gateway/src/chongming/app/core/worker_manager.py:73-379` | +| 请求信封构造（meta.method / user_info） | `utils/gateway/src/chongming/app/core/route_generator.py:241-405` | | 信封解析 / 分派 / 三种响应形态 | `utils/worker/src/chongming_worker/runtime.py:27-220` | | 进程生命周期（10s 重发布 / 优雅退出） | `utils/worker/src/chongming_worker/process.py:35-197` | | 配置读取（json5_path / 插件配置） | `utils/worker/src/chongming_worker/process.py:39-136` | diff --git a/docs/process-plugin.md b/docs/process-plugin.md index 5789d60..8468042 100644 --- a/docs/process-plugin.md +++ b/docs/process-plugin.md @@ -27,9 +27,7 @@ --- -## 2. 运行原理 - -### 2.1 架构 +## 2. 工作机制（简述） ```mermaid graph TB @@ -37,7 +35,7 @@ graph TB REG[网关 RouteRegistry
订阅 chongming/routes/**] -->|动态挂载/摘除| API API --> WM[WorkerManager
负载均衡 + 超时重试] - subgraph GW[网关容器 chongming] + subgraph GW[网关容器] API REG WM @@ -54,36 +52,18 @@ graph TB WS -.->|SIGTERM 时 delete 摘除| REG ``` -### 2.2 Worker 进程启动流程（`chongming_worker.process.worker_serve`） +`chongming-worker --type demo` 启动后的流程： -```text -chongming-worker --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 → 取消副本任务 → 关闭会话 -``` +1. 读根 `config.toml` 的 `[zenoh]` 节打开会话，自读 + `config/workers/demo/config.toml` 的 `[workers]` 节（含 `[workers.config]` 运行时配置） +2. import 插件模块触发 `@rpc_method` 注册 → **进程内契约校验**（不通过则退出码 1） +3. `create_workers` 创建副本（默认 `count` 个，`--count` 可覆盖），注册 liveliness +4. 经 Zenoh 把路由描述符 put 到 `chongming/routes//`，网关动态挂载 HTTP 路由； + 每 10s 重发布（网关后启动或样本丢失时自愈） +5. 收到 SIGTERM/SIGINT → delete 描述符（路由摘除）→ 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 据此做负载均衡，网关侧无需任何改动。 +要点：网关容器**无需**放置 external 插件的配置文件（RouteRegistry 常驻订阅，与盘扫解耦）； +描述符字段与 keyexpr 细节见 [线协议规范](plugin-wire-protocol.md)。 --- @@ -114,8 +94,7 @@ workers// # 项目根 = zip 根 = 容器工作目录 └── start.sh # 容器入口（uv sync + uv run chongming-worker --type ） ``` -> 参考示例：`workers/demo/`（完整项目，含 5 个端点，配合 `test_worker_isolation.py` / -> `test_process_mode_e2e.py` 使用）。 +> 参考示例：`workers/demo/`（完整项目，含 5 个端点）。 --- @@ -291,16 +270,10 @@ uv run --project .. chongming-worker --type demo > 若配置了 `load_mode` 非 `"external"`，会输出告警：网关也会在进程内加载该插件， > 外部进程将与网关内副本**重复服务**。 -### 第 7 步：验证 +### 第 7 步：联调验证 -```bash -# 1. 进程隔离性（阻断 chongming 导入 + AST 检查 import 语句；自动切到 demo 项目根） -uv run python test_worker_isolation.py - -# 2. 端到端（需先启动网关 uv run serve）：从 demo 项目根拉起 worker 进程， -# 路由动态注册 → 跨进程 RPC → 退出摘除(404) → 重启恢复 -uv run python test_process_mode_e2e.py -``` +启动网关（`uv run serve`）后调用插件路由即可验证端到端：网关日志可见路由动态挂载， +worker 退出后路由自动摘除（请求变 404），重启 worker 后恢复。 ### 第 8 步：打包 @@ -442,11 +415,8 @@ API 一览（`chongming_worker.storage.StorageClient`）： - `zenoh` / `pydantic` 等第三方库 - 插件自身的扩展依赖（项目根 `pyproject.toml` 中声明） -由 `test_worker_isolation.py` 双重保证： - -1. **运行时阻断**：`sys.modules["chongming"] = None` 后走完整链路 - （SDK 导入 → 插件配置自读 → 插件 import → 契约校验），任何网关依赖立即 ImportError -2. **静态检查**：AST 解析 SDK 各模块源码，确认无 `chongming` / `chongming.*` import 语句 +隔离性由框架保证：SDK 导入链路中一旦出现 `chongming` / `chongming.*` 的 import， +运行时立即 `ImportError`（启动即暴露），无需等到部署阶段。 --- @@ -476,7 +446,7 @@ Worker 进程启动时自读 TOML 并**在自己进程内**做三方一致性校 | 路由一直 404 | worker 是否已退出（优雅退出会摘除路由）；进程被 SIGKILL 时描述符残留，重启 worker 后重发布自愈覆盖 | | 找不到插件配置 | `chongming-worker` 须在**插件项目根** `workers//` 下运行（读 `config/workers//config.toml`，`--config` 可指定全局配置路径） | | storage.* 端点报「storage 未配置」 | 插件 config.toml 未配置 `[workers.config.storage]`（或 `enabled = false` / 连接失败降级）。补齐配置后重启 worker；启动日志可见「对象存储已就绪」或「storage 连接失败」告警 | -| 容器内 import `workers.demo` 失败 | 确认在容器工作目录（zip 根）启动，`start.sh` 内含 `cd "$(dirname "$0")"`，worker_serve 会把 CWD 加入 sys.path | +| 容器内 import `workers.demo` 失败 | 确认在容器工作目录（zip 根）启动，`start.sh` 内含 `cd "$(dirname "$0")"`，启动入口会把 CWD 加入 sys.path | --- @@ -489,8 +459,6 @@ uv run chongming-worker --type # 本地调试独立 Worker ./start.sh # 与容器一致的启动方式（uv sync + 启动） cd .. && uv run build_plugin # 进程插件打包 → dist/.zip uv run gen_models workers//config/workers/ workers//workers/ # 生成模型 -uv run python test_worker_isolation.py # 进程隔离性测试 -uv run python test_process_mode_e2e.py # 端到端测试（先启动网关） ``` --- diff --git a/docs/rust-plugin.md b/docs/rust-plugin.md index 4d9e279..7641dcf 100644 --- a/docs/rust-plugin.md +++ b/docs/rust-plugin.md @@ -281,28 +281,28 @@ unzip myplugin.zip -d /app && cd /app && ./start.sh --- -## 9. 本地验证 +## 9. 本地联调 ```bash -# Zenoh Router（替代 zenohd） +# 起一个本地 Zenoh Router（替代 zenohd） .venv/bin/python utils/worker-rs/scripts/zenoh_router.py & -# 协议级 15 项断言（描述符/RPC/流式/错误/摘除/自愈） +# 协议级端到端自检（描述符 / RPC / 流式 / 错误 / 摘除 / 自愈） .venv/bin/python utils/worker-rs/scripts/verify_e2e.py \ --type myplugin --bin workers/myplugin/target/debug/myplugin ``` --- -## 10. 与 Python SDK 的差异（有意为之） +## 10. Rust 侧行为要点 -| 差异 | 说明 | +| 要点 | 说明 | | ---- | ---- | -| 契约校验 | Python 进程内三方校验（硬失败）；Rust 为**发布前自检**：无 handler 端点告警跳过，不发布残缺路由 | -| 模型生成 | Rust 用 gen_models 生成强类型结构体 + **启动时指纹校验**（TOML 变更未重生成即告警，比 Python 更强） | +| 契约校验 | **发布前自检**：无 handler 的端点告警跳过，不发布残缺路由 | +| 模型生成 | `gen_models` 生成强类型结构体 + **启动时指纹校验**（TOML 变更未重生成即告警） | | handler 注入 | worker_id 经 `Request.worker_id`、user_info 经 `Request.user_info` 注入（非函数参数） | -| 配置缓存后端 | Rust 支持 sqlite（`local-cache`，本机与 Python 共享同一 .db）+ Redis（`redis-cache`，跨主机）；缺省无缓存回退 | -| 数据库 | Rust 走 sqlx（见 §7），配置键与 Python `chongming_database` 对齐 | +| 配置缓存后端 | `local-cache`（sqlite，与 Python 共享同一 .db）/ `redis-cache`（跨主机）；缺省无缓存回退 | +| 数据库 | 走 sqlx（见 §7），配置键与 `chongming_database` 对齐 | --- diff --git a/docs/worker-management.md b/docs/worker-management.md new file mode 100644 index 0000000..f624295 --- /dev/null +++ b/docs/worker-management.md @@ -0,0 +1,163 @@ +# Worker 管理与多实例运维 + +网关（chongming-gateway）对 Worker 副本的运维能力：管理指令与 REST 端点、扩缩容语义、 +实例定向调用，以及多网关/多实例部署时的收敛前提。inline 与 external（进程）插件共用同一套 +管理协议。 + +> 使用入口速览见主 [README](../README.md#worker-管理与扩缩容)。 + +## 1. 管理协议 + +每个用 Worker 框架创建的副本都会注册管理 Queryable： + +```text +worker///__admin__ +``` + +| 指令 | 说明 | +| ---- | ---- | +| `status` | 查询副本状态（含 `instance_id`，实例定向据此寻址） | +| `restart` | 重启该副本：副本槽位由管理循环驱动，重新生成新 `worker_id` 并更新 liveliness | +| `stop` | 停止该副本 | +| `drain` | **优雅排空**：摘除 liveliness/业务 Queryable 隔绝新请求，等待在途请求完成后退出，管理循环不重建 | +| `rescale` | 调整副本数：不带 `instance_id` 为「全局对称分摊」；带 `instance_id + absolute=true` 且命中本实例时为「逐实例精细扩缩」，target 非本实例则安全忽略 | +| `config_set` | 更新 `[workers.config]` 运行时配置（路由配置除外）；成功则写回配置并重启该类型副本，任一失败返回失败明细 | + +> 旧版运行时未声明 `__admin__` 管理能力时，管理指令会超时（HTTP 502，日志有提示）。 + +## 2. REST 管理端点（仅超管） + +| 方法 | 路径 | 说明 | +| ---- | ---- | ---- | +| `GET` | `/api/v1/admin/workers` | 盘点全部副本（inline 本地注册 + external liveliness 发现） | +| `POST` | `/api/v1/admin/workers/restart` | 重启单个副本（`worker_type` / `worker_id` 查询参数） | +| `POST` | `/api/v1/admin/workers//stop` | 停止单个副本 / 某插件全部副本 | +| `GET` | `/api/v1/admin/workers//config` | 读取现有 `[workers.config]`（响应含 `source`/`mode` 标注取值来路） | +| `POST` | `/api/v1/admin/workers//config` | 更新配置（body `{updates:{k:v}}`），下发并重启 | +| `POST` | `/api/v1/admin/workers//scale` | 调整副本数（body `{count:<全局总量>}`），全局对称分摊 | +| `GET` | `/api/v1/admin/workers//instances` | 按进程实例（instance_id）分组读取副本 | +| `POST` | `/api/v1/admin/workers//scale_instance` | 逐实例精细扩缩（body `{instance_id, count}`） | + +## 3. 配置热更新 + +- 回写插件 `config.toml` 的 `[workers]` 段（保留其他段），拒绝写保留键（`type` / `module` / + `count` 等路由与加载控制键）。 +- 插件按新配置先 `on_cleanup()` 释放旧资源、再 `on_init()` 重建客户端（AI/DB client 等）， + 配置**无需重启网关即生效**。 +- 仅 index0 副本落盘避免并发写冲突；其余副本同样应用新配置。 + +## 4. 扩缩容语义 + +- **扩容**：新增副本槽位并注册 liveliness，随即参与负载均衡。 +- **缩容**：多余副本走 `drain` 优雅排空——先隔绝新请求，等在途请求完成后关闭，避免中断业务。 +- **持久化**：目标数量写回插件 `config.toml` 的 `[workers].count`，重启后按新数量加载。 +- **全局对称 vs 逐实例**：`/scale` 调整某类型的**全局总量**（多实例下由各实例均分）； + 需单独调整**某一个**外部实例时用 `/scale_instance`。 + +## 5. 多实例收敛（副本数分摊 / 领导选举 / MQTT Bridge 单例） + +多网关或多进程部署时，以下三件事都以**共享 `[cache]` 为单一事实来源**： + +```toml +[cache] +backend = "redis" # 多实例必须共享（redis 或共享卷 sqlite）；单机 sqlite 即可 +launch_worker_count = "toml" # 副本数/运行时配置以 toml（默认）还是 cache 为准 +``` + +**前提**：所有网关与 external 进程指向同一 cache 后端。若各实例各用本机 cache， +副本数无法收敛、锁跨不了主机，多实例行为会退化。 + +### 5.1 副本数（count）按在线实例分摊 + +- `count` 语义为**该类 Worker 的全局目标总量**，各参与者按在线数量自动分摊。 +- 在线清单与心跳：网关启动登记、每 15s 心跳续约，超 60s 未续约的实例自动清理；关闭时注销。 + 同宿主多网关需显式设置 `CHONGMING_GATEWAY_ID`（缺省回退 hostname）。 +- external 进程实例同样登记参与分摊，避免多实例把副本数翻倍；实例标识用 + `--instance` / `CHONGMING_WORKER_INSTANCE`（缺省 `{hostname}:{pid}`，跨重启持续命中请显式指定）。 +- `/scale` 写全局 count 后向该类型全部存活实例广播 `rescale`，各实例按同一份数据计算 quota， + 重复下发幂等。 + +### 5.2 inline 领导选举（多 worker 兼容） + +- 同一网关的多个服务器 worker（gunicorn `-w` 等）中，同一时刻仅一个进程托管 inline worker + （副本数精确等于 `count`，而非 worker 数 × count）；其余进程只做 HTTP 转发。 +- 持有者崩溃后，其余进程经周期性巡检抢锁接管并拉起 worker（`set_nx` 唯一胜出，无双持有）。 +- key 按 `gateway_id` 隔离，多网关各自选举、互不抢占。 + +### 5.3 MQTT Bridge 单例（多实例免重复响应） + +- Bridge 启动采用共享缓存分布式锁（`set_nx` + 租约续约），全局仅持锁网关消费 MQTT 请求， + 避免共享外部 Broker 时「一条请求被各消费一次、各回一份响应」。 +- 持有者失联后租约到期，其余网关经周期巡检（`bridge_supervise_interval`，默认 15s）自动接管重建。 +- 内嵌 Broker 单例按宿主隔离（`mqtt_broker:`）：同宿主多进程仅启动一个 Broker， + 接管时会清理前任遗留的孤儿 Broker 子进程。 + +### 5.4 部署形态速查 + +| 部署形态 | MQTT Broker | `[cache]` 后端 | +| -------- | ----------- | -------------- | +| 单机（含多 worker） | 内嵌 amqtt | 本地 `sqlite` | +| 多实例分布式 | **外部 Broker（必须）** | **redis**（或共享卷 `sqlite`） | + +## 6. 逐实例精细扩缩 + +同一类型跑多个外部进程（多机/多容器）时，除全局对称扩缩外，可只对**某一个实例**精确扩缩： + +- **语义**：把目标实例的副本数设为绝对值 N，其它实例不动；与全局 `scale` 共存—— + 全局对称是默认/一键复位（`scale` 会清除整张覆盖表回到均分），逐实例是定制覆盖。 +- **覆盖表**：cache 键 `worker:cm:instcounts:` = `{instance_id: 绝对副本数}`； + 无记录的实例回退到全局均分。 +- **命中判断**：`rescale` 命中本实例（`instance_id` 与登记一致）才执行，目标是其它实例则忽略。 +- **REST / UI**：`POST /admin/workers//scale_instance` 下发； + `GET /admin/workers//instances` 供前端按实例分组展示。 + +## 7. 实例定向调用 + +默认逐请求**随机负载均衡**，适合无状态插件。若插件持有只存在于某个进程本地的状态 +（本地 SQLite/文件、串口、设备连接），可用实例定向把请求固定到该进程： + +- **指定方式**（三入口语义一致）： + - HTTP 请求头 `X-Chongming-Instance: <实例id>` + - MQTT 信封载荷顶层 `target_instance`（网关读后摘除，业务侧不可见） + - 程序内 `worker_manager.call(...)` / `call_raw(...)` / `call_stream(...)` 的 `instance_id=` 参数 +- **实例 id 取值**：`--instance` → `CHONGMING_WORKER_INSTANCE` → `CHONGMING_GATEWAY_ID` → + `{hostname}:{pid}`；容器环境（K8s 注入 `CHONGMING_GATEWAY_ID`）下两侧自动一致， + 裸机同宿主多网关请显式指定。可读列表见 `GET /api/v1/admin/workers//instances`。 +- **粒度**：定向到**进程实例**；实例内有多个副本时，请求仍在这几个副本间随机均衡。 +- **实例离线时严格报错**（HTTP 503 / MQTT `{"error": ...}`），**不静默回退**到其它实例。 +- **重试不跨实例**；流式端点不重试，实例离线即失败。 +- **缓存按实例隔离**：定向且 `cacheable=true` 时缓存键带实例前缀。 + +```bash +# 把这次请求固定到实例 edge-01 的进程 +curl -H 'X-Chongming-Instance: edge-01' http://127.0.0.1:8000/api/v1/demo/state + +# MQTT：载荷顶层带 target_instance（网关转发前摘掉） +# topic: chongming/demo/state payload: {"meta":{}, "data":{}, "target_instance":"edge-01"} +``` + +## 8. 前端管理页 + +`Worker 管理`（`/workers`）： + +- **登录**：内置 `POST /auth/login` 领取管理 Token（不依赖 auth 插件），仅 `is_superuser` 可操作。 +- **列表**：副本类型 / 副本 ID / 模式（inline|external）/ 状态，支持关键字、模式、副本数筛选与分页； + 筛选同时作用于下方逐实例分组表。`worker_type == "auth"` 的配置项不展示、不可改。 +- **操作**：单副本重启、批量重启、插件级重启、配置编辑（结构化 JSON 校验/格式化）、 + **调整副本数量**（全局对称）、**逐实例精细扩缩**（按实例分组，仅调该实例）。 +- **自动刷新**：每 10s 轮询列表与分组表，异步操作结果自动反映。 +- 调小副本数会触发优雅排空，前端确认提示中会说明。 + +## 9. external 插件的可管理性前提 + +1. 插件容器运行含管理能力（`__admin__` + `reload` + `rescale`）的 `chongming-worker`； +2. `config_set` / `rescale` 由副本在**其进程 cwd** 下写 `config/workers//config.toml`， + 容器需挂载与网关同款的配置目录才能写回（含 `[workers].count` 持久化）； +3. 其它语言自行实现的进程插件需实现 `__admin__` Queryable + liveliness + 数量管理， + 否则只能被发现，无法改配置或扩缩容。 + +## 10. 多 worker 与 worker 副本的分工 + +gunicorn `-w N` 提升的是**网关的 HTTP 接入/转发能力**；插件计算能力由 Worker **副本数** +（`count`）或 external 进程数决定。高并发场景按瓶颈扩：转发瓶颈加 `-w`，计算瓶颈加 `count` +或改跑 external。