feat: Add /ready endpoint to WebhookServer

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
 

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<Option<Duration>>,
-) -> 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<CustomResourceDefinition> = 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(())
 }

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.

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<AtomicBool>,
+}
+
+impl HealthCheck {
+    fn new(name: impl Into<String>) -> 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<HealthCheck>,
+}
+
+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<String>) -> 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());
+    }
+}

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;

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.

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<Box<dyn Webhook>>,
         options: WebhookServerOptions,
+        readiness_checks: HealthCheckRegistry,
     ) -> Result<Self> {
         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)

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();

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();