diff --git a/backend/crates/gateway-api/src/health.rs b/backend/crates/gateway-api/src/health.rs index 4fb6b2205..a972fe1c5 100644 --- a/backend/crates/gateway-api/src/health.rs +++ b/backend/crates/gateway-api/src/health.rs @@ -1,4 +1,4 @@ -//! `/healthz` 对 Core、Store 与 Host 关键 worker 健康事实的聚合 +//! 请求就绪检查与综合健康诊断 use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -67,22 +67,7 @@ impl HealthStatus { ), }) .collect(); - // 逐个探针共享截止时间,保留已完成结果并指出具体超时组件 - let deadline = tokio::time::Instant::now() + HEALTH_CHECK_TIMEOUT; - let checks = self.probes.iter().map(|probe| async move { - let (state, reason) = match tokio::time::timeout_at(deadline, probe.check()).await { - Ok(HealthState::Healthy) => return None, - Ok(HealthState::Degraded(reason)) => ("degraded", reason), - Ok(HealthState::Unhealthy(reason)) => ("unhealthy", reason), - Err(_) => ("timed_out", "health probe exceeded 2s deadline".to_owned()), - }; - Some(HealthFailure { - component: format!("probe:{}", probe.name()), - state, - reason, - }) - }); - failures.extend(join_all(checks).await.into_iter().flatten()); + failures.extend(probe_failures(self.probes.iter()).await); let healthy = failures.is_empty(); let changed = { let mut previous = self @@ -128,8 +113,42 @@ impl HealthStatus { } } +async fn probe_failures<'a>( + probes: impl Iterator>, +) -> Vec { + // 探针并行共用截止时间,保留已完成结果并指出具体超时组件 + let deadline = tokio::time::Instant::now() + HEALTH_CHECK_TIMEOUT; + let checks = probes.map(|probe| async move { + let (state, reason) = match tokio::time::timeout_at(deadline, probe.check()).await { + Ok(HealthState::Healthy) => return None, + Ok(HealthState::Degraded(reason)) => ("degraded", reason), + Ok(HealthState::Unhealthy(reason)) => ("unhealthy", reason), + Err(_) => ("timed_out", "health probe exceeded 2s deadline".to_owned()), + }; + Some(HealthFailure { + component: format!("probe:{}", probe.name()), + state, + reason, + }) + }); + join_all(checks).await.into_iter().flatten().collect() +} + +pub(crate) async fn readyz(State(state): State) -> StatusCode { + let required = state + .health() + .probes + .iter() + .filter(|probe| probe.required_for_readiness()); + status_code(probe_failures(required).await.is_empty()) +} + pub(crate) async fn healthz(State(state): State) -> StatusCode { - if state.health().healthy().await { + status_code(state.health().healthy().await) +} + +fn status_code(healthy: bool) -> StatusCode { + if healthy { StatusCode::NO_CONTENT } else { StatusCode::SERVICE_UNAVAILABLE diff --git a/backend/crates/gateway-api/src/lib.rs b/backend/crates/gateway-api/src/lib.rs index 78cef67c7..49b3ec430 100644 --- a/backend/crates/gateway-api/src/lib.rs +++ b/backend/crates/gateway-api/src/lib.rs @@ -216,6 +216,7 @@ pub fn initialize( }; let index = config.asset_directory.join("index.html"); let mut router = Router::new() + .route("/readyz", get(health::readyz)) .route("/healthz", get(health::healthz)) .merge(openai::router::router()) .merge(provider::router()) diff --git a/backend/crates/gateway-api/tests/health.rs b/backend/crates/gateway-api/tests/health.rs index 3bc5f8808..aef9b55e6 100644 --- a/backend/crates/gateway-api/tests/health.rs +++ b/backend/crates/gateway-api/tests/health.rs @@ -1,4 +1,4 @@ -//! 验证健康接口按基础设施与关键 Worker 状态判定可用性 +//! 验证请求就绪、综合健康状态与故障诊断 use std::sync::{Arc, Mutex}; @@ -86,6 +86,54 @@ async fn healthz_should_reject_critical_worker_failure() { assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); } +#[tokio::test] +async fn readiness_ignores_freeze_recovery_and_logging_failures_without_clearing_alerts() { + let diagnostics = Arc::new(RecordingDiagnostics::default()); + let router = health_router( + vec![Arc::new(MutableProbe(Mutex::new(HealthState::Degraded( + "file logging maintenance failed".to_owned(), + ))))], + vec![worker_snapshot( + WorkerKind::AccountFreezeRecovery, + WorkerRuntimeState::BackingOff, + )], + diagnostics.clone(), + ) + .await; + + for (path, expected) in [ + ("/healthz", StatusCode::SERVICE_UNAVAILABLE), + ("/readyz", StatusCode::NO_CONTENT), + ("/healthz", StatusCode::SERVICE_UNAVAILABLE), + ] { + assert_eq!( + health_response(router.clone(), path).await.status(), + expected + ); + } + assert_eq!(diagnostics.0.lock().unwrap().len(), 2); +} + +#[tokio::test] +async fn readiness_rejects_an_unavailable_runtime_snapshot() { + let router = health_router( + vec![Arc::new( + gateway_core::runtime::RuntimeSnapshotHandle::default(), + )], + Vec::new(), + Arc::new(RecordingDiagnostics::default()), + ) + .await; + let response = health_response(router, "/readyz").await; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!( + axum::body::to_bytes(response.into_body(), 1024) + .await + .unwrap() + .is_empty() + ); +} + #[tokio::test] async fn healthz_records_probe_changes_once_and_records_recurrence_after_recovery() { let probe = Arc::new(MutableProbe(Mutex::new(HealthState::Unhealthy( @@ -95,7 +143,8 @@ async fn healthz_records_probe_changes_once_and_records_recurrence_after_recover let router = health_router(vec![probe.clone()], Vec::new(), diagnostics.clone()).await; let responses = - futures::future::join_all((0..8).map(|_| health_response(router.clone()))).await; + futures::future::join_all((0..8).map(|_| health_response(router.clone(), "/healthz"))) + .await; for response in responses { assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); assert!( @@ -126,18 +175,18 @@ async fn healthz_records_probe_changes_once_and_records_recurrence_after_recover *probe.0.lock().unwrap() = HealthState::Degraded("file logging maintenance failed".to_owned()); assert_eq!( - health_response(router.clone()).await.status(), + health_response(router.clone(), "/healthz").await.status(), StatusCode::SERVICE_UNAVAILABLE ); assert_eq!(diagnostics.0.lock().unwrap().len(), 2); *probe.0.lock().unwrap() = HealthState::Healthy; assert_eq!( - health_response(router.clone()).await.status(), + health_response(router.clone(), "/healthz").await.status(), StatusCode::NO_CONTENT ); *probe.0.lock().unwrap() = HealthState::Degraded("file logging maintenance failed".to_owned()); assert_eq!( - health_response(router).await.status(), + health_response(router, "/healthz").await.status(), StatusCode::SERVICE_UNAVAILABLE ); assert_eq!(diagnostics.0.lock().unwrap().len(), 3); @@ -168,7 +217,11 @@ async fn healthz_retains_worker_and_completed_probe_failures_when_another_probe_ .await; assert_eq!( - health_response(router).await.status(), + health_response(router.clone(), "/readyz").await.status(), + StatusCode::SERVICE_UNAVAILABLE + ); + assert_eq!( + health_response(router, "/healthz").await.status(), StatusCode::SERVICE_UNAVAILABLE ); let events = diagnostics.0.lock().unwrap(); @@ -202,12 +255,12 @@ async fn healthz_retries_failed_diagnostic_admission_without_changing_health_res .await; assert_eq!( - health_response(router.clone()).await.status(), + health_response(router.clone(), "/healthz").await.status(), StatusCode::SERVICE_UNAVAILABLE ); assert!(diagnostics.recorded.0.lock().unwrap().is_empty()); assert_eq!( - health_response(router).await.status(), + health_response(router, "/healthz").await.status(), StatusCode::SERVICE_UNAVAILABLE ); assert_eq!(diagnostics.recorded.0.lock().unwrap().len(), 1); @@ -258,14 +311,9 @@ async fn health_router( .router() } -async fn health_response(router: axum::Router) -> axum::response::Response { +async fn health_response(router: axum::Router, path: &str) -> axum::response::Response { router - .oneshot( - Request::builder() - .uri("/healthz") - .body(Body::empty()) - .unwrap(), - ) + .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) .await .unwrap() } @@ -289,6 +337,10 @@ impl HealthProbe for PendingProbe { "pending" } + fn required_for_readiness(&self) -> bool { + true + } + fn check(&self) -> BoxFuture<'_, HealthState> { Box::pin(std::future::pending()) } diff --git a/backend/crates/gateway-core/src/health.rs b/backend/crates/gateway-core/src/health.rs index 3ff1066f2..ca17cb650 100644 --- a/backend/crates/gateway-core/src/health.rs +++ b/backend/crates/gateway-core/src/health.rs @@ -18,6 +18,11 @@ pub enum HealthState { pub trait HealthProbe: Send + Sync { fn name(&self) -> &'static str; + /// 仅接收请求必需的依赖参与就绪检查 + fn required_for_readiness(&self) -> bool { + false + } + fn check(&self) -> BoxFuture<'_, HealthState>; } diff --git a/backend/crates/gateway-core/src/runtime/publications.rs b/backend/crates/gateway-core/src/runtime/publications.rs index 7a43688d1..305f6814f 100644 --- a/backend/crates/gateway-core/src/runtime/publications.rs +++ b/backend/crates/gateway-core/src/runtime/publications.rs @@ -115,6 +115,10 @@ impl HealthProbe for RuntimeSnapshotHandle { "runtime_snapshot" } + fn required_for_readiness(&self) -> bool { + true + } + fn check(&self) -> BoxFuture<'_, HealthState> { Box::pin(async move { if self.acquire().is_ok() { diff --git a/backend/crates/gateway-store/src/workers.rs b/backend/crates/gateway-store/src/workers.rs index ad9257bb0..e4194584c 100644 --- a/backend/crates/gateway-store/src/workers.rs +++ b/backend/crates/gateway-store/src/workers.rs @@ -276,6 +276,10 @@ impl HealthProbe for PostgresHealthProbe { "postgres" } + fn required_for_readiness(&self) -> bool { + true + } + fn check(&self) -> futures::future::BoxFuture<'_, HealthState> { Box::pin(health_state_with_one_retry( || self.check_once(), @@ -319,6 +323,10 @@ impl HealthProbe for RedisHealthProbe { "redis" } + fn required_for_readiness(&self) -> bool { + true + } + fn check(&self) -> futures::future::BoxFuture<'_, HealthState> { Box::pin(async move { let mut connection = self.connection.clone(); diff --git a/deploy/README.md b/deploy/README.md index f4c26e7a7..652442085 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -358,16 +358,19 @@ docker compose -f deploy/compose.yaml up -d --no-build --wait docker compose -f deploy/compose.yaml ps ``` -健康检查: +Consul 或负载均衡器用 `GET /readyz` 决定是否接入流量。升级后先在实例上确认返回 `204`, +再把服务注册中的检查路径切到 `/readyz`。如果路由还依赖 Docker 容器健康状态, +也要将应用容器的 `healthcheck.test` 指向 `/readyz` ```bash -curl -i http://127.0.0.1:8080/healthz +curl -i http://127.0.0.1:8080/readyz ``` -`204 No Content` 表示应用、PostgreSQL、Redis 和后台任务的健康检查通过,不代表每个上游账号都可用 +用独立监控定期请求 `GET /healthz`,对文件日志故障、账号解冻等关键后台任务异常告警。镜像和 Compose 默认使用这项综合检查。 +两个接口的状态码见[健康检查 API](../docs/api.md#2-健康检查) -`503 Service Unavailable` 的失败组件与原因保存在运维错误记录中,组件为 `health`,关联标识以 -`probe:` 或 `worker:` 开头;探针超时也会指明组件。连续相同故障去重,恢复后再次失败重新记录, +`/healthz` 返回 `503` 时,失败组件与原因保存在运维错误记录中,组件为 `health`,关联标识以 +`probe:` 或 `worker:` 开头。Worker 事件的错误详情包含任务状态和 `last_error`。连续相同故障去重,恢复后再次失败重新记录, 应用日志记录故障与恢复摘要。健康响应保持空正文,详情沿用有界观测队列,队列或存储故障时可能缺失 HTTP 在数据库迁移和初始化完成后才监听。迁移期间启动日志每 15 秒报告已等待时间, diff --git a/docs/api.md b/docs/api.md index e8d6662eb..e3465b7e9 100644 --- a/docs/api.md +++ b/docs/api.md @@ -203,7 +203,10 @@ Dashboard 的 `lastUsedRelative` 和账号额度的 `refreshedAtRelative` 由后 | 方法 | 路由 | 鉴权 | 说明 | | --- | --- | --- | --- | -| `GET` | `/healthz` | 无 | Core、Store 和后台任务健康时返回 `204`,否则返回 `503` | +| `GET` | `/readyz` | 无 | 流量就绪检查。运行时快照可用且 PostgreSQL、Redis 探针健康时返回 `204`,否则返回 `503` | +| `GET` | `/healthz` | 无 | 综合健康检查。在就绪依赖之外检查文件日志和关键后台任务,全部健康时返回 `204`,否则返回 `503` | + +两个接口均返回空正文。探针并行执行,共用 2 秒截止时间 ## 3. OpenAI 数据面与模型目录 diff --git a/docs/architecture.md b/docs/architecture.md index 44a0d3405..50d7b71d7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1216,7 +1216,8 @@ Admin worker 复用连接测试,按凭据及冻结代次比较后解除或顺 ### 启动与关闭 启动只有在配置、PostgreSQL、Redis、Provider、Core、Admin、API 和 Worker 全部初始化成功后才进入服务。 -健康检查综合 Core、Store 与 Worker 状态,但不会把单个 Provider 的业务降级等同于整个进程失活。 +Core 与 Store 在各自的探针上声明接收请求所需的依赖。API 的就绪检查只运行这些探针, +综合健康检查另纳入 Host 文件日志与关键 Worker 状态,并记录故障变化。 客户端下载解析由 Host 的 `ClientDistributionResolver` 实现,组合根注入 Admin;只在管理员请求时访问外部来源, 不阻塞启动。失败时使用官方稳定地址,临时直链不持久化或代理下载,字段与来源规则见 [客户端下载 API](api.md#windows-客户端下载)