已合并
[Feature] FaultReporter 对齐 vLLM FT 框架：ZMQ 订阅改为 HTTP 轮询 #678
jason lyu创建于 27 天前
[Feature] FaultReporter 对齐 vLLM FT 框架：ZMQ 订阅改为 HTTP 轮询 #678
已合并
共 11 个文件变更+819-518
| @@ -35,8 +35,8 @@ NodeManager Process (sidecar, one per pod) — NodeManager(Application) main loo | |||
| 35 | │ Fault detection: 120s grace period + 5 consecutive abnormal reports → suicide | 35 | │ Fault detection: 120s grace period + 5 consecutive abnormal reports → suicide |
| 36 | │ | 36 | │ |
| 37 | └── FaultReporter (ThreadSafeSingleton) | 37 | └── FaultReporter (ThreadSafeSingleton) |
| 38 | - ZMQ SUB sockets on engine PUB ports (topic: vllm_fault) | 38 | + HTTP poll GET {business_port}/fault_tolerance/status per engine |
| 39 | - → report_software_fault to Controller | 39 | + (vLLM FT REST API) → report_software_fault to Controller |
| 40 | ``` | 40 | ``` |
| 41 | 41 | ||
| 42 | All core components use `ThreadSafeSingleton` — `__new__` + `threading.Lock` ensures one instance per process with thread-safe lazy initialization. | 42 | All core components use `ThreadSafeSingleton` — `__new__` + `threading.Lock` ensures one instance per process with thread-safe lazy initialization. |
| @@ -204,7 +204,7 @@ When `is_restored_from_host_side_snapshot()` returns True: | |||
| 204 | | `motor/node_manager/core/services/registry.py` | `@register_service` decorator + `_MODULE_MAP`; discovers active services by pod profile | | 204 | | `motor/node_manager/core/services/registry.py` | `@register_service` decorator + `_MODULE_MAP`; discovers active services by pod profile | |
| 205 | | `motor/node_manager/core/services/memcache/` | KV-store (memcache) service implementation | | 205 | | `motor/node_manager/core/services/memcache/` | KV-store (memcache) service implementation | |
| 206 | | `motor/node_manager/core/heartbeat_manager.py` | Two daemon threads: status polling + heartbeat reporting, fault detection state machine, reregister | | 206 | | `motor/node_manager/core/heartbeat_manager.py` | Two daemon threads: status polling + heartbeat reporting, fault detection state machine, reregister | |
| 207 | -| `motor/node_manager/core/fault_reporter.py` | ZMQ SUB on engine PUB ports (topic `vllm_fault`) → `report_software_fault` to Controller | | 207 | +| `motor/node_manager/core/fault_reporter.py` | HTTP poll of engine `GET /fault_tolerance/status` (business port) → `report_software_fault` to Controller | |
| 208 | | `motor/node_manager/api_client/controller_api_client.py` | Sync HTTP client to Controller: `/register`, `/reregister`, `/heartbeat` | | 208 | | `motor/node_manager/api_client/controller_api_client.py` | Sync HTTP client to Controller: `/register`, `/reregister`, `/heartbeat` | |
| 209 | | `motor/node_manager/api_client/engine_server_api_client.py` | Sync HTTP client: `GET /status` on engine's mgmt port | | 209 | | `motor/node_manager/api_client/engine_server_api_client.py` | Sync HTTP client: `GET /status` on engine's mgmt port | |
| 210 | | `motor/config/node_manager.py` | `NodeManagerConfig`: BasicConfig, APIConfig, EndpointConfig, SnapshotConfig, SingleContainerConfig, PortAllocatorConfig | | 210 | | `motor/config/node_manager.py` | `NodeManagerConfig`: BasicConfig, APIConfig, EndpointConfig, SnapshotConfig, SingleContainerConfig, PortAllocatorConfig | |
| @@ -47,29 +47,26 @@ Controller 侧: | |||
| 47 | 47 | ||
| 48 | NodeManager 侧: | 48 | NodeManager 侧: |
| 49 | FaultReporter (EngineManager 聚合) | 49 | FaultReporter (EngineManager 聚合) |
| 50 | - ├── ZMQ SUB → 订阅 vllm ClientSentinel PUB (每引擎一个 socket) | 50 | + ├── HTTP 轮询 → GET {endpoint.business_port}/fault_tolerance/status (vLLM FT API) |
| 51 | - ├── msgspec.msgpack 解码引擎状态 | ||
| 52 | ├── 状态去重 → 仅上报 dead/unhealthy 变更 | 51 | ├── 状态去重 → 仅上报 dead/unhealthy 变更 |
| 52 | + ├── 连续 max_poll_failures 次轮询失败 → 按 dead 上报 | ||
| 53 | └── HTTP POST → Controller /controller/report_software_fault | 53 | └── HTTP POST → Controller /controller/report_software_fault |
| 54 | ``` | 54 | ``` |
| 55 | 55 | ||
| 56 | ## 故障上报链路 (端到端) | 56 | ## 故障上报链路 (端到端) |
| 57 | 57 | ||
| 58 | ```text | 58 | ```text |
| 59 | -vllm EngineCore 异常 | 59 | +vllm EngineCore 异常 → 引擎状态变为 unhealthy/dead |
| 60 | - → (ZMQ DEALER) EngineCoreSentinel 发送 FaultInfo | 60 | + → (HTTP) FaultReporter 轮询 GET /fault_tolerance/status (每 poll_interval_sec) |
| 61 | - → (ZMQ ROUTER) ClientSentinel 接收并更新状态 | 61 | + → 解析 engines[] 状态 → 去重后 (仅上报状态变更) |
| 62 | - → (ZMQ PUB) 广播 engine_status (msgpack) 到 fault_state_pub_socket | 62 | + → _send_fault_to_controller() 注入 pod_ip |
| 63 | - → (ZMQ SUB) FaultReporter._loop() 订阅 (每引擎一个 socket) | 63 | + → ControllerApiClient.report_software_fault() |
| 64 | - → _process_zmq_engine_status() 去重后 | 64 | + → POST /controller/report_software_fault |
| 65 | - → _send_fault_to_controller() 注入 pod_ip | 65 | + → FaultManager.report_software_fault(pod_ip) |
| 66 | - → ControllerApiClient.report_software_fault() | 66 | + → NodeMetadata.software_fault_infos += fault |
| 67 | - → POST /controller/report_software_fault | 67 | + → _refresh_instance_fault_level() |
| 68 | - → FaultManager.report_software_fault(pod_ip) | 68 | + → 综合硬件+软件 → 更新 InstanceMetadata.fault_level |
| 69 | - → NodeMetadata.software_fault_infos += fault | 69 | + → 策略中心 → 生成/升级恢复策略 |
| 70 | - → _refresh_instance_fault_level() | ||
| 71 | - → 综合硬件+软件 → 更新 InstanceMetadata.fault_level | ||
| 72 | - → 策略中心 → 生成/升级恢复策略 | ||
| 73 | ``` | 70 | ``` |
| 74 | 71 | ||
| 75 | ## 数据结构 | 72 | ## 数据结构 |
| @@ -427,5 +424,7 @@ L4/L5/L6 → 根据实例角色 (decode) → ScaleP2DStrategy | |||
| 427 | 424 | ||
| 428 | | 参数 | 类型 | 说明 | | 425 | | 参数 | 类型 | 说明 | |
| 429 | |---|---|---| | 426 | |---|---|---| |
| 430 | -| `enable_fault_tolerance` | bool | 是否启用故障上报线程。默认: `false` | | 427 | +| `enable_fault_tolerance` | bool | 显式启用故障上报线程；引擎 user config 检测到 FT 时自动启用。默认: `false` | |
| 431 | -| `zmq_pub_port` | int | ZMQ SUB 订阅的基端口 (每个引擎 = base_port + engine_id)。默认: `0` | | 428 | +| `poll_interval_sec` | float | 轮询引擎 FT 状态接口的间隔 (秒)。默认: `5.0` | |
| 429 | +| `poll_timeout_sec` | float | 单次轮询的 HTTP 超时 (秒)。默认: `5.0` | | ||
| 430 | +| `max_poll_failures` | int | 连续轮询失败阈值，达到后按 `dead` 上报。默认: `3` | | ||
| @@ -516,7 +516,7 @@ flowchart LR | |||
| 516 | K8sNode[K8s Node 状态] -->|Watch| RM_Node[ResourceMonitor] | 516 | K8sNode[K8s Node 状态] -->|Watch| RM_Node[ResourceMonitor] |
| 517 | end | 517 | end |
| 518 | subgraph SW[软件故障感知] | 518 | subgraph SW[软件故障感知] |
| 519 | - Engine[vLLM Engine] -->|ZMQ PUB/SUB| FR[FaultReporter<br/>NodeManager] | 519 | + Engine[vLLM Engine] -->|HTTP 轮询<br/>GET /fault_tolerance/status| FR[FaultReporter<br/>NodeManager] |
| 520 | FR -->|HTTP| API[ControllerAPI<br/>/report_software_fault] | 520 | FR -->|HTTP| API[ControllerAPI<br/>/report_software_fault] |
| 521 | end | 521 | end |
| 522 | RM_HW --> FM[FaultManager] | 522 | RM_HW --> FM[FaultManager] |
| @@ -23,7 +23,7 @@ Node Manager 是部署在推理节点上的管理进程，负责连接 Controlle | |||
| 23 | | `LocalService` | `motor/node_manager/core/services/memcache/lifecycle.py` | memcache 后端生命周期管理：配置准备、子进程拉起（通过 `memcache/worker.py`）、健康检查与重启 | | 23 | | `LocalService` | `motor/node_manager/core/services/memcache/lifecycle.py` | memcache 后端生命周期管理：配置准备、子进程拉起（通过 `memcache/worker.py`）、健康检查与重启 | |
| 24 | | `EngineManager` | `motor/node_manager/core/engine_manager.py` | 注册/重注册、校验启动命令、处理 ranktable、快照元数据和故障上报 | | 24 | | `EngineManager` | `motor/node_manager/core/engine_manager.py` | 注册/重注册、校验启动命令、处理 ranktable、快照元数据和故障上报 | |
| 25 | | `HeartbeatManager` | `motor/node_manager/core/heartbeat_manager.py` | 轮询 endpoint 状态、上报心跳、维护暂停/恢复状态并触发异常自杀 | | 25 | | `HeartbeatManager` | `motor/node_manager/core/heartbeat_manager.py` | 轮询 endpoint 状态、上报心跳、维护暂停/恢复状态并触发异常自杀 | |
| 26 | -| `FaultReporter` | `motor/node_manager/core/fault_reporter.py` | 订阅 Engine Server 的 ZMQ 软件故障消息并转发给 Controller | | 26 | +| `FaultReporter` | `motor/node_manager/core/fault_reporter.py` | 轮询引擎 FT 状态接口并上报软件故障给 Controller | |
| 27 | | `ControllerApiClient` | `motor/node_manager/api_client/controller_api_client.py` | 调用 Controller 的注册、重注册、心跳和故障上报接口 | | 27 | | `ControllerApiClient` | `motor/node_manager/api_client/controller_api_client.py` | 调用 Controller 的注册、重注册、心跳和故障上报接口 | |
| 28 | | `EngineServerApiClient` | `motor/node_manager/api_client/engine_server_api_client.py` | 调用 Engine Server 管理面的 `GET /status` | | 28 | | `EngineServerApiClient` | `motor/node_manager/api_client/engine_server_api_client.py` | 调用 Engine Server 管理面的 `GET /status` | |
| 29 | 29 | ||
| @@ -186,8 +186,10 @@ Node Manager 从 `engine_config.nnodes` 推导每节点 `local_world_size`。当 | |||
| 186 | | `basic_config.nnodes` | `1` | 从 `engine_config.nnodes` 派生的跨节点数量 | | 186 | | `basic_config.nnodes` | `1` | 从 `engine_config.nnodes` 派生的跨节点数量 | |
| 187 | | `kv_cache_store_config.mode` | `combined` | 部署模式：`combined` 表示 Engine 与 KV-store 在同一 Pod，`separated` 表示 KV-store 独立 Pod（不拉 Engine、不注册、不心跳） | | 187 | | `kv_cache_store_config.mode` | `combined` | 部署模式：`combined` 表示 Engine 与 KV-store 在同一 Pod，`separated` 表示 KV-store 独立 Pod（不拉 Engine、不注册、不心跳） | |
| 188 | | `mgmt_tls_config.enable_tls` | `false` | Node Manager、Controller 和 Engine Server 管理面通信是否启用 TLS | | 188 | | `mgmt_tls_config.enable_tls` | `false` | Node Manager、Controller 和 Engine Server 管理面通信是否启用 TLS | |
| 189 | -| `fault_tolerance_config.enable_fault_tolerance` | `false` | 是否启动软件故障订阅线程 | | 189 | +| `fault_tolerance_config.enable_fault_tolerance` | `false` | 显式开启软件故障轮询；引擎 user config 检测到 FT 时自动开启，无需配置 | |
| 190 | -| `fault_tolerance_config.zmq_pub_port` | `0` | ZMQ PUB 基础端口；每个 endpoint 使用 `base_port + endpoint.id` | | 190 | +| `fault_tolerance_config.poll_interval_sec` | `5.0` | 轮询引擎 FT 状态的时间间隔（秒） | |
| 191 | +| `fault_tolerance_config.poll_timeout_sec` | `5.0` | 单次轮询的 HTTP 超时（秒） | | ||
| 192 | +| `fault_tolerance_config.max_poll_failures` | `3` | 连续轮询失败阈值，达到后按 `dead` 上报 | | ||
| 191 | | `snapshot_config.enable_snapshot` | `false` | 是否启用容器快照流程 | | 193 | | `snapshot_config.enable_snapshot` | `false` | 是否启用容器快照流程 | |
| 192 | | `snapshot_config.snapshot_metadata_path` | 空 | 自定义快照元数据路径；用户需预先创建并挂载该文件。为空时进入快照默认应用场景，即 MindCluster 实例重调度 | | 194 | | `snapshot_config.snapshot_metadata_path` | 空 | 自定义快照元数据路径；用户需预先创建并挂载该文件。为空时进入快照默认应用场景，即 MindCluster 实例重调度 | |
| 193 | | `port_allocator_config.enable` | `true` | 是否在启动时自动检查并调整端口 | | 195 | | `port_allocator_config.enable` | `true` | 是否在启动时自动检查并调整端口 | |
| @@ -203,7 +205,7 @@ Node Manager 从 `engine_config.nnodes` 推导每节点 `local_world_size`。当 | |||
| 203 | 1. `_refresh_check_interval()` — 从配置刷新 daemon loop 间隔。 | 205 | 1. `_refresh_check_interval()` — 从配置刷新 daemon loop 间隔。 |
| 204 | 2. 遍历所有模块调用 `update_config()`： | 206 | 2. 遍历所有模块调用 `update_config()`： |
| 205 | - `HeartbeatManager` 动态更新 `heartbeat_interval_seconds`。 | 207 | - `HeartbeatManager` 动态更新 `heartbeat_interval_seconds`。 |
| 206 | - - `EngineManager` 更新配置，并根据 `enable_fault_tolerance`、endpoint、Pod IP 或 `zmq_pub_port` 的变化启停或重建 `FaultReporter`。 | 208 | + - `EngineManager` 更新配置，并根据 `enable_fault_tolerance`、endpoint 的变化启停或重建 `FaultReporter`。 |
| 207 | 3. 打印更新后的配置摘要 `log_configuration_summary()`。 | 209 | 3. 打印更新后的配置摘要 `log_configuration_summary()`。 |
| 208 | 4. API 监听地址、监听端口、TLS 和 `Daemon` 已缓存的设备参数不会热重启，修改后需要重启 Node Manager。 | 210 | 4. API 监听地址、监听端口、TLS 和 `Daemon` 已缓存的设备参数不会热重启，修改后需要重启 Node Manager。 |
| 209 | 211 | ||
| @@ -271,19 +273,29 @@ registry.add_discovery_path("new_backend", "path.to.new_backend_module") | |||
| 271 | 273 | ||
| 272 | ## 软件故障上报 | 274 | ## 软件故障上报 |
| 273 | 275 | ||
| 274 | -开启 `fault_tolerance_config.enable_fault_tolerance` 后，`FaultReporter` 为每个 endpoint 连接一个 ZMQ SUB socket，订阅主题 `vllm_fault`。端口为： | 276 | +`FaultReporter` 在以下任一条件满足时自动启用：`fault_tolerance_config.enable_fault_tolerance` 显式开启，或 user config 的引擎配置（如 `motor_engine_prefill_config.engine_config`）中检测到 `enable-fault-tolerance` / `enable_fault_tolerance` 为 `true`（无需 NodeManager 显式配置）。启用后在后台线程中按 `poll_interval_sec` 间隔轮询每个 endpoint 的 FT 状态接口： |
| 275 | 277 | ||
| 276 | ```text | 278 | ```text |
| 277 | -fault_tolerance_config.zmq_pub_port + endpoint.id | 279 | +GET http://{endpoint.ip}:{endpoint.business_port}/fault_tolerance/status |
| 278 | ``` | 280 | ``` |
| 279 | 281 | ||
| 280 | -消息中的状态映射为： | 282 | +响应格式（vLLM FaultTolerance 框架提供，见 vllm-project/vllm#44428）： |
| 283 | + | ||
| 284 | +```json | ||
| 285 | +{ | ||
| 286 | + "schema_version": 1, | ||
| 287 | + "total_engines": 1, | ||
| 288 | + "engines": [{"id": 0, "status": "healthy|dead|unhealthy", "fault_info": "..."}] | ||
| 289 | +} | ||
| 290 | +``` | ||
| 291 | + | ||
| 292 | +状态映射为： | ||
| 281 | 293 | ||
| 282 | - `healthy`：记录状态，不上报故障。 | 294 | - `healthy`：记录状态，不上报故障。 |
| 283 | - `dead`：上报 `EngineDeadError`。 | 295 | - `dead`：上报 `EngineDeadError`。 |
| 284 | -- `unhealthy`：上报 `EngineUnhealthyError`。 | 296 | +- `unhealthy`：上报 `EngineUnhealthyError`；响应携带 `fault_info` 时以其作为 `exception_type`（如 `RuntimeError`）。 |
| 285 | 297 | ||
| 286 | -同一 Engine 的相同非健康状态只在成功发送给 Controller 后标记为已上报；发送失败时后续消息仍可重试。ZMQ 发生错误后等待 5 秒并重建订阅。 | 298 | +同一 Engine 的相同非健康状态只在成功发送给 Controller 后标记为已上报；发送失败时后续轮询仍可重试。引擎连续 `max_poll_failures` 次轮询失败（连接拒绝/超时）时按 `dead` 上报 `EngineDeadError`（异常消息注明不可达轮询次数），引擎恢复可轮询后重新计数。 |
| 287 | 299 | ||
| 288 | ## 容器快照 | 300 | ## 容器快照 |
| 289 | 301 | ||
| @@ -553,7 +553,11 @@ motor_engine_union_config字段用于**PD混部场景**，配置同一类union E | |||
| 553 | }, | 553 | }, |
| 554 | "single_container_config": {... | 554 | "single_container_config": {... |
| 555 | }, | 555 | }, |
| 556 | - "fault_tolerance_config": {... | 556 | + "fault_tolerance_config": { |
| 557 | + "enable_fault_tolerance": false, | ||
| 558 | + "poll_interval_sec": 5.0, | ||
| 559 | + "poll_timeout_sec": 5.0, | ||
| 560 | + "max_poll_failures": 3 | ||
| 557 | }, | 561 | }, |
| 558 | "port_allocator_config": { | 562 | "port_allocator_config": { |
| 559 | "enable": true, | 563 | "enable": true, |
| @@ -579,6 +583,10 @@ motor_engine_union_config字段用于**PD混部场景**，配置同一类union E | |||
| 579 | | endpoint_config.base_port |int | 端点端口起始号。默认值：`10000` | | 583 | | endpoint_config.base_port |int | 端点端口起始号。默认值：`10000` | |
| 580 | | endpoint_config.mgmt_ports |array | 各端点管控端口列表（整数数组）。默认值：`[]` | | 584 | | endpoint_config.mgmt_ports |array | 各端点管控端口列表（整数数组）。默认值：`[]` | |
| 581 | | endpoint_config.service_ports |array | 各端点推理服务端口列表（整数数组）。默认值：`[]` | | 585 | | endpoint_config.service_ports |array | 各端点推理服务端口列表（整数数组）。默认值：`[]` | |
| 586 | +| fault_tolerance_config.enable_fault_tolerance |bool|是否显式开启引擎软件故障轮询，默认值：false。<br>引擎 user config 检测到 FT 开关时自动开启，无需显式配置。| | ||
| 587 | +| fault_tolerance_config.poll_interval_sec |float|轮询引擎 FT 状态的时间间隔，单位：秒，默认值：5.0。| | ||
| 588 | +| fault_tolerance_config.poll_timeout_sec |float|单次轮询的 HTTP 超时，单位：秒，默认值：5.0。| | ||
| 589 | +| fault_tolerance_config.max_poll_failures |int|连续轮询失败阈值，达到后按 dead 上报，默认值：3。| | ||
| 582 | | snapshot_config.enable_snapshot |bool|是否使能容器快照功能总开关，默认值：false。<br>开启后，用户可对实例容器制作快照镜像，并支持由快照恢复的实例向控制面注册。| | 590 | | snapshot_config.enable_snapshot |bool|是否使能容器快照功能总开关，默认值：false。<br>开启后，用户可对实例容器制作快照镜像，并支持由快照恢复的实例向控制面注册。| |
| 583 | | snapshot_config.snapshot_metadata_path |string|容器快照元数据文件路径，包含容器快照制作与恢复过程中依赖的元数据，默认值为空。| | 591 | | snapshot_config.snapshot_metadata_path |string|容器快照元数据文件路径，包含容器快照制作与恢复过程中依赖的元数据，默认值为空。| |
| 584 | | logging_config.log_level | string | 日志级别，默认值：INFO<ul><li>DEBUG</li><li>INFO</li><li>WARNING</li><li>ERROR</li></ul>| | 592 | | logging_config.log_level | string | 日志级别，默认值：INFO<ul><li>DEBUG</li><li>INFO</li><li>WARNING</li><li>ERROR</li></ul>| |
| @@ -667,7 +675,11 @@ motor_engine_prefill_config和motor_engine_decode_config字段用于**PD分离 | |||
| 667 | }, | 675 | }, |
| 668 | "single_container_config": {... | 676 | "single_container_config": {... |
| 669 | }, | 677 | }, |
| 670 | - "fault_tolerance_config": {... | 678 | + "fault_tolerance_config": { |
| 679 | + "enable_fault_tolerance": false, | ||
| 680 | + "poll_interval_sec": 5.0, | ||
| 681 | + "poll_timeout_sec": 5.0, | ||
| 682 | + "max_poll_failures": 3 | ||
| 671 | }, | 683 | }, |
| 672 | "port_allocator_config": { | 684 | "port_allocator_config": { |
| 673 | "enable": true, | 685 | "enable": true, |
| @@ -738,7 +750,11 @@ motor_engine_prefill_config和motor_engine_decode_config字段用于**PD分离 | |||
| 738 | }, | 750 | }, |
| 739 | "single_container_config": {... | 751 | "single_container_config": {... |
| 740 | }, | 752 | }, |
| 741 | - "fault_tolerance_config": {... | 753 | + "fault_tolerance_config": { |
| 754 | + "enable_fault_tolerance": false, | ||
| 755 | + "poll_interval_sec": 5.0, | ||
| 756 | + "poll_timeout_sec": 5.0, | ||
| 757 | + "max_poll_failures": 3 | ||
| 742 | }, | 758 | }, |
| 743 | "port_allocator_config": { | 759 | "port_allocator_config": { |
| 744 | "enable": true, | 760 | "enable": true, |
| @@ -765,6 +781,10 @@ motor_engine_prefill_config和motor_engine_decode_config字段用于**PD分离 | |||
| 765 | | endpoint_config.base_port |int | 端点端口起始号。默认值：`10000` | | 781 | | endpoint_config.base_port |int | 端点端口起始号。默认值：`10000` | |
| 766 | | endpoint_config.mgmt_ports |array | 各端点管控端口列表（整数数组）。默认值：`[]` | | 782 | | endpoint_config.mgmt_ports |array | 各端点管控端口列表（整数数组）。默认值：`[]` | |
| 767 | | endpoint_config.service_ports |array | 各端点推理服务端口列表（整数数组）。默认值：`[]` | | 783 | | endpoint_config.service_ports |array | 各端点推理服务端口列表（整数数组）。默认值：`[]` | |
| 784 | +| fault_tolerance_config.enable_fault_tolerance |bool|是否显式开启引擎软件故障轮询，默认值：false。<br>引擎 user config 检测到 FT 开关时自动开启，无需显式配置。| | ||
| 785 | +| fault_tolerance_config.poll_interval_sec |float|轮询引擎 FT 状态的时间间隔，单位：秒，默认值：5.0。| | ||
| 786 | +| fault_tolerance_config.poll_timeout_sec |float|单次轮询的 HTTP 超时，单位：秒，默认值：5.0。| | ||
| 787 | +| fault_tolerance_config.max_poll_failures |int|连续轮询失败阈值，达到后按 dead 上报，默认值：3。| | ||
| 768 | | snapshot_config.enable_snapshot |bool|是否使能容器快照功能总开关，默认值：false。<br>开启后，用户可对实例容器制作快照镜像，并支持由快照恢复的实例向控制面注册。| | 788 | | snapshot_config.enable_snapshot |bool|是否使能容器快照功能总开关，默认值：false。<br>开启后，用户可对实例容器制作快照镜像，并支持由快照恢复的实例向控制面注册。| |
| 769 | | snapshot_config.snapshot_metadata_path |string|容器快照元数据文件路径，包含容器快照制作与恢复过程中依赖的元数据，默认值为空。| | 789 | | snapshot_config.snapshot_metadata_path |string|容器快照元数据文件路径，包含容器快照制作与恢复过程中依赖的元数据，默认值为空。| |
| 770 | | logging_config.log_level | string | 日志级别，默认值：INFO<ul><li>DEBUG</li><li>INFO</li><li>WARNING</li><li>ERROR</li></ul>| | 790 | | logging_config.log_level | string | 日志级别，默认值：INFO<ul><li>DEBUG</li><li>INFO</li><li>WARNING</li><li>ERROR</li></ul>| |
| @@ -358,7 +358,9 @@ | |||
| 358 | }, | 358 | }, |
| 359 | "fault_tolerance_config": { | 359 | "fault_tolerance_config": { |
| 360 | "enable_fault_tolerance": false, | 360 | "enable_fault_tolerance": false, |
| 361 | - "zmq_pub_port": 0 | 361 | + "poll_interval_sec": 5.0, |
 | |||
| 362 | + "poll_timeout_sec": 5.0, | ||
| 363 | + "max_poll_failures": 3 | ||
| 362 | }, | 364 | }, |
| 363 | "port_allocator_config": { | 365 | "port_allocator_config": { |
| 364 | "enable": true, | 366 | "enable": true, |
| @@ -1,4 +1,3 @@ | |||
| 1 | -# -*- coding: utf-8 -*- | ||
| 2 | # Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved. | 1 | # Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved. |
| 3 | # MindIE is licensed under Mulan PSL v2. | 2 | # MindIE is licensed under Mulan PSL v2. |
| 4 | # You can use this software according to the terms and conditions of the Mulan PSL v2. | 3 | # You can use this software according to the terms and conditions of the Mulan PSL v2. |
| @@ -18,6 +17,12 @@ __all__ = [ | |||
| 18 | "CHAT_COMPLETION_PREFIX", | 17 | "CHAT_COMPLETION_PREFIX", |
| 19 | "COMPLETION_PREFIX", | 18 | "COMPLETION_PREFIX", |
| 20 | "COMPLETION_SUFFIX", | 19 | "COMPLETION_SUFFIX", |
| 20 | + "FT_STATUS_PATH", | ||
| 21 | + "ENGINE_STATUS_HEALTHY", | ||
| 22 | + "ENGINE_STATUS_DEAD", | ||
| 23 | + "ENGINE_STATUS_UNHEALTHY", | ||
| 24 | + "ENGINE_STATUS_UNKNOWN", | ||
| 25 | + "ENGINE_FT_TIMEOUT", | ||
| 21 | ] | 26 | ] |
| 22 | 27 | ||
| 23 | from typing import Final | 28 | from typing import Final |
| @@ -27,3 +32,15 @@ CHAT_COMPLETION_PREFIX: Final[str] = "chatcmpl-" | |||
| 27 | COMPLETION_PREFIX: Final[str] = "cmpl-" | 32 | COMPLETION_PREFIX: Final[str] = "cmpl-" |
| 28 | # /v1/completions: cmpl-xxx-0 | 33 | # /v1/completions: cmpl-xxx-0 |
| 29 | COMPLETION_SUFFIX: Final[str] = "-0" | 34 | COMPLETION_SUFFIX: Final[str] = "-0" |
| 35 | + | ||
| 36 | +# Engine FT API routes (vLLM FaultTolerance framework, vllm-project/vllm#44428). | ||
| 37 | +FT_STATUS_PATH: Final[str] = "/fault_tolerance/status" | ||
| 38 | + | ||
| 39 | +# Engine FT status vocabulary. | ||
| 40 | +ENGINE_STATUS_HEALTHY: Final[str] = "healthy" | ||
| 41 | +ENGINE_STATUS_DEAD: Final[str] = "dead" | ||
| 42 | +ENGINE_STATUS_UNHEALTHY: Final[str] = "unhealthy" | ||
| 43 | +ENGINE_STATUS_UNKNOWN: Final[str] = "unknown" | ||
| 44 | + | ||
| 45 | +# Engine FT API timeout (business port, non-TLS). | ||
| 46 | +ENGINE_FT_TIMEOUT: Final[float] = 5.0 | ||
| @@ -0,0 +1,32 @@ | |||
| 1 | +# Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved. | ||
| 2 | +# MindIE is licensed under Mulan PSL v2. | ||
| 3 | +# You can use this software according to the terms and conditions of the Mulan PSL v2. | ||
| 4 | +# You may obtain a copy of Mulan PSL v2 at: | ||
| 5 | +# http://license.coscl.org.cn/MulanPSL2 | ||
| 6 | +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, | ||
| 7 | +# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, | ||
| 8 | +# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. | ||
| 9 | +# See the Mulan PSL v2 for more details. | ||
| 10 | +"""Engine FaultTolerance HTTP client: poll the engine's FT status endpoint. | ||
| 11 | + | ||
| 12 | +Single source of truth for talking to the vLLM FT API on the engine's | ||
| 13 | +business port; protocol constants live in ``motor.common.constants``. | ||
| 14 | +""" | ||
| 15 | + | ||
| 16 | +from motor.common.constants import ENGINE_FT_TIMEOUT, FT_STATUS_PATH | ||
| 17 | +from motor.common.http.http_client import SafeHTTPSClient | ||
| 18 | +from motor.common.logger import get_logger | ||
| 19 | +from motor.common.resources.endpoint import Endpoint | ||
| 20 | +from motor.common.utils.net import format_address | ||
| 21 | + | ||
 get(FT_STATUS_PATH) 失败未捕获，会抛异常使恢复命令中断，建议用 try/except 降级为 unknown。   | |||
| 22 | +logger = get_logger(__name__) | ||
| 23 | + | ||
| 24 | + | ||
| 25 | +def query_engine_ft_status(ep: Endpoint, timeout: float = ENGINE_FT_TIMEOUT) -> dict: | ||
| 26 | + """GET one engine's FT status payload; raises ValueError on a non-dict body.""" | ||
| 27 | + address = format_address(ep.ip, ep.business_port) | ||
| 28 | + with SafeHTTPSClient(address=address, tls_config=None, timeout=timeout) as client: | ||
| 29 | + payload = client.get(FT_STATUS_PATH) | ||
| 30 | + if not isinstance(payload, dict): | ||
| 31 | + raise ValueError("unexpected FT status payload type for engine %d: %s" % (ep.id, type(payload).__name__)) | ||
| 32 | + return payload | ||
| @@ -287,7 +287,12 @@ class NodeManagerFaultToleranceConfig: | |||
| 287 | """Fault tolerance configuration for NodeManager""" | 287 | """Fault tolerance configuration for NodeManager""" |
| 288 | 288 | ||
| 289 | enable_fault_tolerance: bool = False | 289 | enable_fault_tolerance: bool = False |
| 290 | - zmq_pub_port: int = 0 | 290 | + poll_interval_sec: float = 5.0 |
| 291 | + #: Polling interval for engine FT status (vLLM /fault_tolerance/status). | ||
| 292 | + poll_timeout_sec: float = 5.0 | ||
| 293 | + #: HTTP timeout for a single status poll. | ||
| 294 | + max_poll_failures: int = 3 | ||
| 295 | + #: Consecutive poll failures before the engine is reported as dead. | ||
| 291 | 296 | ||
| 292 | 297 | ||
| 293 | 298 | ||
| @@ -7,52 +7,124 @@ | |||
| 7 | # EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, | 7 | # EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, |
| 8 | # MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. | 8 | # MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. |
| 9 | # See the Mulan PSL v2 for more details. | 9 | # See the Mulan PSL v2 for more details. |
| 10 | -"""FaultReporter – subscribes to EngineServer ZMQ PUB sockets and forwards | 10 | +"""FaultReporter – polls each engine's FT status endpoint and forwards |
| 11 | software fault status updates to the Controller over HTTP. | 11 | software fault status updates to the Controller over HTTP. |
| 12 | """ | 12 | """ |
| 13 | 13 | ||
| 14 | +import json | ||
| 14 | import threading | 15 | import threading |
| 16 | +import time | ||
| 17 | +from pathlib import Path | ||
| 15 | 18 | ||
| 16 | -import zmq | 19 | +from motor.common.constants import ( |
| 17 | -import msgspec.msgpack | 20 | + ENGINE_STATUS_DEAD, |
| 18 | - | 21 | + ENGINE_STATUS_HEALTHY, |
| 22 | + ENGINE_STATUS_UNHEALTHY, | ||
| 23 | +) | ||
| 24 | +from motor.common.http.engine_ft_client import query_engine_ft_status | ||
| 19 | from motor.common.logger import get_logger | 25 | from motor.common.logger import get_logger |
| 26 | +from motor.common.logger.rate_limited_logger import RateLimitedLogger | ||
| 20 | from motor.common.resources.endpoint import Endpoint | 27 | from motor.common.resources.endpoint import Endpoint |
| 21 | -from motor.common.utils.net import format_address | 28 | +from motor.config.config_utils import ( |
| 29 | + ENGINE_CONFIG, | ||
| 30 | + MOTOR_ENGINE_PREFILL_CONFIG, | ||
| 31 | + MOTOR_ENGINE_UNION_CONFIG, | ||
| 32 | + ConfigKey, | ||
| 33 | +) | ||
| 22 | from motor.config.node_manager import NodeManagerConfig | 34 | from motor.config.node_manager import NodeManagerConfig |
| 23 | from motor.node_manager.api_client.controller_api_client import ControllerApiClient | 35 | from motor.node_manager.api_client.controller_api_client import ControllerApiClient |
| 24 | 36 | ||
| 25 | logger = get_logger(__name__) | 37 | logger = get_logger(__name__) |
| 38 | +_rl = RateLimitedLogger(logger) | ||
| 26 | 39 | ||
| 27 | -# ZMQ PUB topic used by vllm ClientSentinel to broadcast engine status | 40 | +# Engine FT enable keys accepted in the user config's engine_config sections |
| 28 | -_FAULT_STATE_PUB_TOPIC = "vllm_fault" | 41 | +# (vLLM style: --enable-fault-tolerance / --enable_fault_tolerance). |
| 42 | +_ENGINE_FT_ENABLE_KEYS = frozenset({"enable-fault-tolerance", "enable_fault_tolerance"}) | ||
| 29 | 43 | ||
| 30 | -# Map engine status names (from EngineStatusType enum in vllm) to int values | 44 | +# Engine sections of the user config that may carry an engine_config. |
| 45 | +_ENGINE_SECTION_KEYS = ( | ||
| 46 | + MOTOR_ENGINE_PREFILL_CONFIG, | ||
| 47 | + ConfigKey.MOTOR_ENGINE_DECODE.value, | ||
| 48 | + MOTOR_ENGINE_UNION_CONFIG, | ||
| 49 | +) | ||
| 50 | + | ||
| 51 | +# Map engine status names (from the vLLM /fault_tolerance/status response) | ||
| 52 | +# to int values consumed by the Controller. | ||
| 31 | _ENGINE_STATUS_NAME_TO_INT = { | 53 | _ENGINE_STATUS_NAME_TO_INT = { |
| 32 | - "healthy": 0, | 54 | + ENGINE_STATUS_HEALTHY: 0, |
| 33 | - "dead": 1, | 55 | + ENGINE_STATUS_DEAD: 1, |
| 34 | - "unhealthy": 2, | 56 | + ENGINE_STATUS_UNHEALTHY: 2, |
| 35 | } | 57 | } |
| 36 | 58 | ||
| 59 | +# Engines take minutes to load a model; poll failures during this startup | ||
| 60 | +# window must not be reported as dead. | ||
| 61 | +_STARTUP_GRACE_SEC = 300.0 | ||
| 62 | + | ||
| 63 | + | ||
| 64 | +def _engine_ft_enabled(user_config_path: str | None) -> bool: | ||
| 65 | + """Detect whether any engine section of the user config enables fault tolerance. | ||
| 66 | + | ||
| 67 | + Scans the known engine sections (``motor_engine_*_config``) of the user | ||
| 68 | + config for an FT enable key (``enable-fault-tolerance`` / | ||
| 69 | + ``enable_fault_tolerance``) set to true (JSON ``true`` or ``1``). | ||
| 70 | + """ | ||
| 71 | + if not user_config_path: | ||
| 72 | + return False | ||
| 73 | + path = Path(user_config_path) | ||
| 74 | + if not path.exists(): | ||
| 75 | + return False | ||
| 76 | + try: | ||
| 77 | + raw = json.loads(path.read_text(encoding="utf-8")) | ||
| 78 | + except (OSError, json.JSONDecodeError) as e: | ||
| 79 | + logger.warning("Failed to read user config %s for FT detection: %s", user_config_path, e) | ||
| 80 | + return False | ||
| 81 | + if not isinstance(raw, dict): | ||
| 82 | + return False | ||
| 83 | + | ||
| 84 | + for section_key in _ENGINE_SECTION_KEYS: | ||
| 85 | + section = raw.get(section_key) | ||
| 86 | + if not isinstance(section, dict): | ||
| 87 | + continue | ||
| 88 | + engine_config = section.get(ENGINE_CONFIG) | ||
| 89 | + if not isinstance(engine_config, dict): | ||
| 90 | + continue | ||
| 91 | + if any(engine_config.get(ft_key) in (True, 1) for ft_key in _ENGINE_FT_ENABLE_KEYS): | ||
| 92 | + return True | ||
| 93 | + return False | ||
| 94 | + | ||
| 37 | 95 | ||
| 38 | class FaultReporter: | 96 | class FaultReporter: |
| 39 | - """Subscribes to per-engine ZMQ PUB sockets and reports non-healthy | 97 | + """Polls per-engine FT status endpoints and reports non-healthy engines |
| 40 | - engines to the Controller. | 98 | + to the Controller. |
| 41 | 99 | ||
| 42 | - One ZMQ SUB socket is created for each EngineServer endpoint – the | 100 | + Each engine (a vLLM API server on the endpoint's business port) exposes |
| 43 | - EngineServer publishes on ``base_port + engine_id``. Status updates | 101 | + ``GET /fault_tolerance/status`` returning |
| 44 | - are msgpack-encoded and received asynchronously in a background thread. | 102 | + ``{"engines": [{"id", "status", "fault_info"?}]}``. Status is polled in a |
| 103 | + background thread every ``poll_interval_sec``. An engine that cannot be | ||
| 104 | + reached for ``max_poll_failures`` consecutive polls is reported as dead | ||
| 105 | + (after the startup grace period). | ||
| 106 | + | ||
| 107 | + Reporting is enabled when the NodeManager config explicitly enables fault | ||
| 108 | + tolerance OR any engine section of the user config does (auto-detection). | ||
| 45 | """ | 109 | """ |
| 46 | 110 | ||
| 47 | def __init__(self, config: NodeManagerConfig): | 111 | def __init__(self, config: NodeManagerConfig): |
| 48 | self._config = config | 112 | self._config = config |
| 49 | self._config_lock = threading.RLock() | 113 | self._config_lock = threading.RLock() |
| 50 | - self._enabled = config.fault_tolerance_config.enable_fault_tolerance | 114 | + self._enabled = self._compute_enabled(config) |
| 51 | self._thread: threading.Thread | None = None | 115 | self._thread: threading.Thread | None = None |
| 52 | self._stop_event = threading.Event() | 116 | self._stop_event = threading.Event() |
| 53 | self._endpoints: list[Endpoint] = [] | 117 | self._endpoints: list[Endpoint] = [] |
| 54 | 118 | ||
| 119 | + | ||
| 120 | + def _compute_enabled(config: NodeManagerConfig) -> bool: | ||
| 121 | + """Explicit config flag OR engine user config auto-detection.""" | ||
| 122 | + if config.fault_tolerance_config.enable_fault_tolerance: | ||
| 123 | + return True | ||
| 124 | + return _engine_ft_enabled(config.config_path) | ||
| 125 | + | ||
| 55 | def start(self, endpoints: list[Endpoint] | None = None) -> None: | 126 | def start(self, endpoints: list[Endpoint] | None = None) -> None: |
| 127 | + """Start the background polling thread (no-op when fault tolerance is disabled).""" | ||
| 56 | if not self._enabled: | 128 | if not self._enabled: |
| 57 | return | 129 | return |
| 58 | if self._thread is not None and self._thread.is_alive(): | 130 | if self._thread is not None and self._thread.is_alive(): |
| @@ -70,29 +142,39 @@ class FaultReporter: | |||
| 70 | logger.info("FaultReporter started.") | 142 | logger.info("FaultReporter started.") |
| 71 | 143 | ||
| 72 | def stop(self) -> None: | 144 | def stop(self) -> None: |
| 145 | + """Stop the polling thread, waiting through a full poll round for it to exit.""" | ||
| 73 | self._stop_event.set() | 146 | self._stop_event.set() |
| 74 | if self._thread is not None and self._thread.is_alive(): | 147 | if self._thread is not None and self._thread.is_alive(): |
| 75 | - self._thread.join(timeout=5.0) | 148 | + # One loop round can take len(endpoints) x poll_timeout when engines |
| 149 | + # are unreachable; wait through a full round so the loop observes | ||
| 150 | + # the stop event instead of the join timing out mid-round. | ||
| 151 | + with self._config_lock: | ||
| 152 | + poll_timeout = self._config.fault_tolerance_config.poll_timeout_sec | ||
| 153 | + round_time = len(self._endpoints) * poll_timeout | ||
| 154 | + join_timeout = max(5.0, round_time + 1.0) | ||
| 155 | + self._thread.join(timeout=join_timeout) | ||
| 76 | if self._thread.is_alive(): | 156 | if self._thread.is_alive(): |
| 77 | - logger.warning("FaultReporter thread did not stop within timeout") | 157 | + # Keep the reference so start() refuses to spawn a second |
| 158 | + # polling thread while the old one is still alive. | ||
| 159 | + logger.warning( | ||
| 160 | + "FaultReporter thread did not stop within timeout; %s", | ||
| 161 | + "keeping reference to avoid duplicate threads", | ||
| 162 | + ) | ||
| 163 | + return | ||
| 78 | self._thread = None | 164 | self._thread = None |
| 79 | logger.info("FaultReporter stopped.") | 165 | logger.info("FaultReporter stopped.") |
| 80 | 166 | ||
| 81 | def update_config(self, config: NodeManagerConfig, endpoints: list[Endpoint]) -> None: | 167 | def update_config(self, config: NodeManagerConfig, endpoints: list[Endpoint]) -> None: |
| 168 | + """Apply a new config and endpoint set, (re)starting or stopping the reporter as needed.""" | ||
| 82 | with self._config_lock: | 169 | with self._config_lock: |
| 83 | old_enable = self._enabled | 170 | old_enable = self._enabled |
| 84 | old_endpoint_ids = {ep.id for ep in self._endpoints} | 171 | old_endpoint_ids = {ep.id for ep in self._endpoints} |
| 85 | - old_pod_ip = self._config.api_config.pod_ip if self._endpoints else None | ||
| 86 | - old_zmq_port = self._config.fault_tolerance_config.zmq_pub_port | ||
| 87 | self._config = config | 172 | self._config = config |
| 88 | self._endpoints = endpoints | 173 | self._endpoints = endpoints |
| 89 | - self._enabled = config.fault_tolerance_config.enable_fault_tolerance | 174 | + self._enabled = self._compute_enabled(config) |
| 90 | 175 | ||
| 91 | new_endpoint_ids = {ep.id for ep in endpoints} | 176 | new_endpoint_ids = {ep.id for ep in endpoints} |
| 92 | endpoints_changed = old_endpoint_ids != new_endpoint_ids | 177 | endpoints_changed = old_endpoint_ids != new_endpoint_ids |
| 93 | - pod_ip_changed = old_pod_ip is not None and old_pod_ip != config.api_config.pod_ip | ||
| 94 | - zmq_port_changed = old_zmq_port != config.fault_tolerance_config.zmq_pub_port | ||
| 95 | - needs_restart = endpoints_changed or pod_ip_changed or zmq_port_changed | ||
| 96 | 178 | ||
| 97 | if self._enabled != old_enable: | 179 | if self._enabled != old_enable: |
| 98 | if self._enabled: | 180 | if self._enabled: |
| @@ -101,149 +183,169 @@ class FaultReporter: | |||
| 101 | else: | 183 | else: |
| 102 | self.stop() | 184 | self.stop() |
| 103 | logger.info("FaultReporter disabled, stopped thread") | 185 | logger.info("FaultReporter disabled, stopped thread") |
| 104 | - elif self._enabled and needs_restart: | 186 | + elif self._enabled and endpoints_changed: |
| 105 | - # ZMQ connection parameters changed while enabled — restart to rebuild sockets | 187 | + # Polling targets changed while enabled — restart to poll the new engines. |
| 106 | - changes = [] | ||
| 107 | - if endpoints_changed: | ||
| 108 | - changes.append(f"endpoints ({len(old_endpoint_ids)} -> {len(new_endpoint_ids)} engines)") | ||
| 109 | - if pod_ip_changed: | ||
| 110 | - changes.append(f"pod_ip ({old_pod_ip} -> {config.api_config.pod_ip})") | ||
| 111 | - if zmq_port_changed: | ||
| 112 | - changes.append(f"zmq_pub_port ({old_zmq_port} -> {config.fault_tolerance_config.zmq_pub_port})") | ||
| 113 | logger.info( | 188 | logger.info( |
| 114 | - "FaultReporter config changed (%s), restarting to rebuild ZMQ subscriptions", | 189 | + "FaultReporter endpoints changed (%d -> %d), restarting", |
| 115 | - ", ".join(changes), | 190 | + len(old_endpoint_ids), |
| 191 | + len(new_endpoint_ids), | ||
| 116 | ) | 192 | ) |
| 117 | self.stop() | 193 | self.stop() |
| 118 | self.start() | 194 | self.start() |
| 119 | 195 | ||
| 120 | - def _setup_zmq_sub_sockets(self) -> tuple[list, zmq.Poller | None, zmq.Context | None]: | 196 | + def _poll_interval_sec(self) -> float: |
| 121 | - """Create one ZMQ SUB socket per engine endpoint.""" | 197 | + with self._config_lock: |
| 122 | - sub_sockets: list = [] | 198 | + return self._config.fault_tolerance_config.poll_interval_sec |
| 123 | - zmq_ctx = None | ||
| 124 | - try: | ||
| 125 | - with self._config_lock: | ||
| 126 | - pod_ip = self._config.api_config.pod_ip | ||
| 127 | - base_port = self._config.fault_tolerance_config.zmq_pub_port | ||
| 128 | 199 | ||
| 129 | - if base_port > 0 and len(self._endpoints) > 0: | 200 | + def _max_poll_failures(self) -> int: |
| 130 | - zmq_ctx = zmq.Context() | 201 | + with self._config_lock: |
| 131 | - for ep in self._endpoints: | 202 | + return self._config.fault_tolerance_config.max_poll_failures |
| 132 | - port = base_port + ep.id | ||
| 133 | - sub = zmq_ctx.socket(zmq.SUB) | ||
| 134 | - sub.setsockopt(zmq.RECONNECT_IVL, 5000) | ||
| 135 | - zmq_addr = f"tcp://{format_address(pod_ip, port)}" | ||
| 136 | - sub.connect(zmq_addr) | ||
| 137 | - sub.setsockopt_string(zmq.SUBSCRIBE, _FAULT_STATE_PUB_TOPIC) | ||
| 138 | - sub_sockets.append(sub) | ||
| 139 | - logger.info( | ||
| 140 | - "ZMQ SUB connected to %s for engine %d", | ||
| 141 | - zmq_addr, | ||
| 142 | - ep.id, | ||
| 143 | - ) | ||
| 144 | - else: | ||
| 145 | - logger.info( | ||
| 146 | - "ZMQ fault pub port=%d, endpoints=%d, ZMQ subscription disabled", | ||
| 147 | - base_port, | ||
| 148 | - len(self._endpoints), | ||
| 149 | - ) | ||
| 150 | - except Exception as e: | ||
| 151 | - logger.warning("Failed to set up ZMQ SUB sockets: %s, ZMQ subscription disabled", e) | ||
| 152 | 203 | ||
| 153 | - poller = None | 204 | + def _query_engine_status(self, ep: Endpoint) -> dict: |
| 154 | - if sub_sockets: | 205 | + """GET one engine's FT status payload from its business (API) port.""" |
| 155 | - poller = zmq.Poller() | 206 | + with self._config_lock: |
| 156 | - for sub in sub_sockets: | 207 | + timeout = self._config.fault_tolerance_config.poll_timeout_sec |
| 157 | - poller.register(sub, zmq.POLLIN) | 208 | + return query_engine_ft_status(ep, timeout) |
| 158 | - return sub_sockets, poller, zmq_ctx | ||
| 159 | - | ||
| 160 | - | ||
| 161 | - def _teardown_zmq_sub_sockets( | ||
| 162 | - sub_sockets: list, | ||
| 163 | - zmq_ctx: zmq.Context | None, | ||
| 164 | - ) -> None: | ||
| 165 | - for sub in sub_sockets: | ||
| 166 | - sub.close() | ||
| 167 | - if zmq_ctx: | ||
| 168 | - zmq_ctx.term() | ||
| 169 | - | ||
| 170 | - _ZMQ_RECONNECT_DELAY = 5.0 # seconds to wait before retrying ZMQ setup after error | ||
| 171 | 209 | ||
| 172 | def _main_loop(self) -> None: | 210 | def _main_loop(self) -> None: |
| 173 | - """Subscribe to ZMQ PUB sockets and forward faults to Controller.""" | 211 | + """Poll every engine's FT status and forward faults to Controller.""" |
| 174 | logger.info("FaultReporter loop started.") | 212 | logger.info("FaultReporter loop started.") |
| 175 | known_statuses: dict[int, str] = {} | 213 | known_statuses: dict[int, str] = {} |
| 176 | - | 214 | + consecutive_failures: dict[int, int] = {} |
| 177 | - sub_sockets, poller, zmq_ctx = self._setup_zmq_sub_sockets() | 215 | + first_poll_time: dict[int, float] = {} |
| 178 | 216 | ||
| 179 | while not self._stop_event.is_set(): | 217 | while not self._stop_event.is_set(): |
| 180 | - if poller: | 218 | + with self._config_lock: |
| 181 | - try: | 219 | + endpoints = list(self._endpoints) |
| 182 | - socks = dict(poller.poll(timeout=500)) | 220 | + now = time.time() |
| 183 | - for sub in sub_sockets: | 221 | + for ep in endpoints: |
| 184 | - if sub in socks: | 222 | + first_poll_time.setdefault(ep.id, now) |
| 185 | - topic, raw = sub.recv_multipart() | 223 | + self._poll_engine(ep, known_statuses, consecutive_failures, first_poll_time) |
| 186 | - self._process_zmq_engine_status(raw, known_statuses) | 224 | + if self._stop_event.wait(self._poll_interval_sec()): |
| 187 | - except zmq.ZMQError: | 225 | + break |
| 188 | - logger.warning( | ||
| 189 | - "ZMQ error in poll loop, tearing down and reconnecting in %.1fs ...", | ||
| 190 | - self._ZMQ_RECONNECT_DELAY, | ||
| 191 | - ) | ||
| 192 | - self._teardown_zmq_sub_sockets(sub_sockets, zmq_ctx) | ||
| 193 | - # Wait before retry, respecting stop_event | ||
| 194 | - if self._stop_event.wait(timeout=self._ZMQ_RECONNECT_DELAY): | ||
| 195 | - break | ||
| 196 | - logger.info("Reconnecting ZMQ SUB sockets ...") | ||
| 197 | - sub_sockets, poller, zmq_ctx = self._setup_zmq_sub_sockets() | ||
| 198 | - except Exception as e: | ||
| 199 | - logger.error("Error processing ZMQ engine status: %s", e) | ||
| 200 | - else: | ||
| 201 | - # No sockets — avoid busy-wait; wait for stop or config change | ||
| 202 | - if self._stop_event.wait(timeout=1.0): | ||
| 203 | - break | ||
| 204 | - # Retry setup in case endpoints/port are now available | ||
| 205 | - sub_sockets, poller, zmq_ctx = self._setup_zmq_sub_sockets() | ||
| 206 | 226 | ||
| 207 | - self._teardown_zmq_sub_sockets(sub_sockets, zmq_ctx) | ||
| 208 | logger.info("FaultReporter loop stopped.") | 227 | logger.info("FaultReporter loop stopped.") |
| 209 | 228 | ||
| 210 | - def _process_zmq_engine_status( | 229 | + def _poll_engine( |
| 211 | self, | 230 | self, |
| 212 | - raw: bytes, | 231 | + ep: Endpoint, |
| 232 | + known_statuses: dict[int, str], | ||
| 233 | + consecutive_failures: dict[int, int], | ||
| 234 | + first_poll_time: dict[int, float], | ||
| 235 | + ) -> None: | ||
| 236 | + """Poll a single engine: forward new non-healthy statuses, or count | ||
| 237 | + poll failures and report dead once the threshold is exceeded. | ||
| 238 | + """ | ||
| 239 | + try: | ||
| 240 | + payload = self._query_engine_status(ep) | ||
| 241 | + except Exception as e: | ||
| 242 | + # A poll failure must never kill the polling thread — count it | ||
| 243 | + # and let the consecutive-failures threshold decide the engine's fate. | ||
| 244 | + failures = consecutive_failures.get(ep.id, 0) + 1 | ||
| 245 | + consecutive_failures[ep.id] = failures | ||
| 246 | + _rl.error_window( | ||
| 247 | + f"node_manager.fault_reporter.poll.{ep.id}", | ||
| 248 | + f"Failed to poll engine {ep.id} FT status: {e}", | ||
| 249 | + ) | ||
| 250 | + if failures >= self._max_poll_failures(): | ||
| 251 | + self._report_unreachable_dead(ep, failures, known_statuses, first_poll_time) | ||
| 252 | + return | ||
| 253 | + | ||
| 254 | + consecutive_failures[ep.id] = 0 | ||
| 255 | + try: | ||
| 256 | + if not isinstance(payload, dict): | ||
| 257 | + raise TypeError(f"unexpected FT status payload type: {type(payload).__name__}") | ||
| 258 | + for engine in payload.get("engines", []): | ||
| 259 | + self._process_engine_status(ep.id, engine, known_statuses) | ||
| 260 | + except Exception as e: | ||
| 261 | + # A malformed payload must never kill the polling thread — log and | ||
| 262 | + # continue with the next round. | ||
| 263 | + _rl.error_window( | ||
| 264 | + f"node_manager.fault_reporter.parse.{ep.id}", | ||
| 265 | + f"Failed to parse engine {ep.id} FT status: {e}", | ||
| 266 | + ) | ||
| 267 | + | ||
| 268 | + def _process_engine_status( | ||
| 269 | + self, | ||
| 270 | + ep_id: int, | ||
| 271 | + engine: dict, | ||
| 213 | known_statuses: dict[int, str], | 272 | known_statuses: dict[int, str], |
| 214 | ) -> None: | 273 | ) -> None: |
| 215 | - """Decode a ZMQ PUB message and report new non-healthy engines.""" | 274 | + """Report a single engine's status if it is non-healthy and new. |
| 216 | - msg = msgspec.msgpack.decode(raw) | ||
| 217 | - engines = msg.get("engines", []) | ||
| 218 | 275 | ||
| 219 | - for engine in engines: | 276 | + The dedup key is the managed endpoint id (``ep_id``) — the same |
| 220 | - engine_id = engine["id"] | 277 | + namespace used by ``_report_unreachable_dead`` — not the payload's |
| 221 | - status = engine["status"] | 278 | + engine id, which is rank-local per API server and collides across |
| 279 | + endpoints. | ||
| 280 | + """ | ||
| 281 | + if not isinstance(engine, dict): | ||
| 282 | + raise TypeError(f"engine entry is not a dict: {engine!r}") | ||
| 283 | + status = engine.get("status") | ||
| 284 | + if not isinstance(status, str): | ||
| 285 | + raise TypeError(f"engine entry of endpoint {ep_id} has no valid status") | ||
| 222 | 286 | ||
| 223 | - if status == "healthy": | 287 | + if status == ENGINE_STATUS_HEALTHY: |
| 224 | - known_statuses[engine_id] = status | 288 | + known_statuses[ep_id] = status |
| 225 | - continue | 289 | + return |
| 226 | 290 | ||
| 227 | - if known_statuses.get(engine_id) == status: | 291 | + if known_statuses.get(ep_id) == status: |
| 228 | - continue # already reported | 292 | + return # already reported |
| 229 | 293 | ||
| 230 | - engine_status = _ENGINE_STATUS_NAME_TO_INT.get(status) | 294 | + engine_status = _ENGINE_STATUS_NAME_TO_INT.get(status) |
| 231 | - if engine_status is None: | 295 | + if engine_status is None: |
| 232 | - logger.warning("Unknown engine status '%s' for engine %d", status, engine_id) | 296 | + logger.warning("Unknown engine status '%s' for engine %d", status, ep_id) |
| 233 | - continue | 297 | + return |
| 234 | 298 | ||
| 235 | - exception_type = "EngineDeadError" if engine_status == 1 else "EngineUnhealthyError" | 299 | + fault_info = engine.get("fault_info") or "" |
| 236 | - exception_message = "Engine process died" if engine_status == 1 else "Engine unhealthy" | 300 | + if status == "unhealthy" and fault_info: |
| 301 | + exception_type = fault_info | ||
| 302 | + exception_message = f"Engine unhealthy: {fault_info}" | ||
| 303 | + elif status == ENGINE_STATUS_DEAD: | ||
| 304 | + exception_type = "EngineDeadError" | ||
| 305 | + exception_message = "Engine process died" | ||
| 306 | + else: | ||
| 307 | + exception_type = "EngineUnhealthyError" | ||
| 308 | + exception_message = "Engine unhealthy" | ||
| 237 | 309 | ||
| 238 | - fault_data = { | 310 | + fault_data = { |
| 239 | - "exception_type": exception_type, | 311 | + "exception_type": exception_type, |
| 240 | - "exception_message": exception_message, | 312 | + "exception_message": exception_message, |
| 241 | - "engine_id": engine_id, | 313 | + "engine_id": ep_id, |
| 242 | - "engine_status": engine_status, | 314 | + "engine_status": engine_status, |
| 243 | - } | 315 | + } |
| 244 | - # Only mark as reported after successful delivery to Controller | 316 | + # Only mark as reported after successful delivery to Controller |
| 245 | - if self._send_fault_to_controller(fault_data): | 317 | + if self._send_fault_to_controller(fault_data): |
| 246 | - known_statuses[engine_id] = status | 318 | + known_statuses[ep_id] = status |
| 319 | + | ||
| 320 | + def _report_unreachable_dead( | ||
| 321 | + self, | ||
| 322 | + ep: Endpoint, | ||
| 323 | + failures: int, | ||
| 324 | + known_statuses: dict[int, str], | ||
| 325 | + first_poll_time: dict[int, float], | ||
| 326 | + ) -> None: | ||
| 327 | + """Report an engine as dead after repeated poll failures (deduped). | ||
| 328 | + | ||
| 329 | + Poll failures during the startup grace period (engine model load) | ||
| 330 | + are not reported as dead. | ||
| 331 | + """ | ||
| 332 | + if known_statuses.get(ep.id) == ENGINE_STATUS_DEAD: | ||
| 333 | + return | ||
| 334 | + first_poll = first_poll_time.get(ep.id, time.time()) | ||
| 335 | + if time.time() - first_poll < _STARTUP_GRACE_SEC: | ||
| 336 | + logger.debug( | ||
| 337 | + "Engine %d unreachable but within startup grace period, not reporting dead", | ||
| 338 | + ep.id, | ||
| 339 | + ) | ||
| 340 | + return | ||
| 341 | + fault_data = { | ||
| 342 | + "exception_type": "EngineDeadError", | ||
| 343 | + "exception_message": f"Engine unreachable after {failures} consecutive polls", | ||
| 344 | + "engine_id": ep.id, | ||
| 345 | + "engine_status": 1, | ||
| 346 | + } | ||
| 347 | + if self._send_fault_to_controller(fault_data): | ||
| 348 | + known_statuses[ep.id] = ENGINE_STATUS_DEAD | ||
| 247 | 349 | ||
| 248 | def _send_fault_to_controller(self, fault_data: dict) -> bool: | 350 | def _send_fault_to_controller(self, fault_data: dict) -> bool: |
| 249 | """Inject pod_ip and forward a single fault to Controller. | 351 | """Inject pod_ip and forward a single fault to Controller. |
| @@ -9,29 +9,112 @@ | |||
| 9 | # See the Mulan PSL v2 for more details. | 9 | # See the Mulan PSL v2 for more details. |
| 10 | """Tests for motor.node_manager.core.fault_reporter.""" | 10 | """Tests for motor.node_manager.core.fault_reporter.""" |
| 11 | 11 | ||
| 12 | +import json | ||
| 12 | import os | 13 | import os |
| 13 | import sys | 14 | import sys |
| 14 | 15 | ||
| 15 | import pytest | 16 | import pytest |
| 16 | -from unittest.mock import patch, MagicMock | 17 | +from unittest.mock import patch |
| 17 | 18 | ||
| 18 | os.environ["USER_CONFIG_PATH"] = "tests/jsons/useruser_config.json" | 19 | os.environ["USER_CONFIG_PATH"] = "tests/jsons/useruser_config.json" |
| 19 | os.environ["ROLE"] = "both" | 20 | os.environ["ROLE"] = "both" |
| 20 | sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) | 21 | sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) |
| 21 | 22 | ||
| 22 | -from motor.node_manager.core.fault_reporter import FaultReporter | 23 | +from motor.node_manager.core.fault_reporter import FaultReporter, _engine_ft_enabled |
| 23 | from motor.config.node_manager import NodeManagerConfig | 24 | from motor.config.node_manager import NodeManagerConfig |
| 24 | from motor.common.resources.endpoint import Endpoint | 25 | from motor.common.resources.endpoint import Endpoint |
| 25 | 26 | ||
| 26 | # pylint: disable=redefined-outer-name,duplicate-code | 27 | # pylint: disable=redefined-outer-name,duplicate-code |
| 27 | 28 | ||
| 28 | 29 | ||
| 30 | +# -- engine FT auto-detection -------------------------------------------------- | ||
| 31 | + | ||
| 32 | + | ||
| 33 | +def _write_user_config(tmp_path, content: dict) -> str: | ||
| 34 | + path = tmp_path / "user_config.json" | ||
| 35 | + path.write_text(json.dumps(content), encoding="utf-8") | ||
| 36 | + return str(path) | ||
| 37 | + | ||
| 38 | + | ||
| 39 | +def test_engine_ft_enabled_none_path(): | ||
| 40 | + assert _engine_ft_enabled(None) is False | ||
| 41 | + | ||
| 42 | + | ||
| 43 | +def test_engine_ft_enabled_missing_file(tmp_path): | ||
| 44 | + assert _engine_ft_enabled(str(tmp_path / "nope.json")) is False | ||
| 45 | + | ||
| 46 | + | ||
| 47 | +def test_engine_ft_enabled_no_ft_key(tmp_path): | ||
| 48 | + path = _write_user_config(tmp_path, {"motor_engine_prefill_config": {"engine_config": {}}}) | ||
| 49 | + assert _engine_ft_enabled(path) is False | ||
| 50 | + | ||
| 51 | + | ||
| 52 | +def test_engine_ft_enabled_snake_case_key(tmp_path): | ||
| 53 | + path = _write_user_config( | ||
| 54 | + tmp_path, | ||
| 55 | + {"motor_engine_prefill_config": {"engine_config": {"enable_fault_tolerance": True}}}, | ||
| 56 | + ) | ||
| 57 | + assert _engine_ft_enabled(path) is True | ||
| 58 | + | ||
| 59 | + | ||
| 60 | +def test_engine_ft_enabled_hyphen_key(tmp_path): | ||
| 61 | + path = _write_user_config( | ||
| 62 | + tmp_path, | ||
| 63 | + {"motor_engine_decode_config": {"engine_config": {"enable-fault-tolerance": True}}}, | ||
| 64 | + ) | ||
| 65 | + assert _engine_ft_enabled(path) is True | ||
| 66 | + | ||
| 67 | + | ||
| 68 | +def test_engine_ft_enabled_false_value(tmp_path): | ||
| 69 | + path = _write_user_config( | ||
| 70 | + tmp_path, | ||
| 71 | + {"motor_engine_prefill_config": {"engine_config": {"enable_fault_tolerance": False}}}, | ||
| 72 | + ) | ||
| 73 | + assert _engine_ft_enabled(path) is False | ||
| 74 | + | ||
| 75 | + | ||
| 76 | +def test_engine_ft_enabled_broken_json(tmp_path): | ||
| 77 | + path = tmp_path / "user_config.json" | ||
| 78 | + path.write_text("{not json", encoding="utf-8") | ||
| 79 | + assert _engine_ft_enabled(str(path)) is False | ||
| 80 | + | ||
| 81 | + | ||
| 82 | +# -- auto-enable without explicit config --------------------------------------- | ||
| 83 | + | ||
| 84 | + | ||
| 85 | +def test_start_auto_enabled_via_engine_config(tmp_path, endpoints): | ||
| 86 | + """No explicit flag needed: FT in the engine user config enables reporting.""" | ||
| 87 | + cfg = NodeManagerConfig() | ||
| 88 | + cfg.api_config.pod_ip = "192.168.1.1" | ||
| 89 | + cfg.fault_tolerance_config.enable_fault_tolerance = False | ||
| 90 | + cfg.config_path = _write_user_config( | ||
| 91 | + tmp_path, | ||
| 92 | + {"motor_engine_prefill_config": {"engine_config": {"enable-fault-tolerance": True}}}, | ||
| 93 | + ) | ||
| 94 | + r = FaultReporter(cfg) | ||
| 95 | + r.start(endpoints) | ||
| 96 | + assert r._thread is not None | ||
| 97 | + r.stop() | ||
| 98 | + | ||
| 99 | + | ||
| 100 | +def test_start_not_enabled_without_engine_ft(tmp_path, endpoints): | ||
| 101 | + cfg = NodeManagerConfig() | ||
| 102 | + cfg.api_config.pod_ip = "192.168.1.1" | ||
| 103 | + cfg.fault_tolerance_config.enable_fault_tolerance = False | ||
| 104 | + cfg.config_path = _write_user_config(tmp_path, {"motor_engine_prefill_config": {"engine_config": {}}}) | ||
| 105 | + r = FaultReporter(cfg) | ||
| 106 | + r.start(endpoints) | ||
| 107 | + assert r._thread is None | ||
| 108 | + | ||
| 109 | + | ||
| 29 | 110 | ||
| 30 | def config(): | 111 | def config(): |
| 31 | cfg = NodeManagerConfig() | 112 | cfg = NodeManagerConfig() |
| 32 | cfg.api_config.pod_ip = "192.168.1.1" | 113 | cfg.api_config.pod_ip = "192.168.1.1" |
| 33 | cfg.fault_tolerance_config.enable_fault_tolerance = True | 114 | cfg.fault_tolerance_config.enable_fault_tolerance = True |
| 34 | - cfg.fault_tolerance_config.zmq_pub_port = 0 | 115 | + # Endpoints are unreachable in the test env (connect timeout, not refused): |
| 116 | + # a short poll timeout keeps one loop round well under stop()'s join(5s). | ||
| 117 | + cfg.fault_tolerance_config.poll_timeout_sec = 0.1 | ||
| 35 | return cfg | 118 | return cfg |
| 36 | 119 | ||
| 37 | 120 | ||
| @@ -97,350 +180,21 @@ def test_stop_joins_thread(reporter, endpoints): | |||
| 97 | assert reporter._thread is None | 180 | assert reporter._thread is None |
| 98 | 181 | ||
| 99 | 182 | ||
| 100 | -# -- ZMQ setup ----------------------------------------------------------------- | ||
| 101 | - | ||
| 102 | - | ||
| 103 | - | ||
| 104 | -def test_setup_zmq_multi(mock_zmq, config, endpoints): | ||
| 105 | - import zmq as real_zmq | ||
| 106 | - | ||
| 107 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 108 | - r = FaultReporter(config) | ||
| 109 | - r._endpoints = endpoints | ||
| 110 | - | ||
| 111 | - mock_ctx_cls = MagicMock() | ||
| 112 | - mock_ctx_instance = mock_ctx_cls.return_value | ||
| 113 | - mock_sub = MagicMock() | ||
| 114 | - mock_ctx_instance.socket.return_value = mock_sub | ||
| 115 | - mock_zmq.Context = mock_ctx_cls | ||
| 116 | - mock_zmq.SUB = real_zmq.SUB | ||
| 117 | - mock_zmq.Poller.return_value = MagicMock() | ||
| 118 | - | ||
| 119 | - sub_sockets, poller, _ = r._setup_zmq_sub_sockets() | ||
| 120 | - | ||
| 121 | - mock_ctx_cls.assert_called_once() | ||
| 122 | - assert mock_ctx_instance.socket.call_count == 2 | ||
| 123 | - mock_sub.connect.assert_any_call("tcp://192.168.1.1:5555") | ||
| 124 | - mock_sub.connect.assert_any_call("tcp://192.168.1.1:5556") | ||
| 125 | - assert len(sub_sockets) == 2 | ||
| 126 | - assert poller is not None | ||
| 127 | - | ||
| 128 | - | ||
| 129 | - | ||
| 130 | -def test_setup_zmq_ipv6_bracketed_url(mock_zmq, config, endpoints): | ||
| 131 | - import zmq as real_zmq | ||
| 132 | - | ||
| 133 | - config.api_config.pod_ip = "2001:db8::1" | ||
| 134 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 135 | - r = FaultReporter(config) | ||
| 136 | - r._endpoints = endpoints[:1] | ||
| 137 | - | ||
| 138 | - mock_ctx_cls = MagicMock() | ||
| 139 | - mock_ctx_instance = mock_ctx_cls.return_value | ||
| 140 | - mock_sub = MagicMock() | ||
| 141 | - mock_ctx_instance.socket.return_value = mock_sub | ||
| 142 | - mock_zmq.Context = mock_ctx_cls | ||
| 143 | - mock_zmq.SUB = real_zmq.SUB | ||
| 144 | - mock_zmq.Poller.return_value = MagicMock() | ||
| 145 | - | ||
| 146 | - sub_sockets, poller, _ = r._setup_zmq_sub_sockets() | ||
| 147 | - | ||
| 148 | - mock_sub.connect.assert_called_once_with("tcp://[2001:db8::1]:5555") | ||
| 149 | - assert len(sub_sockets) == 1 | ||
| 150 | - assert poller is not None | ||
| 151 | - | ||
| 152 | - | ||
| 153 | - | ||
| 154 | -def test_setup_zmq_no_port(mock_zmq, config, endpoints): | ||
| 155 | - r = FaultReporter(config) | ||
| 156 | - r._endpoints = endpoints | ||
| 157 | - sub_sockets, poller, zmq_ctx = r._setup_zmq_sub_sockets() | ||
| 158 | - assert len(sub_sockets) == 0 | ||
| 159 | - assert poller is None | ||
| 160 | - assert zmq_ctx is None | ||
| 161 | - | ||
| 162 | - | ||
| 163 | - | ||
| 164 | -def test_setup_zmq_no_endpoints(mock_zmq, config): | ||
| 165 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 166 | - r = FaultReporter(config) | ||
| 167 | - sub_sockets, poller, _ = r._setup_zmq_sub_sockets() | ||
| 168 | - assert len(sub_sockets) == 0 | ||
| 169 | - | ||
| 170 | - | ||
| 171 | -# -- ZMQ processing ------------------------------------------------------------ | ||
| 172 | - | ||
| 173 | - | ||
| 174 | - | ||
| 175 | -def test_process_zmq_dead(mock_report, reporter): | ||
| 176 | - import msgspec.msgpack | ||
| 177 | - | ||
| 178 | - msg = { | ||
| 179 | - "schema_version": 1, | ||
| 180 | - "total_engines": 2, | ||
| 181 | - "engines": [{"id": 0, "status": "dead"}, {"id": 1, "status": "healthy"}], | ||
| 182 | - } | ||
| 183 | - raw = msgspec.msgpack.encode(msg) | ||
| 184 | - known = {} | ||
| 185 | - reporter._process_zmq_engine_status(raw, known) | ||
| 186 | - mock_report.assert_called_once() | ||
| 187 | - called = mock_report.call_args[0][0] | ||
| 188 | - assert called["engine_id"] == 0 | ||
| 189 | - assert called["engine_status"] == 1 | ||
| 190 | - assert known == {0: "dead", 1: "healthy"} | ||
| 191 | - | ||
| 192 | - | ||
| 193 | - | ||
| 194 | -def test_process_zmq_dedup(mock_report, reporter): | ||
| 195 | - import msgspec.msgpack | ||
| 196 | - | ||
| 197 | - msg = {"schema_version": 1, "total_engines": 1, "engines": [{"id": 0, "status": "dead"}]} | ||
| 198 | - raw = msgspec.msgpack.encode(msg) | ||
| 199 | - known = {0: "dead"} | ||
| 200 | - reporter._process_zmq_engine_status(raw, known) | ||
| 201 | - mock_report.assert_not_called() | ||
| 202 | - | ||
| 203 | - | ||
| 204 | - | ||
| 205 | -def test_process_zmq_healthy(mock_report, reporter): | ||
| 206 | - import msgspec.msgpack | ||
| 207 | - | ||
| 208 | - msg = {"schema_version": 1, "total_engines": 1, "engines": [{"id": 0, "status": "healthy"}]} | ||
| 209 | - raw = msgspec.msgpack.encode(msg) | ||
| 210 | - known = {} | ||
| 211 | - reporter._process_zmq_engine_status(raw, known) | ||
| 212 | - mock_report.assert_not_called() | ||
| 213 | - assert known == {0: "healthy"} | ||
| 214 | - | ||
| 215 | - | ||
| 216 | - | ||
| 217 | -def test_send_fault_injects_pod_ip(mock_report, reporter): | ||
| 218 | - fault = {"exception_type": "KeyError", "engine_id": 1, "engine_status": 2} | ||
| 219 | - reporter._send_fault_to_controller(fault) | ||
| 220 | - mock_report.assert_called_once() | ||
| 221 | - assert mock_report.call_args[0][0]["pod_ip"] == "192.168.1.1" | ||
| 222 | - | ||
| 223 | - | ||
| 224 | -# -- Main Loop ---------------------------------------------------------------------- | ||
| 225 | - | ||
| 226 | - | ||
| 227 | - | ||
| 228 | - | ||
| 229 | -def test_main_loop_multi_socket(mock_zmq, mock_report, config, endpoints): | ||
| 230 | - import zmq as real_zmq | ||
| 231 | - import msgspec.msgpack | ||
| 232 | - | ||
| 233 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 234 | - r = FaultReporter(config) | ||
| 235 | - r._endpoints = endpoints | ||
| 236 | - | ||
| 237 | - msg_dead = msgspec.msgpack.encode( | ||
| 238 | - { | ||
| 239 | - "schema_version": 1, | ||
| 240 | - "total_engines": 1, | ||
| 241 | - "engines": [{"id": 0, "status": "dead"}], | ||
| 242 | - } | ||
| 243 | - ) | ||
| 244 | - msg_uh = msgspec.msgpack.encode( | ||
| 245 | - { | ||
| 246 | - "schema_version": 1, | ||
| 247 | - "total_engines": 1, | ||
| 248 | - "engines": [{"id": 1, "status": "unhealthy"}], | ||
| 249 | - } | ||
| 250 | - ) | ||
| 251 | - | ||
| 252 | - sub0 = MagicMock() | ||
| 253 | - sub0.recv_multipart.return_value = (b"vllm_fault", msg_dead) | ||
| 254 | - sub1 = MagicMock() | ||
| 255 | - sub1.recv_multipart.return_value = (b"vllm_fault", msg_uh) | ||
| 256 | - | ||
| 257 | - mock_ctx_inst = MagicMock() | ||
| 258 | - mock_ctx_inst.socket.side_effect = [sub0, sub1] | ||
| 259 | - mock_zmq.Context.return_value = mock_ctx_inst | ||
| 260 | - mock_zmq.SUB = real_zmq.SUB | ||
| 261 | - | ||
| 262 | - mock_poller = MagicMock() | ||
| 263 | - mock_zmq.Poller.return_value = mock_poller | ||
| 264 | - | ||
| 265 | - cnt = [0] | ||
| 266 | - | ||
| 267 | - def stop_after(): | ||
| 268 | - def side_effect(*a, **kw): | ||
| 269 | - cnt[0] += 1 | ||
| 270 | - if cnt[0] >= 2: | ||
| 271 | - r._stop_event.set() | ||
| 272 | - return [{sub0: real_zmq.POLLIN}, {sub1: real_zmq.POLLIN}][cnt[0] - 1] | ||
| 273 | - | ||
| 274 | - return side_effect | ||
| 275 | - | ||
| 276 | - mock_poller.poll.side_effect = stop_after() | ||
| 277 | - | ||
| 278 | - r._main_loop() | ||
| 279 | - | ||
| 280 | - assert mock_poller.register.call_count == 2 | ||
| 281 | - assert mock_report.call_count == 2 | ||
| 282 | - assert mock_report.call_args_list[0][0][0]["engine_id"] == 0 | ||
| 283 | - assert mock_report.call_args_list[0][0][0]["engine_status"] == 1 | ||
| 284 | - assert mock_report.call_args_list[1][0][0]["engine_id"] == 1 | ||
| 285 | - assert mock_report.call_args_list[1][0][0]["engine_status"] == 2 | ||
| 286 | - | ||
| 287 | - | ||
| 288 | -# -- ZMQ retry on error -------------------------------------------------------- | ||
| 289 | - | ||
| 290 | - | ||
| 291 | - | ||
| 292 | - | ||
| 293 | -def test_main_loop_retry_after_zmq_error(mock_zmq, mock_report, config, endpoints): | ||
| 294 | - """When ZMQError occurs during poll, the loop tears down old sockets, | ||
| 295 | - reconnects, and continues processing — instead of exiting. | ||
| 296 | - """ | ||
| 297 | - import zmq as real_zmq | ||
| 298 | - import msgspec.msgpack | ||
| 299 | - | ||
| 300 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 301 | - r = FaultReporter(config) | ||
| 302 | - r._endpoints = endpoints | ||
| 303 | - r._ZMQ_RECONNECT_DELAY = 0.0 # skip wait in test | ||
| 304 | - | ||
| 305 | - msg_dead = msgspec.msgpack.encode( | ||
| 306 | - {"schema_version": 1, "total_engines": 1, "engines": [{"id": 0, "status": "dead"}]} | ||
| 307 | - ) | ||
| 308 | - | ||
| 309 | - # First-round mocks: poller raises ZMQError | ||
| 310 | - old_poller = MagicMock() | ||
| 311 | - old_poller.poll.side_effect = real_zmq.ZMQError("connection lost") | ||
| 312 | - old_sub = MagicMock() | ||
| 313 | - old_ctx = MagicMock() | ||
| 314 | - | ||
| 315 | - # Second-round mocks: poller processes one message, then stops | ||
| 316 | - new_poller = MagicMock() | ||
| 317 | - new_sub = MagicMock() | ||
| 318 | - new_sub.recv_multipart.return_value = (b"vllm_fault", msg_dead) | ||
| 319 | - new_ctx = MagicMock() | ||
| 320 | - | ||
| 321 | - call_count = [0] | ||
| 322 | - | ||
| 323 | - def poll_side_effect(*a, **kw): | ||
| 324 | - call_count[0] += 1 | ||
| 325 | - if call_count[0] >= 2: | ||
| 326 | - r._stop_event.set() | ||
| 327 | - return {new_sub: real_zmq.POLLIN} | ||
| 328 | - | ||
| 329 | - new_poller.poll.side_effect = poll_side_effect | ||
| 330 | - | ||
| 331 | - # _setup_zmq_sub_sockets → first returns old mocks, then new mocks | ||
| 332 | - mock_zmq.Context.return_value = old_ctx | ||
| 333 | - old_ctx.socket.return_value = old_sub | ||
| 334 | - mock_zmq.SUB = real_zmq.SUB | ||
| 335 | - mock_zmq.ZMQError = real_zmq.error.ZMQError # pin to real exception class | ||
| 336 | - mock_zmq.Poller.return_value = old_poller | ||
| 337 | - | ||
| 338 | - # After first teardown + retry, switch to new mocks | ||
| 339 | - orig_setup = r._setup_zmq_sub_sockets | ||
| 340 | - setup_count = [0] | ||
| 341 | - | ||
| 342 | - def setup_side_effect(): | ||
| 343 | - setup_count[0] += 1 | ||
| 344 | - if setup_count[0] == 1: | ||
| 345 | - # First call: return old mocks (already configured via mock_zmq) | ||
| 346 | - sub_sockets, poller, ctx = orig_setup() | ||
| 347 | - poller.poll.side_effect = real_zmq.ZMQError("connection lost") | ||
| 348 | - return sub_sockets, poller, ctx | ||
| 349 | - else: | ||
| 350 | - # Retry call: return new mocks | ||
| 351 | - mock_zmq.Context.return_value = new_ctx | ||
| 352 | - new_ctx.socket.return_value = new_sub | ||
| 353 | - mock_zmq.Poller.return_value = new_poller | ||
| 354 | - return orig_setup() | ||
| 355 | - | ||
| 356 | - with patch.object(r, "_setup_zmq_sub_sockets", side_effect=setup_side_effect): | ||
| 357 | - r._main_loop() | ||
| 358 | - | ||
| 359 | - # First setup was called, then teardown, then retry | ||
| 360 | - assert setup_count[0] == 2 | ||
| 361 | - # Old sockets were closed | ||
| 362 | - old_sub.close.assert_called() | ||
| 363 | - old_ctx.term.assert_called() | ||
| 364 | - # New sockets processed a message | ||
| 365 | - mock_report.assert_called_once() | ||
| 366 | - assert mock_report.call_args[0][0]["engine_id"] == 0 | ||
| 367 | - assert mock_report.call_args[0][0]["engine_status"] == 1 | ||
| 368 | - | ||
| 369 | - | ||
| 370 | -# -- Dedup after delivery (retry on failure) ---------------------------------- | ||
| 371 | - | ||
| 372 | - | ||
| 373 | - | ||
| 374 | -def test_process_zmq_failed_report_not_deduped(mock_report, reporter): | ||
| 375 | - """When Controller is unreachable (report returns False), the status must | ||
| 376 | - NOT be marked as known so it will be retried on the next ZMQ message. | ||
| 377 | - """ | ||
| 378 | - import msgspec.msgpack | ||
| 379 | - | ||
| 380 | - mock_report.return_value = False | ||
| 381 | - msg = {"schema_version": 1, "total_engines": 1, "engines": [{"id": 0, "status": "dead"}]} | ||
| 382 | - raw = msgspec.msgpack.encode(msg) | ||
| 383 | - known: dict[int, str] = {} | ||
| 384 | - reporter._process_zmq_engine_status(raw, known) | ||
| 385 | - | ||
| 386 | - # Report was attempted | ||
| 387 | - mock_report.assert_called_once() | ||
| 388 | - # But on failure, known_statuses must NOT contain the engine | ||
| 389 | - assert 0 not in known | ||
| 390 | - | ||
| 391 | - | ||
| 392 | - | ||
| 393 | -def test_process_zmq_successful_report_marked_as_known(mock_report, reporter): | ||
| 394 | - """When Controller confirms delivery (report returns True), the status | ||
| 395 | - IS marked as known so subsequent identical messages are deduplicated. | ||
| 396 | - """ | ||
| 397 | - import msgspec.msgpack | ||
| 398 | - | ||
| 399 | - mock_report.return_value = True | ||
| 400 | - msg = {"schema_version": 1, "total_engines": 1, "engines": [{"id": 0, "status": "dead"}]} | ||
| 401 | - raw = msgspec.msgpack.encode(msg) | ||
| 402 | - known: dict[int, str] = {} | ||
| 403 | - reporter._process_zmq_engine_status(raw, known) | ||
| 404 | - | ||
| 405 | - mock_report.assert_called_once() | ||
| 406 | - assert known == {0: "dead"} | ||
| 407 | - | ||
| 408 | - | ||
| 409 | # -- update_config restart conditions ------------------------------------------ | 183 | # -- update_config restart conditions ------------------------------------------ |
| 410 | 184 | ||
| 411 | 185 | ||
| 412 | -def test_update_config_restart_on_pod_ip_change(config, endpoints): | 186 | +def test_update_config_restart_on_endpoints_change(config, endpoints): |
| 413 | - """When pod_ip changes while enabled, restart to rebuild ZMQ sockets.""" | 187 | + """When endpoints change while enabled, restart to poll the new engines.""" |
| 414 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 415 | r = FaultReporter(config) | 188 | r = FaultReporter(config) |
| 416 | r._endpoints = endpoints | 189 | r._endpoints = endpoints |
| 417 | r.start() | 190 | r.start() |
| 418 | 191 | ||
| 419 | new_config = NodeManagerConfig() | 192 | new_config = NodeManagerConfig() |
| 420 | new_config.fault_tolerance_config.enable_fault_tolerance = True | 193 | new_config.fault_tolerance_config.enable_fault_tolerance = True |
| 421 | - new_config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 422 | - new_config.api_config.pod_ip = "10.0.0.99" # changed | ||
| 423 | - | ||
| 424 | - r.update_config(new_config, endpoints) | ||
| 425 | - | ||
| 426 | - assert r._enabled is True | ||
| 427 | - assert r._thread is not None | ||
| 428 | - r.stop() | ||
| 429 | - | ||
| 430 | - | ||
| 431 | -def test_update_config_restart_on_zmq_port_change(config, endpoints): | ||
| 432 | - """When zmq_pub_port changes while enabled, restart to rebuild ZMQ sockets.""" | ||
| 433 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 434 | - r = FaultReporter(config) | ||
| 435 | - r._endpoints = endpoints | ||
| 436 | - r.start() | ||
| 437 | - | ||
| 438 | - new_config = NodeManagerConfig() | ||
| 439 | - new_config.fault_tolerance_config.enable_fault_tolerance = True | ||
| 440 | - new_config.fault_tolerance_config.zmq_pub_port = 6666 # changed | ||
| 441 | new_config.api_config.pod_ip = "192.168.1.1" | 194 | new_config.api_config.pod_ip = "192.168.1.1" |
| 195 | + new_endpoints = endpoints + [Endpoint(id=2, ip="192.168.1.1", business_port="8002", mgmt_port="9002")] | ||
| 442 | 196 | ||
| 443 | - r.update_config(new_config, endpoints) | 197 | + r.update_config(new_config, new_endpoints) |
| 444 | 198 | ||
| 445 | assert r._enabled is True | 199 | assert r._enabled is True |
| 446 | assert r._thread is not None | 200 | assert r._thread is not None |
| @@ -448,8 +202,7 @@ def test_update_config_restart_on_zmq_port_change(config, endpoints): | |||
| 448 | 202 | ||
| 449 | 203 | ||
| 450 | def test_update_config_no_restart_when_nothing_changed(reporter, config, endpoints): | 204 | def test_update_config_no_restart_when_nothing_changed(reporter, config, endpoints): |
| 451 | - """When pod_ip, zmq_port, and endpoints are all unchanged, no restart.""" | 205 | + """When endpoints and config are unchanged, no restart.""" |
| 452 | - config.fault_tolerance_config.zmq_pub_port = 5555 | ||
| 453 | reporter._endpoints = endpoints | 206 | reporter._endpoints = endpoints |
| 454 | reporter.start() | 207 | reporter.start() |
| 455 | 208 | ||
| @@ -457,3 +210,362 @@ def test_update_config_no_restart_when_nothing_changed(reporter, config, endpoin | |||
| 457 | reporter.update_config(config, endpoints) | 210 | reporter.update_config(config, endpoints) |
| 458 | assert reporter._thread is t1 # Same thread object = no restart | 211 | assert reporter._thread is t1 # Same thread object = no restart |
| 459 | reporter.stop() | 212 | reporter.stop() |
| 213 | + | ||
| 214 | + | ||
| 215 | +def test_update_config_no_restart_on_poll_interval_change(reporter, config, endpoints): | ||
| 216 | + """Poll interval is read inside the loop, so changing it does not restart.""" | ||
| 217 | + reporter._endpoints = endpoints | ||
| 218 | + reporter.start() | ||
| 219 | + | ||
| 220 | + config.fault_tolerance_config.poll_interval_sec = 1.0 | ||
| 221 | + t1 = reporter._thread | ||
| 222 | + reporter.update_config(config, endpoints) | ||
| 223 | + assert reporter._thread is t1 | ||
| 224 | + reporter.stop() | ||
| 225 | + | ||
| 226 | + | ||
| 227 | +# -- engine status processing -------------------------------------------------- | ||
| 228 | + | ||
| 229 | + | ||
| 230 | + | ||
| 231 | +def test_process_healthy_updates_known_no_report(mock_report, reporter): | ||
| 232 | + known = {} | ||
| 233 | + reporter._process_engine_status(0, {"id": 0, "status": "healthy"}, known) | ||
| 234 | + mock_report.assert_not_called() | ||
| 235 | + assert known == {0: "healthy"} | ||
| 236 | + | ||
| 237 | + | ||
| 238 | + | ||
| 239 | +def test_process_unhealthy_with_fault_info(mock_report, reporter): | ||
| 240 | + known = {} | ||
| 241 | + reporter._process_engine_status(0, {"id": 0, "status": "unhealthy", "fault_info": "RuntimeError"}, known) | ||
| 242 | + mock_report.assert_called_once() | ||
| 243 | + called = mock_report.call_args[0][0] | ||
| 244 | + assert called["engine_id"] == 0 | ||
| 245 | + assert called["engine_status"] == 2 | ||
| 246 | + assert called["exception_type"] == "RuntimeError" | ||
| 247 | + assert known == {0: "unhealthy"} | ||
| 248 | + | ||
| 249 | + | ||
| 250 | + | ||
| 251 | +def test_process_unhealthy_without_fault_info(mock_report, reporter): | ||
| 252 | + known = {} | ||
| 253 | + reporter._process_engine_status(0, {"id": 0, "status": "unhealthy"}, known) | ||
| 254 | + mock_report.assert_called_once() | ||
| 255 | + called = mock_report.call_args[0][0] | ||
| 256 | + assert called["exception_type"] == "EngineUnhealthyError" | ||
| 257 | + | ||
| 258 | + | ||
| 259 | + | ||
| 260 | +def test_process_dead(mock_report, reporter): | ||
| 261 | + known = {} | ||
| 262 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 263 | + mock_report.assert_called_once() | ||
| 264 | + called = mock_report.call_args[0][0] | ||
| 265 | + assert called["engine_id"] == 0 | ||
| 266 | + assert called["engine_status"] == 1 | ||
| 267 | + assert called["exception_type"] == "EngineDeadError" | ||
| 268 | + assert known == {0: "dead"} | ||
| 269 | + | ||
| 270 | + | ||
| 271 | + | ||
| 272 | +def test_process_dedup_same_status(mock_report, reporter): | ||
| 273 | + known = {0: "dead"} | ||
| 274 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 275 | + mock_report.assert_not_called() | ||
| 276 | + | ||
| 277 | + | ||
| 278 | + | ||
| 279 | +def test_process_unknown_status(mock_report, reporter): | ||
| 280 | + known = {} | ||
| 281 | + reporter._process_engine_status(0, {"id": 0, "status": "weird"}, known) | ||
| 282 | + mock_report.assert_not_called() | ||
| 283 | + assert known == {} | ||
| 284 | + | ||
| 285 | + | ||
| 286 | + | ||
| 287 | +def test_process_recovered_then_faulted_again(mock_report, reporter): | ||
| 288 | + """After a healthy recovery resets the known status, a new fault is reported.""" | ||
| 289 | + known = {0: "unhealthy"} | ||
| 290 | + reporter._process_engine_status(0, {"id": 0, "status": "healthy"}, known) | ||
| 291 | + mock_report.assert_not_called() | ||
| 292 | + reporter._process_engine_status(0, {"id": 0, "status": "unhealthy"}, known) | ||
| 293 | + mock_report.assert_called_once() | ||
| 294 | + assert known == {0: "unhealthy"} | ||
| 295 | + | ||
| 296 | + | ||
| 297 | + | ||
| 298 | +def test_process_failed_report_not_deduped(mock_report, reporter): | ||
| 299 | + """When Controller is unreachable (report returns False), the status must | ||
| 300 | + NOT be marked as known so it will be retried on the next poll. | ||
| 301 | + """ | ||
| 302 | + mock_report.return_value = False | ||
| 303 | + known: dict[int, str] = {} | ||
| 304 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 305 | + | ||
| 306 | + mock_report.assert_called_once() | ||
| 307 | + assert 0 not in known | ||
| 308 | + | ||
| 309 | + | ||
| 310 | + | ||
| 311 | +def test_process_successful_report_marked_as_known(mock_report, reporter): | ||
| 312 | + """When Controller confirms delivery (report returns True), the status | ||
| 313 | + IS marked as known so subsequent identical polls are deduplicated. | ||
| 314 | + """ | ||
| 315 | + mock_report.return_value = True | ||
| 316 | + known: dict[int, str] = {} | ||
| 317 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 318 | + | ||
| 319 | + mock_report.assert_called_once() | ||
| 320 | + assert known == {0: "dead"} | ||
| 321 | + | ||
| 322 | + | ||
| 323 | +# -- status polling ------------------------------------------------------------ | ||
| 324 | + | ||
| 325 | + | ||
| 326 | + | ||
| 327 | +def test_query_engine_status_uses_business_port(mock_client_cls, config, endpoints): | ||
| 328 | + """FT status is fetched from the engine's business (API) port.""" | ||
| 329 | + config.fault_tolerance_config.poll_timeout_sec = 7.0 | ||
| 330 | + r = FaultReporter(config) | ||
| 331 | + mock_client = mock_client_cls.return_value | ||
| 332 | + mock_client.__enter__.return_value = mock_client # with-client pattern | ||
| 333 | + mock_client.get.return_value = {"engines": []} | ||
| 334 | + | ||
| 335 | + r._query_engine_status(endpoints[0]) | ||
| 336 | + | ||
| 337 | + mock_client_cls.assert_called_once_with(address="192.168.1.1:8000", tls_config=None, timeout=7.0) | ||
| 338 | + mock_client.get.assert_called_once_with("/fault_tolerance/status") | ||
| 339 | + | ||
| 340 | + | ||
| 341 | +def test_poll_engine_healthy_resets_failures(reporter, endpoints): | ||
| 342 | + """A successful poll clears the consecutive-failure counter and reports nothing.""" | ||
| 343 | + ep = endpoints[0] | ||
| 344 | + known: dict[int, str] = {} | ||
| 345 | + failures: dict[int, int] = {0: 2} | ||
| 346 | + | ||
| 347 | + with patch.object( | ||
| 348 | + reporter, | ||
| 349 | + "_query_engine_status", | ||
| 350 | + return_value={"engines": [{"id": 0, "status": "healthy"}]}, | ||
| 351 | + ): | ||
| 352 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 353 | + | ||
| 354 | + assert failures == {0: 0} | ||
| 355 | + assert known == {0: "healthy"} | ||
| 356 | + | ||
| 357 | + | ||
| 358 | + | ||
| 359 | +def test_poll_engine_unhealthy_reports(mock_report, reporter, endpoints): | ||
| 360 | + ep = endpoints[0] | ||
| 361 | + known: dict[int, str] = {} | ||
| 362 | + failures: dict[int, int] = {} | ||
| 363 | + | ||
| 364 | + with patch.object( | ||
| 365 | + reporter, | ||
| 366 | + "_query_engine_status", | ||
| 367 | + return_value={"engines": [{"id": 0, "status": "unhealthy", "fault_info": "KeyError"}]}, | ||
| 368 | + ): | ||
| 369 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 370 | + | ||
| 371 | + mock_report.assert_called_once() | ||
| 372 | + assert known == {0: "unhealthy"} | ||
| 373 | + assert failures == {0: 0} | ||
| 374 | + | ||
| 375 | + | ||
| 376 | +def test_poll_failures_below_threshold_no_report(reporter, endpoints): | ||
| 377 | + """Fewer than max_poll_failures consecutive failures are not reported.""" | ||
| 378 | + config = reporter._config | ||
| 379 | + config.fault_tolerance_config.max_poll_failures = 3 | ||
| 380 | + ep = endpoints[0] | ||
| 381 | + known: dict[int, str] = {} | ||
| 382 | + failures: dict[int, int] = {} | ||
| 383 | + | ||
| 384 | + with patch.object(reporter, "_query_engine_status", side_effect=RuntimeError("boom")): | ||
| 385 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 386 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 387 | + | ||
| 388 | + assert failures == {0: 2} | ||
| 389 | + assert known == {} | ||
| 390 | + | ||
| 391 | + | ||
| 392 | + | ||
| 393 | +def test_poll_failures_reach_threshold_reports_dead(mock_report, reporter, endpoints): | ||
| 394 | + """max_poll_failures consecutive failures are reported as dead.""" | ||
| 395 | + config = reporter._config | ||
| 396 | + config.fault_tolerance_config.max_poll_failures = 3 | ||
| 397 | + ep = endpoints[0] | ||
| 398 | + known: dict[int, str] = {} | ||
| 399 | + failures: dict[int, int] = {} | ||
| 400 | + | ||
| 401 | + with patch.object(reporter, "_query_engine_status", side_effect=RuntimeError("boom")): | ||
| 402 | + for _ in range(3): | ||
| 403 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 404 | + | ||
| 405 | + mock_report.assert_called_once() | ||
| 406 | + called = mock_report.call_args[0][0] | ||
| 407 | + assert called["engine_id"] == 0 | ||
| 408 | + assert called["engine_status"] == 1 | ||
| 409 | + assert called["exception_type"] == "EngineDeadError" | ||
| 410 | + assert "unreachable" in called["exception_message"] | ||
| 411 | + assert known == {0: "dead"} | ||
| 412 | + | ||
| 413 | + | ||
| 414 | + | ||
| 415 | +def test_poll_failures_dedup_dead(mock_report, reporter, endpoints): | ||
| 416 | + """Continued failures after dead was reported do not re-report.""" | ||
| 417 | + config = reporter._config | ||
| 418 | + config.fault_tolerance_config.max_poll_failures = 2 | ||
| 419 | + ep = endpoints[0] | ||
| 420 | + known: dict[int, str] = {0: "dead"} # already reported | ||
| 421 | + failures: dict[int, int] = {} | ||
| 422 | + | ||
| 423 | + with patch.object(reporter, "_query_engine_status", side_effect=RuntimeError("boom")): | ||
| 424 | + for _ in range(5): | ||
| 425 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 426 | + | ||
| 427 | + mock_report.assert_not_called() | ||
| 428 | + | ||
| 429 | + | ||
| 430 | +def test_poll_failures_then_recover(reporter, endpoints): | ||
| 431 | + """After failures, a successful poll resets the counter; later failures | ||
| 432 | + restart the counting from zero. | ||
| 433 | + """ | ||
| 434 | + config = reporter._config | ||
| 435 | + config.fault_tolerance_config.max_poll_failures = 3 | ||
| 436 | + ep = endpoints[0] | ||
| 437 | + known: dict[int, str] = {} | ||
| 438 | + failures: dict[int, int] = {} | ||
| 439 | + | ||
| 440 | + side_effects = [ | ||
| 441 | + RuntimeError("boom"), | ||
| 442 | + RuntimeError("boom"), | ||
| 443 | + {"engines": [{"id": 0, "status": "healthy"}]}, | ||
| 444 | + RuntimeError("boom"), | ||
| 445 | + ] | ||
| 446 | + with patch.object(reporter, "_query_engine_status", side_effect=side_effects): | ||
| 447 | + for _ in range(4): | ||
| 448 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) | ||
| 449 | + | ||
| 450 | + # 2 failures -> success (reset) -> 1 failure | ||
| 451 | + assert failures == {0: 1} | ||
| 452 | + assert known == {0: "healthy"} | ||
| 453 | + | ||
| 454 | + | ||
| 455 | +# -- main loop ----------------------------------------------------------------- | ||
| 456 | + | ||
| 457 | + | ||
| 458 | +def test_main_loop_polls_all_endpoints_then_stops(reporter, endpoints): | ||
| 459 | + """The loop polls every endpoint once per tick and exits on stop_event.""" | ||
| 460 | + config = reporter._config | ||
| 461 | + config.fault_tolerance_config.poll_interval_sec = 0.01 | ||
| 462 | + r = reporter | ||
| 463 | + r._endpoints = endpoints | ||
| 464 | + | ||
| 465 | + polled: list[int] = [] | ||
| 466 | + | ||
| 467 | + def fake_poll(ep, known, failures, first_poll_time): | ||
| 468 | + polled.append(ep.id) | ||
| 469 | + r._stop_event.set() # stop after the first full round | ||
| 470 | + | ||
| 471 | + with patch.object(r, "_poll_engine", side_effect=fake_poll): | ||
| 472 | + r._main_loop() | ||
| 473 | + | ||
| 474 | + assert polled == [0, 1] | ||
| 475 | + | ||
| 476 | + | ||
| 477 | + | ||
| 478 | +def test_main_loop_reports_via_sentinel(mock_report, config, endpoints): | ||
| 479 | + """End-to-end loop: one engine unhealthy -> reported once, then deduped.""" | ||
| 480 | + config.fault_tolerance_config.poll_interval_sec = 0.01 | ||
| 481 | + r = FaultReporter(config) | ||
| 482 | + r._endpoints = endpoints | ||
| 483 | + | ||
| 484 | + payload = {"engines": [{"id": 0, "status": "unhealthy", "fault_info": "RuntimeError"}]} | ||
| 485 | + poll_count = [0] | ||
| 486 | + | ||
| 487 | + def fake_query(ep): | ||
| 488 | + poll_count[0] += 1 | ||
| 489 | + if poll_count[0] >= 3: | ||
| 490 | + r._stop_event.set() | ||
| 491 | + return payload | ||
| 492 | + | ||
| 493 | + with patch.object(r, "_query_engine_status", side_effect=fake_query): | ||
| 494 | + r._main_loop() | ||
| 495 | + | ||
| 496 | + # Both endpoints report unhealthy (dedup keyed per endpoint); each is | ||
| 497 | + # reported once, then deduped on subsequent polls. | ||
| 498 | + assert mock_report.call_count == 2 | ||
| 499 | + assert {c.args[0]["engine_id"] for c in mock_report.call_args_list} == {0, 1} | ||
| 500 | + | ||
| 501 | + | ||
| 502 | +# -- robustness fixes (code review) ------------------------------------------- | ||
| 503 | + | ||
| 504 | + | ||
| 505 | +def test_poll_engine_malformed_payload_does_not_raise(reporter, endpoints): | ||
| 506 | + """A malformed FT status payload must not kill the polling thread.""" | ||
| 507 | + ep = endpoints[0] | ||
| 508 | + known: dict[int, str] = {} | ||
| 509 | + failures: dict[int, int] = {} | ||
| 510 | + | ||
| 511 | + for bad_payload in ([], None, "ok", {"engines": [{"id": 0}]}, {"engines": ["x"]}): | ||
| 512 | + with patch.object(reporter, "_query_engine_status", return_value=bad_payload): | ||
| 513 | + reporter._poll_engine(ep, known, failures, {ep.id: 0}) # must not raise | ||
| 514 | + | ||
| 515 | + | ||
| 516 | +def test_process_engine_status_uses_endpoint_id_key(reporter): | ||
| 517 | + """Dedup keys are endpoint ids, not payload ids — multi-endpoint safe.""" | ||
| 518 | + known: dict[int, str] = {} | ||
| 519 | + # endpoint 0 dead, endpoint 1 healthy: payload ids collide on 0, keys must not. | ||
| 520 | + with patch("motor.node_manager.core.fault_reporter.ControllerApiClient.report_software_fault") as mock_report: | ||
| 521 | + mock_report.return_value = True | ||
| 522 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 523 | + reporter._process_engine_status(1, {"id": 0, "status": "healthy"}, known) | ||
| 524 | + reporter._process_engine_status(0, {"id": 0, "status": "dead"}, known) | ||
| 525 | + | ||
| 526 | + assert known == {0: "dead", 1: "healthy"} | ||
| 527 | + mock_report.assert_called_once() # dedup: second dead report suppressed | ||
| 528 | + | ||
| 529 | + | ||
| 530 | +def test_report_unreachable_dead_within_grace_period_not_reported(reporter, endpoints): | ||
| 531 | + """Poll failures during engine startup (model load) are not reported dead.""" | ||
| 532 | + ep = endpoints[0] | ||
| 533 | + known: dict[int, str] = {} | ||
| 534 | + first_poll_time = {ep.id: __import__("time").time()} | ||
| 535 | + | ||
| 536 | + with patch("motor.node_manager.core.fault_reporter.ControllerApiClient.report_software_fault") as mock_report: | ||
| 537 | + reporter._report_unreachable_dead(ep, 3, known, first_poll_time) | ||
| 538 | + | ||
| 539 | + mock_report.assert_not_called() | ||
| 540 | + assert known == {} | ||
| 541 | + | ||
| 542 | + | ||
| 543 | +def test_report_unreachable_dead_after_grace_period_reported(reporter, endpoints): | ||
| 544 | + ep = endpoints[0] | ||
| 545 | + known: dict[int, str] = {} | ||
| 546 | + first_poll_time = {ep.id: 0} # long ago, grace period over | ||
| 547 | + | ||
| 548 | + with patch("motor.node_manager.core.fault_reporter.ControllerApiClient.report_software_fault") as mock_report: | ||
| 549 | + mock_report.return_value = True | ||
| 550 | + reporter._report_unreachable_dead(ep, 3, known, first_poll_time) | ||
| 551 | + | ||
| 552 | + mock_report.assert_called_once() | ||
| 553 | + assert known == {0: "dead"} | ||
| 554 | + | ||
| 555 | + | ||
| 556 | +def test_engine_ft_enabled_int_value(tmp_path): | ||
| 557 | + """Config value 1 (not just JSON true) enables FT auto-detection.""" | ||
| 558 | + path = _write_user_config( | ||
| 559 | + tmp_path, | ||
| 560 | + {"motor_engine_prefill_config": {"engine_config": {"enable_fault_tolerance": 1}}}, | ||
| 561 | + ) | ||
| 562 | + assert _engine_ft_enabled(path) is True | ||
| 563 | + | ||
| 564 | + | ||
| 565 | +def test_engine_ft_enabled_ignores_non_engine_sections(tmp_path): | ||
| 566 | + """A nested 'engine_config' outside the known engine sections is ignored.""" | ||
| 567 | + path = _write_user_config( | ||
| 568 | + tmp_path, | ||
| 569 | + {"motor_deploy_config": {"engine_config": {"enable_fault_tolerance": True}}}, | ||
| 570 | + ) | ||
| 571 | + assert _engine_ft_enabled(path) is False | ||
config_sample 已换成 poll_interval_sec 等字段，但 docs/zh/user_guide/configuration/config_reference.md 里 NodeManager 的 fault_tolerance_config 还是 {...} 占位，没列这三个新字段，也没写 zmq_pub_port 已移除。用户查配置参考会找不到，建议同步改 config_reference。