diff --git a/crates/stackable-operator/CHANGELOG.md b/crates/stackable-operator/CHANGELOG.md index e739a373a..cda478d2c 100644 --- a/crates/stackable-operator/CHANGELOG.md +++ b/crates/stackable-operator/CHANGELOG.md @@ -21,8 +21,15 @@ All notable changes to this project will be documented in this file. - BREAKING: `ClusterResources` now warns about `objectOverrides` entries that did not match any of the objects it created. To enable this, the signatures of `apply_deep_merge` and `ObjectOverrides::apply_to` needed to be adjusted ([#1264]). +### Removed + +- BREAKING: Removed `timeout_duration` parameter from `signal::crd_established`. It now waits indefinitely and + the timeout should be handled with a startup probe instead. As a consequence `DEFAULT_CRD_ESTABLISHED_TIMEOUT` + also got removed ([#1272]). + [#1264]: https://github.com/stackabletech/operator-rs/pull/1264 [#1269]: https://github.com/stackabletech/operator-rs/pull/1269 +[#1272]: https://github.com/stackabletech/operator-rs/pull/1272 ## [0.116.0] - 2026-08-14 diff --git a/crates/stackable-operator/src/utils/signal.rs b/crates/stackable-operator/src/utils/signal.rs index 135952426..f626b17c7 100644 --- a/crates/stackable-operator/src/utils/signal.rs +++ b/crates/stackable-operator/src/utils/signal.rs @@ -1,7 +1,6 @@ use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition; use kube::runtime::wait; use snafu::{ResultExt, Snafu}; -use stackable_shared::time::Duration; use tokio::{ signal::unix::{SignalKind, signal}, sync::watch, @@ -77,45 +76,41 @@ impl SignalWatcher<()> { } } -pub const DEFAULT_CRD_ESTABLISHED_TIMEOUT: Duration = Duration::from_secs(5); - #[derive(Debug, Snafu)] pub enum CrdEstablishedError { - #[snafu(display("failed to meet CRD established condition before the timeout elapsed"))] - TimeoutElapsed { source: tokio::time::error::Elapsed }, - - #[snafu(display("failed to await CRD established condition due to api error"))] - Api { source: kube::runtime::wait::Error }, + #[snafu(display( + "failed to await CRD established condition for {crd_name:?} due to api error" + ))] + Api { + source: kube::runtime::wait::Error, + crd_name: String, + }, } -/// Waits for a CRD named `crd_name` to be established before `timeout_duration` (or by default -/// [`DEFAULT_CRD_ESTABLISHED_TIMEOUT`]) is elapsed. +/// Waits for a CRD named `crd_name` to be established. /// /// The same caveats from [`conditions::is_crd_established`](wait::conditions::is_crd_established) /// apply here as well. /// /// ### Errors /// -/// This function returns errors either if the timeout elapsed without the condition being met or -/// when the underlying API returned errors (CRD is unknown to the Kubernetes API server or due to -/// missing permissions). -pub async fn crd_established( - client: &Client, - crd_name: &str, - timeout_duration: impl Into>, -) -> Result<(), CrdEstablishedError> { +/// This function returns errors when the underlying API returned errors +/// (CRD is unknown to the Kubernetes API server or due to missing permissions). +pub async fn crd_established(client: &Client, crd_name: &str) -> Result<(), CrdEstablishedError> { + tracing::info!( + k8s.crd.name = crd_name, + "Waiting for the custom resource definition to be established" + ); + let api: kube::Api = client.get_api(&()); - let crd_established = - wait::await_condition(api, crd_name, wait::conditions::is_crd_established()); - let _ = tokio::time::timeout( - *timeout_duration - .into() - .unwrap_or(DEFAULT_CRD_ESTABLISHED_TIMEOUT), - crd_established, - ) - .await - .context(TimeoutElapsedSnafu)? - .context(ApiSnafu)?; + wait::await_condition(api, crd_name, wait::conditions::is_crd_established()) + .await + .context(ApiSnafu { crd_name })?; + + tracing::info!( + k8s.crd.name = crd_name, + "The custom resource definition is established" + ); Ok(()) } diff --git a/crates/stackable-shared/CHANGELOG.md b/crates/stackable-shared/CHANGELOG.md index 37cef226e..488d0ce9e 100644 --- a/crates/stackable-shared/CHANGELOG.md +++ b/crates/stackable-shared/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- Add `health` module containing `HealthCheck` and `HealthCheckRegistry` for health endpoints ([#1272]). + +[#1272]: https://github.com/stackabletech/operator-rs/pull/1272 + ## [0.1.2] - 2026-07-06 Note: There are only dependency bumps in this release. diff --git a/crates/stackable-shared/src/health.rs b/crates/stackable-shared/src/health.rs new file mode 100644 index 000000000..abe77dec0 --- /dev/null +++ b/crates/stackable-shared/src/health.rs @@ -0,0 +1,114 @@ +use std::{ + fmt::Display, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; + +/// A single named check contributing to one health endpoint. +/// +/// A check only passes once [`HealthCheck::mark_passed`] has been called. +#[derive(Clone)] +pub struct HealthCheck { + name: String, + passed: Arc, +} + +impl HealthCheck { + fn new(name: impl Into) -> Self { + Self { + name: name.into(), + passed: Arc::new(AtomicBool::new(false)), + } + } + + pub fn mark_passed(&self) { + self.passed.store(true, Ordering::Release); + } + + fn passed(&self) -> bool { + self.passed.load(Ordering::Acquire) + } +} + +/// A set of checks to be used for a health endpoint a probe can call. +/// +/// # Example +/// +/// ``` +/// use stackable_shared::health::HealthCheckRegistry; +/// +/// let mut startup_checks = HealthCheckRegistry::new(); +/// let crds_established = startup_checks.register("crds-established"); +/// +/// assert!(!startup_checks.all_passed()); +/// crds_established.mark_passed(); +/// assert!(startup_checks.all_passed()); +/// ``` +#[derive(Default)] +pub struct HealthCheckRegistry { + checks: Vec, +} + +impl HealthCheckRegistry { + pub fn new() -> Self { + Self::default() + } + + /// Registers a new [`HealthCheck`] with the provided name and returns it. + pub fn register(&mut self, name: impl Into) -> HealthCheck { + let check = HealthCheck::new(name); + self.checks.push(check.clone()); + check + } + + /// Returns `true` if all the registered health checks have passed or no health checks are + /// registered. + pub fn all_passed(&self) -> bool { + self.checks.iter().all(HealthCheck::passed) + } +} + +impl Display for HealthCheckRegistry { + /// Renders one line per check, with the check's name and status only. Anything else, error + /// causes in particular, must not end up in a response to an unauthenticated endpoint. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.checks.is_empty() { + return writeln!(f, "[ok] no checks registered"); + } + + for check in &self.checks { + let status = if check.passed() { "ok" } else { "pending" }; + writeln!(f, "[{status}] {name}", name = check.name)?; + } + Ok(()) + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn passed_on_empty_registry() { + let registry = HealthCheckRegistry::new(); + + assert!(registry.all_passed()); + } + + #[test] + fn passed_only_once_every_check_is() { + let mut registry = HealthCheckRegistry::new(); + let crds = registry.register("crds-established"); + let migration = registry.register("database-migrated"); + + assert!(!registry.all_passed()); + + crds.mark_passed(); + assert!(!registry.all_passed()); + + migration.mark_passed(); + assert!(registry.all_passed()); + } +} diff --git a/crates/stackable-shared/src/lib.rs b/crates/stackable-shared/src/lib.rs index 767726d3d..59c3a80d5 100644 --- a/crates/stackable-shared/src/lib.rs +++ b/crates/stackable-shared/src/lib.rs @@ -2,6 +2,7 @@ //! workspace. pub mod crd; +pub mod health; pub mod secret; pub mod time; pub mod yaml; diff --git a/crates/stackable-webhook/CHANGELOG.md b/crates/stackable-webhook/CHANGELOG.md index bf49f4f19..601c75fc5 100644 --- a/crates/stackable-webhook/CHANGELOG.md +++ b/crates/stackable-webhook/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- BREAKING: The `WebhookServer` now serves a `/ready` endpoint for a startup probe. + For that, `WebhookServer::new` takes an additional `HealthCheckRegistry` argument ([#1272]). + +[#1272]: https://github.com/stackabletech/operator-rs/pull/1272 + ## [0.9.2] - 2026-07-06 Note: There are only dependency bumps in this release. diff --git a/crates/stackable-webhook/src/lib.rs b/crates/stackable-webhook/src/lib.rs index 10b52b0aa..37277a5df 100644 --- a/crates/stackable-webhook/src/lib.rs +++ b/crates/stackable-webhook/src/lib.rs @@ -11,13 +11,17 @@ //! //! For usage please look at the [`WebhookServer`] docs as well as the specific [`Webhook`] you are //! using. -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::Arc, +}; use ::x509_cert::Certificate; -use axum::{Router, routing::get}; +use axum::{Router, http::StatusCode, routing::get}; use futures_util::TryFutureExt; use k8s_openapi::ByteString; use snafu::{ResultExt, Snafu}; +use stackable_shared::health::HealthCheckRegistry; use stackable_telemetry::AxumTraceLayer; use tokio::{sync::mpsc, try_join}; use tower::ServiceBuilder; @@ -54,6 +58,7 @@ pub enum WebhookServerError { /// ### Example usage /// /// ``` +/// use stackable_shared::health::HealthCheckRegistry; /// use stackable_webhook::{WebhookServer, WebhookServerOptions, webhooks::Webhook}; /// use tokio::time::{Duration, sleep}; /// @@ -65,7 +70,10 @@ pub enum WebhookServerError { /// webhook_namespace: "my-namespace".to_owned(), /// webhook_service_name: "my-operator".to_owned(), /// }; -/// let webhook_server = WebhookServer::new(webhooks, webhook_options).await.unwrap(); +/// let readiness_checks = HealthCheckRegistry::new(); +/// let webhook_server = WebhookServer::new(webhooks, webhook_options, readiness_checks) +/// .await +/// .unwrap(); /// let shutdown_signal = sleep(Duration::from_millis(100)); /// /// webhook_server.run(shutdown_signal).await.unwrap(); @@ -111,6 +119,7 @@ impl WebhookServer { pub async fn new( webhooks: Vec>, options: WebhookServerOptions, + readiness_checks: HealthCheckRegistry, ) -> Result { tracing::trace!("create new webhook server"); @@ -132,12 +141,26 @@ impl WebhookServer { router = webhook.register_routes(router); } + // Create the route handler for the startup probe. + let readiness_checks = Arc::new(readiness_checks); + let ready_route = move || async move { + let status = if readiness_checks.all_passed() { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + }; + // The response body carries check names and their status. Error causes etc. go to the + // log, never into a response to not leak internal information to the public endpoint. + (status, readiness_checks.to_string()) + }; + let router = router // Enrich spans for routes added above. // Routes defined below it will not be instrumented to reduce noise. .layer(trace_service_builder) - // The health route is below the AxumTraceLayer so as not to be instrumented - .route("/health", get(|| async { "ok" })); + // The health and ready routes are below the AxumTraceLayer so as not to be instrumented + .route("/health", get(|| async { "ok" })) + .route("/ready", get(ready_route)); tracing::debug!("create TLS server"); let (tls_server, cert_rx) = TlsServer::new(router, &options) diff --git a/crates/stackable-webhook/src/webhooks/conversion_webhook.rs b/crates/stackable-webhook/src/webhooks/conversion_webhook.rs index 62b1a7e11..31853b40c 100644 --- a/crates/stackable-webhook/src/webhooks/conversion_webhook.rs +++ b/crates/stackable-webhook/src/webhooks/conversion_webhook.rs @@ -45,6 +45,7 @@ pub enum ConversionWebhookError { /// Client, /// core::admission::{AdmissionRequest, AdmissionResponse}, /// }, +/// shared::health::HealthCheckRegistry, /// }; /// use stackable_webhook::{ /// WebhookServer, @@ -72,9 +73,14 @@ pub enum ConversionWebhookError { /// ConversionWebhook::new(crds_and_handlers, client, conversion_webhook_options); /// /// let webhook_options = todo!(); -/// let webhook_server = WebhookServer::new(vec![Box::new(conversion_webhook)], webhook_options) -/// .await -/// .unwrap(); +/// let readiness_checks = HealthCheckRegistry::new(); +/// let webhook_server = WebhookServer::new( +/// vec![Box::new(conversion_webhook)], +/// webhook_options, +/// readiness_checks, +/// ) +/// .await +/// .unwrap(); /// let shutdown_signal = sleep(Duration::from_millis(100)); /// /// webhook_server.run(shutdown_signal).await.unwrap(); diff --git a/crates/stackable-webhook/src/webhooks/mutating_webhook.rs b/crates/stackable-webhook/src/webhooks/mutating_webhook.rs index 9f2b18a28..d9655b336 100644 --- a/crates/stackable-webhook/src/webhooks/mutating_webhook.rs +++ b/crates/stackable-webhook/src/webhooks/mutating_webhook.rs @@ -42,9 +42,12 @@ pub enum MutatingWebhookError { /// use k8s_openapi::api::{ /// admissionregistration::v1::MutatingWebhookConfiguration, apps::v1::StatefulSet, /// }; -/// use stackable_operator::kube::{ -/// Client, -/// core::admission::{AdmissionRequest, AdmissionResponse}, +/// use stackable_operator::{ +/// kube::{ +/// Client, +/// core::admission::{AdmissionRequest, AdmissionResponse}, +/// }, +/// shared::health::HealthCheckRegistry, /// }; /// use stackable_webhook::{ /// WebhookServer, @@ -73,9 +76,11 @@ pub enum MutatingWebhookError { /// )); /// /// let webhook_options = todo!(); -/// let webhook_server = WebhookServer::new(vec![mutating_webhook], webhook_options) -/// .await -/// .unwrap(); +/// let readiness_checks = HealthCheckRegistry::new(); +/// let webhook_server = +/// WebhookServer::new(vec![mutating_webhook], webhook_options, readiness_checks) +/// .await +/// .unwrap(); /// let shutdown_signal = sleep(Duration::from_millis(100)); /// /// webhook_server.run(shutdown_signal).await.unwrap();