执行摘要

本 PR 在 Rust 前端 gRPC 监听器上注册了标准的 grpc.health.v1.Health 服务，通过 EngineCoreClient 的 Tokio watch 通道接收引擎健康变更，并将健康状态报告为 SERVING（启动后）或 NOT_SERVING（引擎不健康或关闭时）。同时保持了活跃健康流的持续交付，并在 graceful shutdown 时正确关闭。没有新增端口或监听器。

功能与动机

PR 目的是补充 Rust 前端缺失的标准 gRPC 健康检查端点。目前 vLLM 的 gRPC 服务只提供 vllm.Generate，外部监控与负载均衡工具无法通过 gRPC 标准协议获取服务器健康状态。此变更实现 grpc.health.v1.Health 协议，使基础设施可以依赖统一健康检查。

实现拆解

1. 引擎客户端扩展：在 rust/src/engine-core-client/src/client/imp.rs 中添加 health_tx: watch::Sender<bool> 字段，初始化为 true；提供 subscribe_health() 返回接收器；在 close_registries() 中调用 publish_unhealthy() 发送 false，实现 sticky 不健康转换。
2. 健康监控模块：新建 rust/src/server/src/grpc/health.rs，包含 monitor_health 异步函数，通过 tokio::select! 等待引擎不健康或关闭信号，然后设置 HealthReporter 状态为 NOT_SERVING，并在关闭后清理服务状态。
3. 服务器集成：在 rust/src/server/src/lib.rs 中，当启用 gRPC 端口时，创建 tonic_health::server::health_reporter() 并订阅引擎健康；将健康服务和生成服务注册到同一 gRPC 服务器；通过 tokio::join! 并发运行服务器和健康监控任务。
4. 测试配套：在 tests.rs 中新增 start_grpc_test_server 辅助函数，以及两个集成测试：验证引擎不健康时健康状态转换为 NOT_SERVING，验证优雅关闭时健康 watch 正确结束。
5. 依赖更新：修改 rust/Cargo.toml 和 rust/src/server/Cargo.toml 添加 tonic-health 依赖，并统一其他 tonic 相关版本至 0.14.6。

rust/src/server/src/grpc/health.rs

核心新增文件，实现了基于引擎状态的 gRPC 健康监控逻辑

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
​
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use tonic::server::NamedService;
use tonic_health::ServingStatus;
use tonic_health::server::HealthReporter;
use tracing::{info, warn};
​
use super::GenerateGrpcService;
​
/// 监控引擎健康状态，并驱动 gRPC 健康报告。
/// 当引擎不健康或收到关闭信号时，将服务状态标记为 NOT_SERVING。
pub(crate) async fn monitor_health(
    mut health_reporter: HealthReporter,
    mut engine_health: watch::Receiver<bool>,
    shutdown: CancellationToken,
) {
    let generate_service = GenerateGrpcService::NAME;
    let status = ServingStatus::NotServing;
    // 等待引擎健康变为 false 或 shutdown 触发
    let health_event_first = tokio::select! {
        result = engine_health.wait_for(|healthy| !*healthy) => {
            match result {
                Ok(_) => warn!(
                    generate_service,
                    overall_service = true,
                    status = ?status,
                    reason = "engine_unhealthy",
                    "标记 gRPC 健康服务为不可用（引擎不健康）"
                ),
                Err(error) => warn!(
                    %error,
                    generate_service,
                    overall_service = true,
                    status = ?status,
                    reason = "health_channel_closed",
                    "引擎健康通道关闭，标记 gRPC 健康服务为不可用"
                ),
            }
            true // 引擎健康事件先发生
        }
        _ = shutdown.cancelled() => {
            info!(
                generate_service,
                overall_service = true,
                status = ?status,
                reason = "server_shutdown",
                "服务器关闭中，标记 gRPC 健康服务为不可用"
            );
            false // shutdown 先触发
        }
    };
​
    // 设置服务状态为 NOT_SERVING
    health_reporter.set_not_serving::<GenerateGrpcService>().await;
    // 整体服务镜像 Generate 服务状态
    health_reporter.set_service_status("", status).await;
​
    if health_event_first {
        // 如果是引擎事件先发生，等待关闭完成再关闭 watch
        shutdown.cancelled().await;
        info!(
            generate_service,
            overall_service = true,
            reason = "server_shutdown",
            "服务器关闭中，关闭 gRPC 健康 watch"
        );
    }
​
    // 清理服务状态，确保客户端收到终止
    health_reporter.clear_service_status(generate_service).await;
    health_reporter.clear_service_status("").await;
}

rust/src/engine-core-client/src/client/imp.rs

引擎客户端核心实现，添加 health_tx 字段和订阅/发布方法

// 在 ClientInner 中添加 health_tx 字段
pub(crate) struct ClientInner {
    // ... existing fields ...
    health_error: ArcSwapOption<Error>,
    /// 通过 watch 通道向外部传播健康状态变化。
    /// 初始值为 `true`，一旦引擎失败变为 `false` 并保持。
    health_tx: watch::Sender<bool>,
}
​
impl ClientInner {
    pub fn new(/* ... */) -> Self {
        // ... existing init ...
        Self {
            // ...
            health_error: ArcSwapOption::empty(),
            health_tx: watch::Sender::new(true), // 初始健康
        }
    }
​
    /// 订阅引擎健康变化。返回的 Receiver 当前值为 `true`，
    /// 在引擎永久失败后变为 `false`。
    pub fn subscribe_health(&self) -> watch::Receiver<bool> {
        self.health_tx.subscribe()
    }
​
    /// 当引擎发生第一个持久性健康错误时，发布健康转换。
    /// 使用 `send_if_modified` 确保只从 `true` 变为 `false` 一次。
    fn publish_unhealthy(&self) {
        self.health_tx.send_if_modified(|healthy| std::mem::replace(healthy, false));
    }
​
    /// 关闭所有注册表并发布不健康状态
    pub fn close_registries(&self, error: Arc<Error>) {
        let persistent_error = self.record_health_error(error);
        // 在关闭请求和工具注册表之前发布不健康信号，
        // 以便健康 watch 接收者能尽早得知状态变化。
        self.publish_unhealthy();
        let request_senders = self.request_reg.lock().close();
        let utility_senders = self.utility_reg.lock().close();
        // ... 关闭发送器 ...
    }
}

评论区精华

• 提取到独立文件：BugenZhao 建议将 monitor_health 从 lib.rs 提取到 grpc 子模块。作者创建了 health.rs，已解决。
• 简化 publish_unhealthy：njhill 提出使用 send_if_modified 模式简化实现。作者采纳并修改。
• 统一 tonic 版本：njhill 建议将所有 tonic 依赖升级到 0.14.6。作者在 Cargo.toml 中统一了版本。

风险与影响

风险较低。主要风险在于健康报告完全依赖引擎健康状态的转换，如果引擎健康检测逻辑本身存在缺陷，可能导致错误报告。新代码使用 tonic-health 标准库，稳定性高。影响范围限于 Rust 前端 gRPC 服务，对外暴露标准健康端点，提升可观测性。团队需关注新依赖的版本兼容。

关联脉络

没有发现与近期历史 PR 的直接关联。此 PR 是 Rust 前端在可观测性方面的重要补充，为后续更丰富的监控功能奠定基础。对于部署在 Kubernetes 等服务网格环境的用户，标准健康检查端点可直接用于就绪探针和存活探针。