modify: worker 管理端点

2026-08-27 09:49:47 +08:00
parent 2270f2f6e8
commit 378128fc16
10 changed files with 967 additions and 143 deletions
@@ -28,6 +28,7 @@
 | **流媒体代理**        | ffmpeg + MediaMTX RTSP 流转发                                        |
 | **构建打包**          | PyInstaller / Nuitka 一键打包为独立可执行文件                        |
 | **Vue 3 前端**        | Element Plus UI、Pinia 状态管理、类型安全的 API 客户端               |
+| **Worker 运维管理**   | 前端经 Zenoh 向单个副本下发 restart/stop/改配置指令（超管），前端可视化列表 + 分页管理 |

 ---

@@ -207,18 +208,107 @@ uv run serve → FastAPI lifespan:
  2. setup_zenoh()             — Zenoh peer 会话
  3. setup_mqtt_broker()       — 内嵌 amqtt Broker（子进程 + 文件锁单例；
                                  `use_embedded_broker=false` 时跳过，改用外部 Broker）
-  4. setup_worker_manager()    — WorkerManager + 定期 liveliness 刷新
+  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 主题动态推送）
+  8. setup_mqtt()              — MQTT Bridge（inline 主题静态收集 + external 主题动态推送；基于共享缓存的分布式锁保证全局单实例）
 ```

 ---

+## Worker 管理系统
+
+通过 Zenoh 向 `worker/<type>/<worker_id>/__admin__` 管理端点下发指令，**精细化控制单个 Worker 副本**（inline 与 external 进程插件共用同一管理协议）。提供 REST 管理端点 + Vue 前端页面，操作需超级管理员鉴权。
+
+### 管理协议
+
+每个用 Worker 框架（`chongming_worker.runtime.create_workers`）创建的副本都会注册一个管理 Queryable：
+
+```text
+worker/<type>/<worker_id>/__admin__
+```
+
+| 指令        | 说明                                                                   |
+| ----------- | ---------------------------------------------------------------------- |
+| `status`    | 查询副本状态                                                           |
+| `restart`   | 重启该副本，副本槽位（index）由管理循环驱动，重新生成新 `worker_id`（`worker-<index>-<epoch>-<pid>`）并经 liveliness 更新 |
+| `stop`      | 停止该副本                                                             |
+| `drain`     | **优雅排空**：摘除 liveliness/业务 Queryable 隔绝新请求，等待正在处理的在途请求完成后再退出，管理循环不重建（数量-1） |
+| `rescale`   | 按 `data.scale` 把**当前进程内**该类型副本数调整为指定值（`data:{"scale":N}`），并把 `[workers].count` 写回本进程侧配置 |
+| `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/<type>/stop`     | 停止 single 副本 / 某插件全部副本      |
+| `GET`  | `/api/v1/admin/workers/<type>/config`   | 读取该插件的现有 `[workers.config]`（展示当前值） |
+| `POST` | `/api/v1/admin/workers/<type>/config`   | 更新配置（body `{updates:{k:v}}`），下发并重启 |
+| `POST` | `/api/v1/admin/workers/<type>/scale`    | 调整该类型副本数（body `{count:N}`），扩缩容 + 持久化 |
+
+> 旧版运行时未声明 `__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`，由其**所在进程内**的池扩缩并写回自身配置（无需重启整个进程/容器）。
+
+### 全局副本数协调（多网关）
+
+多个网关共享同一份 `chongming-cache`（根 `config.toml` 的 `[cache]` 段）时，副本数「count」语义为**该类 worker 的全局目标总量**，由 chongming-cache 作为**单一事实来源**，各网关按在线数量自动分摊。这样只在任意一台网关执行扩缩容，即可全局生效。
+
+- **单一事实来源**：`count` 存于 cache 键 `worker:cm:counts:<type>`，在线网关 id 列表存 `worker:cm:gateways`，各网关分摊配额存 `worker:cm:quota:<type>:<gw>`（可观测）。
+- **网关登记/注销 + 心跳租约**：网关启动时 `register_gateway()` 加入在线列表，并每 15s `heartbeat_gateway()` 刷新自身时间戳；超过租约（60s）未心跳的死网关会被自动清理（crash/kill -9 也不残留，避免配额失真）。当前由 `CHONGMING_GATEWAY_ID` 环境变量指定（多实例需唯一，尤其**同宿主多网关必须显式指定**；缺省回退 `hostname`，稳定且重启不变，不残留）；关闭时 `unregister_gateway()` 移除，其余网关实时重算分摊。
+- **分摊算法**：`own_quota()` 将全局总量按在线网关 sorted(id) 均分，余量按序补前若干。`/admin/workers/<type>/scale`（body `{count:<全局总量>}`）写全局 count 后，各网关把自己的 session 副本调到自己的 quota。
+- **共享方式与后端**：单实例本机 `diskcache` 即可；**多实例必须共享同一 cache 后端**（同一共享卷的 diskcache 或 redis），否则各写各的 cache 无法收敛成全局量。
+- **启动优先级开关**：根 `config.toml` 的 `[cache].launch_worker_count` 决定网关启动时副本数的来源：
+
+  ```toml
+  [cache]
+  backend = "diskcache"          # "diskcache"（默认） | "redis"
+  cache_store_path = "./cache_data"
+  launch_worker_count = "toml"   # "toml"（默认)=启动用 toml 的 count 覆盖并写回 cache；"cache"=cache 为单一事实来源，toml 仅在缺失时引导
+  ```
+
+- **进程插件（external）不受该开关影响**：网关只对 **inline** 插件用 cache 解析启动副本数（见「启动流程」第 5 步 `load_workers`）；external 插件由外部编排按其自身配置 `config/workers/<type>/config.toml` 的 `count` 启动，不经网关、也不读 cache。运行时 `scale` 对 external 仍走「向存活副本下发 rescale」路径，同样与该开关无关。
+
+### 前端管理页
+
+`Worker 管理`（`/workers`，深色主题）：
+
+- **登录**：经 auth 插件 `POST /auth/login` 领取管理 Token（无需手填 JWT），从 JWT 解析当前用户与超管标识。
+- **鉴权**：仅 `is_superuser=true` 可管理；权限不足时前端/后端弹出对应提示（403）。
+- **列表分页**：展示副本（插件类型 / 副本 ID / 模式 inline|external / 状态），内置分页。
+- **操作**：单副本重启、批量重启（选中）、插件级重启全部副本、更新插件配置（弹窗展示现有配置值，点行带入编辑）、**调整副本数量**（当前数 + `el-input-number` 设目标 + 应用，可扩缩容）。
+- **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/<type>/config.toml`，external 容器需挂载与网关同款配置目录才能写回网关侧配置（含 `[workers].count` 持久化）；
+3. 纯自定义语言（如 Rust `rsdemo`）进程插件需自行实现 `__admin__` Queryable + liveliness + 数量管理，否则只能发现、无法改写配置或扩缩容。
+
+---
+
 ## 端点类型与路由注册

 ### 端点类型
