diff --git a/gears/system/api-gateway/src/gear.rs b/gears/system/api-gateway/src/gear.rs index 402bbe34b..2168da042 100644 --- a/gears/system/api-gateway/src/gear.rs +++ b/gears/system/api-gateway/src/gear.rs @@ -77,6 +77,8 @@ pub struct ApiGateway { pub(crate) final_router: Mutex>, // AuthN Resolver client (resolved during init, None when auth_disabled) pub(crate) authn_client: Mutex>>, + // Readiness healthcheck registry — populated during REST wiring phase + pub(crate) healthcheck_registry: Mutex>>, // Duplicate detection (per (method, path) and per handler id) pub(crate) registered_routes: DashMap<(Method, String), ()>, @@ -92,6 +94,7 @@ impl Default for ApiGateway { router_cache: RouterCache::new(default_router), final_router: Mutex::new(None), authn_client: Mutex::new(None), + healthcheck_registry: Mutex::new(None), registered_routes: DashMap::new(), registered_handlers: DashMap::new(), } @@ -99,17 +102,43 @@ impl Default for ApiGateway { } impl ApiGateway { - fn apply_prefix_nesting(mut router: Router, prefix: &str) -> Router { + fn apply_prefix_nesting( + &self, + router: Router, + prefix: &str, + hc_registry: Arc, + authn_client: Option>, + ) -> Result { if prefix.is_empty() { - return router; + return Ok(router); } - let top = Router::new() - .route("/health", get(web::health_check)) - .route("/healthz", get(|| async { "ok" })); + // Root-mounted health routes must go through the same middleware stack + // (request ID, tracing, metrics, CatchPanicLayer) as the prefixed API. + let top = self.apply_middleware_stack(Self::health_routes(hc_registry), authn_client)?; + let nested = Router::new().nest(prefix, router); + Ok(top.merge(nested)) + } - router = Router::new().nest(prefix, router); - top.merge(router) + fn health_routes(hc_registry: Arc) -> Router { + let reg_h = hc_registry.clone(); + let reg_r = hc_registry; + Router::new() + .route( + "/health", + get(move || { + let r = reg_h.clone(); + async move { web::health_detail(r).await } + }), + ) + .route("/healthz", get(|| async { "ok" })) + .route( + "/readyz", + get(move || { + let r = reg_r.clone(); + async move { web::readyz_check(r).await } + }), + ) } /// Create a new `ApiGateway` instance with the given configuration @@ -122,6 +151,7 @@ impl ApiGateway { router_cache: RouterCache::new(default_router), final_router: Mutex::new(None), authn_client: Mutex::new(None), + healthcheck_registry: Mutex::new(None), registered_routes: DashMap::new(), registered_handlers: DashMap::new(), } @@ -160,6 +190,7 @@ impl ApiGateway { // Always mark built-in health check routes as public public_routes.insert((Method::GET, "/health".to_owned())); public_routes.insert((Method::GET, "/healthz".to_owned())); + public_routes.insert((Method::GET, "/readyz".to_owned())); public_routes.insert((Method::GET, "/docs".to_owned())); public_routes.insert((Method::GET, "/openapi.json".to_owned())); @@ -484,19 +515,23 @@ impl ApiGateway { } tracing::debug!("Building new router (standalone/fallback mode)"); - // In standalone mode (no REST pipeline), register both health endpoints here. + // In standalone mode (no REST pipeline), register health endpoints here. // In normal operation, rest_prepare() registers these instead. - let mut router = Router::new() - .route("/health", get(web::health_check)) - .route("/healthz", get(|| async { "ok" })); + let hc_registry = self + .healthcheck_registry + .lock() + .clone() + .unwrap_or_else(|| Arc::new(toolkit::RestHealthcheckRegistry::new())); + + let mut router = Self::health_routes(hc_registry.clone()); // Apply all middleware layers including auth, above the router let authn_client = self.authn_client.lock().clone(); - router = self.apply_middleware_stack(router, authn_client)?; + router = self.apply_middleware_stack(router, authn_client.clone())?; let config = self.get_cached_config(); let prefix = Self::normalize_prefix_path(&config.prefix_path)?; - router = Self::apply_prefix_nesting(router, &prefix); + router = self.apply_prefix_nesting(router, &prefix, hc_registry, authn_client)?; // Cache the built router for future use self.router_cache.store(router.clone()); @@ -706,17 +741,24 @@ impl toolkit::Gear for ApiGateway { impl toolkit::contracts::ApiGatewayCapability for ApiGateway { fn rest_prepare( &self, - _ctx: &toolkit::context::GearCtx, + ctx: &toolkit::context::GearCtx, router: axum::Router, ) -> anyhow::Result { - // Add health check endpoints: - // - /health: detailed JSON response with status and timestamp - // - /healthz: simple "ok" liveness probe (Kubernetes-style) - let router = router - .route("/health", get(web::health_check)) - .route("/healthz", get(|| async { "ok" })); - - // You may attach global middlewares here (trace, compression, cors), but do not start server. + // Retrieve the healthcheck registry created in run_rest_phase before this call. + // Falls back to an empty registry so standalone tests still compile. + let hc_registry = ctx + .client_hub() + .get::() + .unwrap_or_else(|_| Arc::new(toolkit::RestHealthcheckRegistry::new())); + + *self.healthcheck_registry.lock() = Some(hc_registry.clone()); + + // Add health endpoints: + // - /healthz: shallow liveness probe, never runs user checks + // - /readyz: readiness probe, runs user REST healthchecks + // - /health: detailed readiness JSON with per-component status + let router = router.merge(Self::health_routes(hc_registry)); + tracing::debug!("REST host prepared base router with health check endpoints"); Ok(router) } @@ -735,10 +777,16 @@ impl toolkit::contracts::ApiGatewayCapability for ApiGateway { // Apply middleware stack (including auth) to the final router tracing::debug!("Applying middleware stack to finalized router"); let authn_client = self.authn_client.lock().clone(); - router = self.apply_middleware_stack(router, authn_client)?; + router = self.apply_middleware_stack(router, authn_client.clone())?; + + let hc_registry = self + .healthcheck_registry + .lock() + .clone() + .unwrap_or_else(|| Arc::new(toolkit::RestHealthcheckRegistry::new())); let prefix = Self::normalize_prefix_path(&config.prefix_path)?; - router = Self::apply_prefix_nesting(router, &prefix); + router = self.apply_prefix_nesting(router, &prefix, hc_registry, authn_client)?; // Keep the finalized router to be used by `serve()` *self.final_router.lock() = Some(router.clone()); diff --git a/gears/system/api-gateway/src/web.rs b/gears/system/api-gateway/src/web.rs index d24f95210..ea5c15f61 100644 --- a/gears/system/api-gateway/src/web.rs +++ b/gears/system/api-gateway/src/web.rs @@ -1,10 +1,15 @@ use axum::{ http::StatusCode, - response::{Html, Json}, + response::{Html, IntoResponse, Json, Response}, routing::{MethodRouter, get}, }; use chrono::{SecondsFormat, Utc}; -use serde_json::{Value, json}; +use serde_json::json; +use std::sync::Arc; +use std::time::Duration; +use toolkit::RestHealthcheckRegistry; + +const HEALTHCHECK_TIMEOUT: Duration = Duration::from_millis(500); /// Returns a 501 Not Implemented handler for operations without implementations #[allow(dead_code)] @@ -20,11 +25,36 @@ pub fn placeholder_handler_501() -> MethodRouter { }) } -pub async fn health_check() -> Json { - Json(json!({ - "status": "healthy", - "timestamp": Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true) - })) +/// Detailed readiness JSON endpoint (`GET /health`). +/// +/// Runs all registered REST healthchecks and returns a JSON report. +/// Returns 200 (Healthy or Degraded), 503 (Unhealthy). +pub async fn health_detail(registry: Arc) -> Response { + let report = registry.report(HEALTHCHECK_TIMEOUT).await; + let status_code = status_for_report(report.is_ready()); + let body = Json(json!({ + "status": report.status, + "timestamp": Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), + "components": report.components, + })); + (status_code, body).into_response() +} + +/// Readiness probe (`GET /readyz`). +/// +/// Deep check: runs all registered REST healthchecks. +/// Returns 200 (Healthy or Degraded), 503 (Unhealthy). +pub async fn readyz_check(registry: Arc) -> StatusCode { + let report = registry.report(HEALTHCHECK_TIMEOUT).await; + status_for_report(report.is_ready()) +} + +fn status_for_report(report_ready: bool) -> StatusCode { + if report_ready { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + } } #[cfg(not(feature = "embed_elements"))] diff --git a/gears/system/api-gateway/tests/health_endpoints.rs b/gears/system/api-gateway/tests/health_endpoints.rs new file mode 100644 index 000000000..e8ba3e97e --- /dev/null +++ b/gears/system/api-gateway/tests/health_endpoints.rs @@ -0,0 +1,348 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +//! Integration tests for health endpoints (/health, /healthz, /readyz). + +use async_trait::async_trait; +use axum::{ + Router, + body::Body, + http::{Request, StatusCode}, +}; +use serde_json::json; +use std::sync::Arc; +use toolkit::{ + ClientHub, Gear, Healthcheck, HealthcheckResult, RestApiCapability, contracts::OpenApiRegistry, +}; +use tower::ServiceExt; +use uuid::Uuid; + +// ---------- helpers ---------- + +struct TestConfigProvider; + +impl toolkit::config::ConfigProvider for TestConfigProvider { + fn get_gear_config(&self, _gear: &str) -> Option<&serde_json::Value> { + None + } +} + +fn test_ctx_with_hub(hub: Arc) -> toolkit::GearCtx { + toolkit::GearCtx::new( + "test-gear", + Uuid::new_v4(), + Arc::new(TestConfigProvider), + hub, + tokio_util::sync::CancellationToken::new(), + ) +} + +/// Build a router via `rest_prepare` with a pre-populated registry in the hub. +fn build_router(setup: impl FnOnce(&toolkit::RestHealthcheckRegistry)) -> Router { + use api_gateway::ApiGateway; + use toolkit::contracts::ApiGatewayCapability; + + let hub = Arc::new(ClientHub::new()); + let registry = Arc::new(toolkit::RestHealthcheckRegistry::new()); + setup(®istry); + hub.register::(registry); + + let ctx = test_ctx_with_hub(hub); + let gw = ApiGateway::default(); + gw.rest_prepare(&ctx, Router::new()) + .expect("rest_prepare failed") +} + +async fn get(router: Router, path: &str) -> axum::http::Response { + router + .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) + .await + .unwrap() +} + +async fn body_json(resp: axum::http::Response) -> serde_json::Value { + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + serde_json::from_slice(&bytes).unwrap() +} + +// ---------- check stubs ---------- + +struct HealthyCheck; +#[async_trait] +impl Healthcheck for HealthyCheck { + fn name(&self) -> &'static str { + "always-healthy" + } +} + +struct DegradedCheck; +#[async_trait] +impl Healthcheck for DegradedCheck { + fn name(&self) -> &'static str { + "always-degraded" + } + async fn check(&self) -> HealthcheckResult { + HealthcheckResult::degraded("cache warming up") + } +} + +struct UnhealthyCheck; +#[async_trait] +impl Healthcheck for UnhealthyCheck { + fn name(&self) -> &'static str { + "always-unhealthy" + } + async fn check(&self) -> HealthcheckResult { + HealthcheckResult::unhealthy("database unreachable") + } +} + +// ---------- /healthz ---------- + +#[tokio::test] +async fn healthz_returns_200_with_no_checks() { + let router = build_router(|_| {}); + let resp = get(router, "/healthz").await; + assert_eq!(resp.status(), StatusCode::OK); +} + +#[tokio::test] +async fn healthz_returns_200_even_when_unhealthy_check_registered() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(UnhealthyCheck)); + }); + let resp = get(router, "/healthz").await; + assert_eq!(resp.status(), StatusCode::OK); +} + +// ---------- /readyz ---------- + +#[tokio::test] +async fn readyz_returns_200_with_no_checks() { + let router = build_router(|_| {}); + let resp = get(router, "/readyz").await; + assert_eq!(resp.status(), StatusCode::OK); +} + +#[tokio::test] +async fn readyz_returns_200_when_all_healthy() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(HealthyCheck)); + }); + let resp = get(router, "/readyz").await; + assert_eq!(resp.status(), StatusCode::OK); +} + +#[tokio::test] +async fn readyz_returns_200_when_degraded() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(DegradedCheck)); + }); + let resp = get(router, "/readyz").await; + assert_eq!(resp.status(), StatusCode::OK); +} + +#[tokio::test] +async fn readyz_returns_503_when_unhealthy() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(UnhealthyCheck)); + }); + let resp = get(router, "/readyz").await; + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); +} + +// ---------- /health ---------- + +#[tokio::test] +async fn health_returns_json_with_components() { + let router = build_router(|reg| { + reg.register("my-gear", Arc::new(HealthyCheck)); + }); + let resp = get(router, "/health").await; + assert_eq!(resp.status(), StatusCode::OK); + let body = body_json(resp).await; + assert_eq!(body["status"], "healthy"); + assert!(body["timestamp"].is_string()); + let components = body["components"].as_array().unwrap(); + assert_eq!(components.len(), 1); + assert_eq!(components[0]["gear"], "my-gear"); + assert_eq!(components[0]["status"], "healthy"); +} + +#[tokio::test] +async fn health_returns_503_when_unhealthy() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(UnhealthyCheck)); + }); + let resp = get(router, "/health").await; + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = body_json(resp).await; + assert_eq!(body["status"], "unhealthy"); +} + +#[tokio::test] +async fn health_returns_degraded_status_when_degraded() { + let router = build_router(|reg| { + reg.register("gear", Arc::new(DegradedCheck)); + }); + let resp = get(router, "/health").await; + assert_eq!(resp.status(), StatusCode::OK); + let body = body_json(resp).await; + assert_eq!(body["status"], "degraded"); +} + +// ---------- outside prefix ---------- + +#[tokio::test] +async fn health_endpoints_accessible_outside_prefix() { + use api_gateway::ApiGateway; + use toolkit::contracts::ApiGatewayCapability; + + struct ConfigMap(serde_json::Value); + impl toolkit::config::ConfigProvider for ConfigMap { + fn get_gear_config(&self, gear: &str) -> Option<&serde_json::Value> { + self.0.get(gear) + } + } + + let config = json!({ + "api-gateway": { + "config": { + "bind_addr": "0.0.0.0:18081", + "auth_disabled": true, + "prefix_path": "/cf", + } + } + }); + + let hub = Arc::new(ClientHub::new()); + let registry = Arc::new(toolkit::RestHealthcheckRegistry::new()); + hub.register::(registry); + + let gw = ApiGateway::default(); + let ctx = toolkit::GearCtx::new( + "api-gateway", + Uuid::new_v4(), + Arc::new(ConfigMap(config)), + hub, + tokio_util::sync::CancellationToken::new(), + ); + gw.init(&ctx).await.expect("init failed"); + + let router = gw + .rest_prepare(&ctx, Router::new()) + .expect("rest_prepare failed"); + let router = gw + .rest_finalize(&ctx, router) + .expect("rest_finalize failed"); + + // Health endpoints must be reachable at root paths, not nested under /cf + let resp = router + .clone() + .oneshot( + Request::builder() + .uri("/healthz") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let resp = router + .clone() + .oneshot( + Request::builder() + .uri("/readyz") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let resp = router + .oneshot( + Request::builder() + .uri("/health") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); +} + +// ---------- RestApiCapability default ---------- + +struct MinimalGear; + +#[async_trait] +impl toolkit::Gear for MinimalGear { + async fn init(&self, _ctx: &toolkit::GearCtx) -> anyhow::Result<()> { + Ok(()) + } +} + +impl RestApiCapability for MinimalGear { + fn register_rest( + &self, + _ctx: &toolkit::GearCtx, + router: Router, + _openapi: &dyn OpenApiRegistry, + ) -> anyhow::Result { + Ok(router) + } + // healthcheck() is intentionally NOT overridden — tests the default +} + +#[test] +fn existing_rest_gear_compiles_without_healthcheck_impl() { + let gear = MinimalGear; + let ctx = toolkit::GearCtx::new( + "test", + Uuid::new_v4(), + Arc::new(TestConfigProvider), + Arc::new(ClientHub::new()), + tokio_util::sync::CancellationToken::new(), + ); + assert!(gear.healthcheck(&ctx).is_none()); +} + +struct GearWithHealthcheck; + +#[async_trait] +impl toolkit::Gear for GearWithHealthcheck { + async fn init(&self, _ctx: &toolkit::GearCtx) -> anyhow::Result<()> { + Ok(()) + } +} + +impl RestApiCapability for GearWithHealthcheck { + fn register_rest( + &self, + _ctx: &toolkit::GearCtx, + router: Router, + _openapi: &dyn OpenApiRegistry, + ) -> anyhow::Result { + Ok(router) + } + + fn healthcheck(&self, _ctx: &toolkit::GearCtx) -> Option> { + Some(Arc::new(HealthyCheck)) + } +} + +#[test] +fn gear_can_return_some_healthcheck() { + let gear = GearWithHealthcheck; + let ctx = toolkit::GearCtx::new( + "test", + Uuid::new_v4(), + Arc::new(TestConfigProvider), + Arc::new(ClientHub::new()), + tokio_util::sync::CancellationToken::new(), + ); + assert!(gear.healthcheck(&ctx).is_some()); +} diff --git a/libs/toolkit/src/contracts.rs b/libs/toolkit/src/contracts.rs index 127476d14..cdd9b9698 100644 --- a/libs/toolkit/src/contracts.rs +++ b/libs/toolkit/src/contracts.rs @@ -60,6 +60,25 @@ pub trait RestApiCapability: Send + Sync { router: Router, openapi: &dyn OpenApiRegistry, ) -> anyhow::Result; + + /// Return a readiness healthcheck for this gear, if any. + /// + /// Called once during the REST wiring phase, after [`register_rest`](Self::register_rest) + /// succeeds. The returned check is registered in [`RestHealthcheckRegistry`] and run on + /// every `/readyz` and `/health` request. + /// + /// Return `None` (the default) to opt out of readiness probing. + /// + /// # gRPC healthcheck + /// + /// TODO: Add gRPC healthcheck hook in a follow-up PR. + /// This PR intentionally implements REST readiness healthchecks only. + fn healthcheck( + &self, + _ctx: &crate::context::GearCtx, + ) -> Option> { + None + } } /// API Gateway capability: handles gateway hosting with prepare/finalize phases. diff --git a/libs/toolkit/src/healthcheck.rs b/libs/toolkit/src/healthcheck.rs new file mode 100644 index 000000000..e77aa9c97 --- /dev/null +++ b/libs/toolkit/src/healthcheck.rs @@ -0,0 +1,585 @@ +//! REST readiness healthcheck infrastructure for probing gear health. +//! +//! This module provides concurrent healthcheck execution with timeout protection. +//! Gears implement [`Healthcheck`] and register themselves via [`RestHealthcheckRegistry`]. +//! The API Gateway calls [`report()`](RestHealthcheckRegistry::report) on `/health` and `/readyz` +//! requests to aggregate per-component readiness status. +//! +//! # Usage +//! +//! ```rust,ignore +//! use async_trait::async_trait; +//! use std::sync::Arc; +//! use toolkit::{Healthcheck, HealthcheckResult, contracts::RestApiCapability}; +//! +//! struct MyHealthcheck; +//! +//! #[async_trait] +//! impl Healthcheck for MyHealthcheck { +//! fn name(&self) -> &'static str { +//! "my-gear-readiness" +//! } +//! +//! async fn check(&self) -> HealthcheckResult { +//! HealthcheckResult::healthy() +//! } +//! } +//! +//! impl RestApiCapability for MyGear { +//! fn healthcheck( +//! &self, +//! _ctx: &toolkit::context::GearCtx, +//! ) -> Option> { +//! Some(Arc::new(MyHealthcheck)) +//! } +//! +//! fn register_rest( +//! &self, +//! ctx: &toolkit::context::GearCtx, +//! router: axum::Router, +//! openapi: &dyn toolkit::contracts::OpenApiRegistry, +//! ) -> anyhow::Result { +//! Ok(router) +//! } +//! } +//! ``` +//! +//! # Kubernetes probe configuration +//! +//! ```yaml +//! livenessProbe: +//! httpGet: +//! path: /healthz +//! port: http +//! periodSeconds: 10 +//! timeoutSeconds: 2 +//! failureThreshold: 3 +//! +//! readinessProbe: +//! httpGet: +//! path: /readyz +//! port: http +//! periodSeconds: 5 +//! timeoutSeconds: 2 +//! failureThreshold: 3 +//! ``` +//! +//! `/healthz` is liveness — always shallow, never runs user checks. +//! `/readyz` is readiness — runs user REST healthchecks and removes the pod +//! from traffic when unhealthy without triggering a restart. + +use async_trait::async_trait; +use parking_lot::RwLock; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +/// Status of a single healthcheck. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum HealthcheckStatus { + Healthy, + Degraded, + Unhealthy, +} + +/// Aggregate readiness status across all registered healthchecks. +/// +/// Kept distinct from [`HealthcheckStatus`] even though the variants currently +/// match 1:1 (see `compute_aggregate`): per-check status and overall aggregate +/// status are different concepts and may diverge (e.g. an aggregate-only +/// variant for "no checks registered yet") without forcing an API change on +/// every [`Healthcheck`] implementor. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum HealthcheckAggregateStatus { + Healthy, + Degraded, + Unhealthy, +} + +/// Result returned by a single [`Healthcheck::check`] call. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HealthcheckResult { + pub status: HealthcheckStatus, + pub message: Option, +} + +impl HealthcheckResult { + #[must_use] + pub fn healthy() -> Self { + Self { + status: HealthcheckStatus::Healthy, + message: None, + } + } + + #[must_use] + pub fn degraded(message: impl Into) -> Self { + Self { + status: HealthcheckStatus::Degraded, + message: Some(message.into()), + } + } + + #[must_use] + pub fn unhealthy(message: impl Into) -> Self { + Self { + status: HealthcheckStatus::Unhealthy, + message: Some(message.into()), + } + } +} + +impl Default for HealthcheckResult { + fn default() -> Self { + Self::healthy() + } +} + +/// Trait implemented by gears that want to participate in readiness probing. +/// +/// The default implementation always returns healthy, so a gear only needs to +/// override `check` when it actually has something to probe. +/// +/// # Contract +/// +/// - `check` must not panic; panics are caught by [`RestHealthcheckRegistry`] and +/// converted to [`HealthcheckStatus::Unhealthy`]. +/// - `check` must complete within the configured per-check timeout +/// (default 500 ms); timeouts are converted to [`HealthcheckStatus::Unhealthy`]. +/// - `check` must never leak secrets, passwords, DSNs, or stack traces in its +/// message; the registry sanitises messages but defence-in-depth is preferred. +#[async_trait] +pub trait Healthcheck: Send + Sync + 'static { + fn name(&self) -> &'static str { + std::any::type_name::() + } + + async fn check(&self) -> HealthcheckResult { + HealthcheckResult::healthy() + } +} + +/// Per-component readiness report entry. +/// +/// Represents the result of a single gear's healthcheck execution, including +/// elapsed time and outcome status. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HealthcheckComponentReport { + /// Gear name. + pub gear: String, + /// Check name from [`Healthcheck::name`]. + pub check: String, + /// Health status (Healthy, Degraded, or Unhealthy). + pub status: HealthcheckStatus, + /// Optional message (sanitized for security). + pub message: Option, + /// Elapsed time in milliseconds. + pub latency_ms: u64, +} + +/// Aggregate readiness report produced by [`RestHealthcheckRegistry::report`]. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HealthcheckReport { + pub status: HealthcheckAggregateStatus, + pub components: Vec, +} + +impl HealthcheckReport { + /// Returns `true` when the service is ready to receive traffic. + /// + /// Both `Healthy` and `Degraded` are considered ready; only `Unhealthy` removes + /// the pod from the load-balancer pool without triggering a restart. + #[must_use] + pub fn is_ready(&self) -> bool { + self.status != HealthcheckAggregateStatus::Unhealthy + } +} + +struct RegistryEntry { + gear: &'static str, + check: Arc, +} + +/// Minimum interval between fanning out a fresh round of healthchecks. +/// Bursty `/health` and `/readyz` probes within this window reuse the last +/// computed report instead of re-running every registered check. +const REPORT_CACHE_TTL: Duration = Duration::from_millis(500); + +/// Registry of REST healthchecks populated during the REST wiring phase. +/// +/// Create one `Arc`, register it in `ClientHub`, then +/// call [`register`](Self::register) for each REST gear that provides a +/// [`Healthcheck`]. The API Gateway retrieves the same `Arc` and calls +/// [`report`](Self::report) on every `/readyz` and `/health` request. +#[derive(Default)] +pub struct RestHealthcheckRegistry { + entries: RwLock>, + cached_report: RwLock>, +} + +impl RestHealthcheckRegistry { + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Register a healthcheck for the given gear. + pub fn register(&self, gear: &'static str, check: Arc) { + self.entries.write().push(RegistryEntry { gear, check }); + } + + /// Run all registered healthchecks concurrently and return an aggregate report. + /// + /// Each check runs in its own `tokio::task::spawn` so panics are caught. + /// Checks that exceed `timeout_per_check` are cancelled and reported as + /// [`HealthcheckStatus::Unhealthy`]. + /// + /// Results are cached for [`REPORT_CACHE_TTL`] so bursty probes reuse the + /// most recent report instead of re-running every check on each request. + pub async fn report(&self, timeout_per_check: Duration) -> HealthcheckReport { + if let Some((ts, cached)) = self.cached_report.read().as_ref() + && ts.elapsed() < REPORT_CACHE_TTL + { + return cached.clone(); + } + + let entries: Vec<(_, _)> = { + let r = self.entries.read(); + r.iter().map(|e| (e.gear, e.check.clone())).collect() + }; + + let report = if entries.is_empty() { + HealthcheckReport { + status: HealthcheckAggregateStatus::Healthy, + components: vec![], + } + } else { + let handles: Vec<_> = entries + .into_iter() + .map(|(gear, check)| tokio::spawn(run_one_check(gear, check, timeout_per_check))) + .collect(); + + let mut components = Vec::with_capacity(handles.len()); + for h in handles { + match h.await { + Ok(component) => components.push(component), + Err(_join_err) => { + // The outer spawn panicked; record a generic unhealthy entry. + components.push(HealthcheckComponentReport { + gear: "unknown".to_owned(), + check: "unknown".to_owned(), + status: HealthcheckStatus::Unhealthy, + message: Some("health check failed".to_owned()), + latency_ms: 0, + }); + } + } + } + + let aggregate = compute_aggregate(&components); + HealthcheckReport { + status: aggregate, + components, + } + }; + + *self.cached_report.write() = Some((Instant::now(), report.clone())); + report + } +} + +async fn run_one_check( + gear: &'static str, + check: Arc, + timeout_per_check: Duration, +) -> HealthcheckComponentReport { + let name = check.name(); + let start = Instant::now(); + + let mut handle = tokio::spawn(async move { check.check().await }); + + let (status, message) = tokio::select! { + result = &mut handle => match result { + Ok(r) => (r.status, r.message.as_deref().map(sanitize_message)), + Err(_join_err) => ( + HealthcheckStatus::Unhealthy, + Some("health check failed".to_owned()), + ), + }, + () = tokio::time::sleep(timeout_per_check) => { + handle.abort(); + let _result = handle.await; + ( + HealthcheckStatus::Unhealthy, + Some("health check timed out".to_owned()), + ) + } + }; + + HealthcheckComponentReport { + gear: gear.to_owned(), + check: name.to_owned(), + status, + message, + latency_ms: u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX), + } +} + +fn compute_aggregate(components: &[HealthcheckComponentReport]) -> HealthcheckAggregateStatus { + let mut status = HealthcheckAggregateStatus::Healthy; + for c in components { + match c.status { + HealthcheckStatus::Unhealthy => return HealthcheckAggregateStatus::Unhealthy, + HealthcheckStatus::Degraded => { + status = HealthcheckAggregateStatus::Degraded; + } + HealthcheckStatus::Healthy => {} + } + } + status +} + +const MAX_MESSAGE_LEN: usize = 256; + +// Deliberately broad, defense-in-depth blocklist for a public/unauthenticated endpoint: +// entries like "tenant", "select ", "update ", "delete " can match benign operator +// diagnostics (e.g. "tenant lookup slow") and collapse the message to the generic +// "health check failed". That false-positive rate is accepted to avoid ever leaking +// credentials, DSNs, or query fragments to an unauthenticated caller. +static SUSPICIOUS_SUBSTRINGS: &[&str] = &[ + "password", + "passwd", + "secret", + "token", + "bearer", + "authorization", + "postgres://", + "postgresql://", + "mysql://", + "sqlite://", + "tenant", + "select ", + "insert ", + "update ", + "delete ", + "panic", + "stack backtrace", +]; + +fn sanitize_message(msg: &str) -> String { + if msg.len() > MAX_MESSAGE_LEN { + return "health check failed".to_owned(); + } + let lower = msg.to_lowercase(); + for sub in SUSPICIOUS_SUBSTRINGS { + if lower.contains(sub) { + return "health check failed".to_owned(); + } + } + msg.to_owned() +} + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::Duration; + use tokio::sync::Notify; + + struct AlwaysHealthy; + #[async_trait] + impl Healthcheck for AlwaysHealthy { + fn name(&self) -> &'static str { + "always-healthy" + } + } + + struct AlwaysDegraded; + #[async_trait] + impl Healthcheck for AlwaysDegraded { + fn name(&self) -> &'static str { + "always-degraded" + } + async fn check(&self) -> HealthcheckResult { + HealthcheckResult::degraded("cache warming") + } + } + + struct AlwaysUnhealthy; + #[async_trait] + impl Healthcheck for AlwaysUnhealthy { + fn name(&self) -> &'static str { + "always-unhealthy" + } + async fn check(&self) -> HealthcheckResult { + HealthcheckResult::unhealthy("database unreachable") + } + } + + struct SlowCheck; + #[async_trait] + impl Healthcheck for SlowCheck { + fn name(&self) -> &'static str { + "slow-check" + } + async fn check(&self) -> HealthcheckResult { + tokio::time::sleep(Duration::from_secs(10)).await; + HealthcheckResult::healthy() + } + } + + struct AbortTrackingCheck { + entered: Arc, + dropped: Arc, + } + + #[async_trait] + impl Healthcheck for AbortTrackingCheck { + fn name(&self) -> &'static str { + "abort-tracking-check" + } + async fn check(&self) -> HealthcheckResult { + struct DropFlag(Arc); + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + let _drop_flag = DropFlag(self.dropped.clone()); + self.entered.notify_one(); + std::future::pending::().await + } + } + + struct PanickingCheck; + #[async_trait] + impl Healthcheck for PanickingCheck { + fn name(&self) -> &'static str { + "panicking-check" + } + async fn check(&self) -> HealthcheckResult { + panic!("intentional panic in healthcheck"); + } + } + + #[test] + fn healthcheck_result_default_is_healthy() { + let r = HealthcheckResult::default(); + assert_eq!(r.status, HealthcheckStatus::Healthy); + assert!(r.message.is_none()); + } + + #[tokio::test] + async fn default_healthcheck_check_returns_healthy() { + let r = AlwaysHealthy.check().await; + assert_eq!(r.status, HealthcheckStatus::Healthy); + } + + #[tokio::test] + async fn empty_registry_report_is_healthy_and_ready() { + let registry = RestHealthcheckRegistry::new(); + let report = registry.report(Duration::from_millis(500)).await; + assert_eq!(report.status, HealthcheckAggregateStatus::Healthy); + assert!(report.is_ready()); + assert!(report.components.is_empty()); + } + + #[tokio::test] + async fn degraded_component_makes_aggregate_degraded_and_still_ready() { + let registry = RestHealthcheckRegistry::new(); + registry.register("my-gear", Arc::new(AlwaysDegraded)); + let report = registry.report(Duration::from_millis(500)).await; + assert_eq!(report.status, HealthcheckAggregateStatus::Degraded); + assert!(report.is_ready()); + } + + #[tokio::test] + async fn unhealthy_component_makes_aggregate_unhealthy_and_not_ready() { + let registry = RestHealthcheckRegistry::new(); + registry.register("my-gear", Arc::new(AlwaysUnhealthy)); + let report = registry.report(Duration::from_millis(500)).await; + assert_eq!(report.status, HealthcheckAggregateStatus::Unhealthy); + assert!(!report.is_ready()); + } + + #[tokio::test] + async fn slow_check_times_out_and_becomes_unhealthy() { + let registry = RestHealthcheckRegistry::new(); + registry.register("slow-gear", Arc::new(SlowCheck)); + let report = registry.report(Duration::from_millis(100)).await; + assert_eq!(report.status, HealthcheckAggregateStatus::Unhealthy); + let comp = &report.components[0]; + assert_eq!(comp.status, HealthcheckStatus::Unhealthy); + } + + #[tokio::test] + async fn timed_out_check_is_aborted() { + let registry = RestHealthcheckRegistry::new(); + let entered = Arc::new(Notify::new()); + let dropped = Arc::new(AtomicBool::new(false)); + registry.register( + "slow-gear", + Arc::new(AbortTrackingCheck { + entered: entered.clone(), + dropped: dropped.clone(), + }), + ); + + let report_task = + tokio::spawn(async move { registry.report(Duration::from_millis(10)).await }); + entered.notified().await; + let report = report_task.await.expect("report task panicked"); + + assert_eq!(report.status, HealthcheckAggregateStatus::Unhealthy); + assert!( + dropped.load(Ordering::SeqCst), + "timed-out healthcheck future must be dropped after abort" + ); + } + + #[tokio::test] + async fn panicking_check_becomes_unhealthy_without_panicking_caller() { + let registry = RestHealthcheckRegistry::new(); + registry.register("panic-gear", Arc::new(PanickingCheck)); + let report = registry.report(Duration::from_millis(500)).await; + assert_eq!(report.status, HealthcheckAggregateStatus::Unhealthy); + } + + #[test] + fn sensitive_message_is_sanitized() { + assert_eq!( + sanitize_message("connection to postgres://user:pass@host/db failed"), + "health check failed" + ); + assert_eq!( + sanitize_message("invalid token in header"), + "health check failed" + ); + assert_eq!( + sanitize_message("SELECT * FROM users failed"), + "health check failed" + ); + assert_eq!( + sanitize_message("thread panicked at some_file.rs:42"), + "health check failed" + ); + } + + #[test] + fn long_message_is_sanitized() { + let long = "x".repeat(257); + assert_eq!(sanitize_message(&long), "health check failed"); + } + + #[test] + fn clean_message_passes_through() { + assert_eq!( + sanitize_message("upstream service unavailable"), + "upstream service unavailable" + ); + } +} diff --git a/libs/toolkit/src/lib.rs b/libs/toolkit/src/lib.rs index f78d0770a..8976e30b8 100644 --- a/libs/toolkit/src/lib.rs +++ b/libs/toolkit/src/lib.rs @@ -136,6 +136,13 @@ pub use directory::{ ServiceInstanceInfo, }; +// REST healthcheck infrastructure +pub mod healthcheck; +pub use healthcheck::{ + Healthcheck, HealthcheckAggregateStatus, HealthcheckComponentReport, HealthcheckReport, + HealthcheckResult, HealthcheckStatus, RestHealthcheckRegistry, +}; + // GTS schema support pub mod gts; diff --git a/libs/toolkit/src/runtime/host_runtime.rs b/libs/toolkit/src/runtime/host_runtime.rs index 64cc87cee..a7fee5fb8 100644 --- a/libs/toolkit/src/runtime/host_runtime.rs +++ b/libs/toolkit/src/runtime/host_runtime.rs @@ -439,6 +439,12 @@ impl HostRuntime { // use host as the registry let registry: &dyn crate::contracts::OpenApiRegistry = host.as_registry(); + // Create the healthcheck registry and publish it to ClientHub so both + // the REST provider loop (below) and the API Gateway can access it. + let hc_registry = Arc::new(crate::healthcheck::RestHealthcheckRegistry::new()); + self.client_hub + .register::(hc_registry.clone()); + // 1) Host prepare: base Router / global middlewares / basic OAS meta router = host.rest_prepare(&host_ctx, router) @@ -463,6 +469,11 @@ impl HostRuntime { gear: e.name, source, })?; + + // Register the gear's readiness healthcheck after successful route registration. + if let Some(hc) = rest.healthcheck(&ctx) { + hc_registry.register(e.name, hc); + } } }