fix(api): separate traffic readiness from worker health (#604)

Expose /readyz for the runtime snapshot, PostgreSQL, and Redis. Keep worker and logging diagnostics on /healthz so a maintenance failure can alert without removing a serving instance from service discovery.

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-11-04 19:29:50 +08:00
committed by GitHub
co-authored by Zyy Codex
parent 19ef484969
commit 13374e2a70
9 changed files with 136 additions and 40 deletions
+37 -18
View File
@@ -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<Item = &'a Arc<dyn HealthProbe>>,
) -> Vec<HealthFailure> {
// 探针并行共用截止时间,保留已完成结果并指出具体超时组件
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<ApiState>) -> 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<ApiState>) -> 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
+1
View File
@@ -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())
+67 -15
View File
@@ -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())
}
@@ -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>;
}
@@ -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() {
@@ -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();
+8 -5
View File
@@ -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 秒报告已等待时间,
+4 -1
View File
@@ -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 数据面与模型目录
+2 -1
View File
@@ -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-客户端下载)