@@ -482,6 +572,23 @@ response_topic = "/board_ping/response"     # 可选，响应主题；默认 "<
 - 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。
 - 全局 `[<env>.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。

+### MQTT Bridge 单例（多实例免重复响应）
+
+MQTT Bridge 是**网关进程内**的 MQTT 客户端组件，负责订阅主题并把请求经 `WorkerManager` 转发给 Worker、回发 `/response`。为了让「同一条请求只被消费一次、只回一份响应」，Bridge 的启动采用**基于共享缓存的分布式锁**（`chongming-cache` 提供，`DistributedLock`），而不再使用本机文件锁。
+
+- **抢锁机制**：启动时 `set_nx`（SET NX + TTL 租约）原子抢锁，key 为 `mqtt_bridge`；抢到锁的网关才启动 Bridge，其余网关跳过。
+- **租约续约**：持锁网关后台定期 `touch` 续约；若持锁网关宕机/失联，租约到期后锁自动释放，其他网关可重新抢到（重启后生效）。
+- **受保护释放**：关闭时仅当 token 仍匹配（仍是自己持有）才删除，绝不误删其他网关抢到的锁。
+
+```toml
+[cache]
+backend = "redis"            # 跨主机收敛必须用 redis（或共享卷 diskcache）
+```
+
+- **关键条件**：跨机器收敛依赖**共享缓存后端**（与「全局副本数协调」同一 `[cache]`，单一事实来源）。仅 `redis`（或共享卷的 `diskcache`）后端时，不同物理机上的网关才会看到同一把锁，真正全局单 Bridge。
+- **`use_embedded_broker = true`（各机内嵌独立 Broker）**：各机 Broker 互不相连、本身不产生重复响应；此时全局单 Bridge 只会连到持锁网关那台机器的 Broker。
+- **`use_embedded_broker = false`（共享外部 Broker）**：原先 N 台网关的 N 个 Bridge 会共同订阅同一批主题，导致一条请求被各消费一次、各回一份响应；改用分布式锁后全局只有一个 Bridge 消费，彻底规避重复响应。
+
 ### 请求负载格式（信封协议）

 MQTT 请求负载为信封结构，控制字段与业务数据分离：
@@ -655,6 +762,7 @@ uv pip install -e utils/<name>            # 安装/更新工具包子包
 | 字段                 | 位置                  | 说明                                    |
 | -------------------- | --------------------- | --------------------------------------- |
 | `count`              | `<插件>/config.toml`  | Worker 副本数（`[workers]` 节）         |
+| `launch_worker_count`| `根 config.toml`      | 启动时副本数来源：`"toml"`=用 toml count 覆盖 cache（默认）；`"cache"`=以 cache 为单一事实来源（`[cache]` 节） |
 | `router_prefix`      | `<插件>/config.toml`  | API 路由前缀                            |
 | `defaults.timeout`   | `<插件>/config.toml`  | 默认超时（秒）                          |
 | `defaults.cacheable` | `<插件>/config.toml`  | 是否缓存 RPC 结果                       |