fix(api): keep provider degradation out of healthz

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-08-20 15:14:28 +08:00
co-authored by Codex
parent 4f85209dc9
commit 7f420257a6
3 changed files with 113 additions and 4 deletions
+21 -3
View File
@@ -1,4 +1,4 @@
//! `/healthz` 对 Core、Store 与 Host worker 健康事实的聚合。
//! `/healthz` 对 Core、Store 与 Host 关键 worker 健康事实的聚合。
use std::sync::Arc;
use std::time::Duration;
@@ -6,8 +6,10 @@ use std::time::Duration;
use axum::{extract::State, http::StatusCode};
use futures::future::join_all;
use gateway_core::health::{
HealthProbe, HealthState, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState,
HealthProbe, HealthState, WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource,
WorkerRuntimeState,
};
use gateway_core::task::WorkerKind;
use crate::ApiState;
@@ -32,7 +34,13 @@ impl HealthStatus {
}
pub(crate) async fn healthy(&self) -> bool {
if !self.workers.snapshot().iter().all(worker_is_healthy) {
if !self
.workers
.snapshot()
.iter()
.filter(|worker| worker_affects_healthz(worker))
.all(worker_is_healthy)
{
return false;
}
let checks = self.probes.iter().map(|probe| probe.check());
@@ -54,6 +62,16 @@ pub(crate) async fn healthz(State(state): State<ApiState>) -> StatusCode {
}
}
fn worker_affects_healthz(worker: &WorkerHealthSnapshot) -> bool {
let WorkerHealthKey::Task(id) = &worker.key else {
return false;
};
!matches!(
id.kind(),
WorkerKind::OAuthRefresh | WorkerKind::QuotaCatalogHealth
)
}
fn worker_is_healthy(worker: &WorkerHealthSnapshot) -> bool {
match worker.state {
WorkerRuntimeState::Disabled | WorkerRuntimeState::Standby => true,
@@ -11,7 +11,9 @@ use gateway_core::{
StartedExecution,
},
error::GatewayError,
health::{WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState},
routing::PublicModelId,
task::{WorkerId, WorkerKind},
};
use tower::ServiceExt;
@@ -31,6 +33,75 @@ async fn healthz_should_return_no_content_when_all_inputs_are_healthy() {
assert_eq!(response.status(), StatusCode::NO_CONTENT);
}
#[tokio::test]
async fn healthz_should_ignore_provider_maintenance_worker_failure() {
let worker_health = StaticWorkerHealth(vec![worker_snapshot(
WorkerKind::QuotaCatalogHealth,
WorkerRuntimeState::BackingOff,
)]);
let response = crate::openai::api_router_with_worker_health(
Arc::new(UnusedExecution),
Arc::new(worker_health),
)
.await
.oneshot(
Request::builder()
.uri("/healthz")
.body(Body::empty())
.expect("health request"),
)
.await
.expect("health response");
assert_eq!(response.status(), StatusCode::NO_CONTENT);
}
#[tokio::test]
async fn healthz_should_reject_critical_worker_failure() {
let worker_health = StaticWorkerHealth(vec![worker_snapshot(
WorkerKind::RuntimeSnapshotReconciliation,
WorkerRuntimeState::BackingOff,
)]);
let response = crate::openai::api_router_with_worker_health(
Arc::new(UnusedExecution),
Arc::new(worker_health),
)
.await
.oneshot(
Request::builder()
.uri("/healthz")
.body(Body::empty())
.expect("health request"),
)
.await
.expect("health response");
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
}
#[derive(Clone)]
struct StaticWorkerHealth(Vec<WorkerHealthSnapshot>);
impl WorkerHealthSource for StaticWorkerHealth {
fn snapshot(&self) -> Vec<WorkerHealthSnapshot> {
self.0.clone()
}
}
fn worker_snapshot(kind: WorkerKind, state: WorkerRuntimeState) -> WorkerHealthSnapshot {
let id = WorkerId::try_new(kind, "health-test").expect("worker ID");
WorkerHealthSnapshot {
key: WorkerHealthKey::Task(id),
state,
consecutive_failures: 1,
completed_cycles: 0,
last_fencing_token: None,
last_success_at: None,
last_failure_at: None,
last_error: Some("test worker failure".to_owned()),
}
}
struct UnusedExecution;
impl ExecutionService for UnusedExecution {
+21 -1
View File
@@ -44,9 +44,29 @@ pub(super) async fn api_router(execution: Arc<dyn ExecutionService>) -> axum::Ro
api_router_with_origins(execution, Vec::new()).await
}
pub(super) async fn api_router_with_worker_health(
execution: Arc<dyn ExecutionService>,
worker_health: Arc<dyn WorkerHealthSource>,
) -> axum::Router {
api_router_with_origins_and_worker_health(execution, Vec::new(), worker_health).await
}
pub(super) async fn api_router_with_origins(
execution: Arc<dyn ExecutionService>,
cors_allowed_origins: Vec<String>,
) -> axum::Router {
api_router_with_origins_and_worker_health(
execution,
cors_allowed_origins,
Arc::new(EmptyWorkerHealth),
)
.await
}
async fn api_router_with_origins_and_worker_health(
execution: Arc<dyn ExecutionService>,
cors_allowed_origins: Vec<String>,
worker_health: Arc<dyn WorkerHealthSource>,
) -> axum::Router {
let admin = crate::admin::AdminTestFixture::new().await;
gateway_api::initialize(
@@ -59,7 +79,7 @@ pub(super) async fn api_router_with_origins(
execution,
admin.services,
Vec::new(),
Arc::new(EmptyWorkerHealth),
worker_health,
Arc::new(TestLifecycle::default()),
)
.expect("API bundle")