feat: Add ConversionWebhookServer::with_maintainer

Techassi · Techassi · commit f5a18416c4fb · 2025-10-08T16:39:23.000+02:00
diff --git a/crates/stackable-operator/Cargo.toml b/crates/stackable-operator/Cargo.toml
@@ -24,7 +24,7 @@ stackable-operator-derive = { path = "../stackable-operator-derive" }
 stackable-shared = { path = "../stackable-shared", features = ["chrono", "time"] }
 stackable-telemetry = { path = "../stackable-telemetry", optional = true, features = ["clap"] }
 stackable-versioned = { path = "../stackable-versioned", optional = true }
-stackable-webhook = { path = "../stackable-webhook", optional = true, features = ["maintainer"]}
+stackable-webhook = { path = "../stackable-webhook", optional = true}
 
 chrono.workspace = true
 clap.workspace = true
diff --git a/crates/stackable-operator/src/crd/mod.rs b/crates/stackable-operator/src/crd/mod.rs
@@ -7,8 +7,6 @@ use serde::{Deserialize, Serialize};
 pub mod authentication;
 pub mod git_sync;
 pub mod listener;
-#[cfg(feature = "webhook")]
-pub mod maintainer;
 pub mod s3;
 
 /// A reference to a product cluster (for example, a `ZookeeperCluster`)
diff --git a/crates/stackable-webhook/Cargo.toml b/crates/stackable-webhook/Cargo.toml
@@ -6,10 +6,6 @@ license.workspace = true
 edition.workspace = true
 repository.workspace = true
 
-[features]
-# Re-export selected items from the x509-cert crate needed by the CRD maintainer
-maintainer = []
-
 [dependencies]
 stackable-certs = { path = "../stackable-certs", features = ["rustls"] }
 stackable-shared = { path = "../stackable-shared" }
diff --git a/crates/stackable-webhook/src/lib.rs b/crates/stackable-webhook/src/lib.rs
@@ -43,18 +43,11 @@ use tower::ServiceBuilder;
 pub use crate::options::WebhookOptions;
 use crate::tls::TlsServer;
 
+pub mod maintainer;
 pub mod options;
 pub mod servers;
 pub mod tls;
 
-#[cfg(feature = "maintainer")]
-pub mod x509_cert {
-    pub use ::x509_cert::{
-        Certificate,
-        der::{EncodePem, Error, pem::LineEnding},
-    };
-}
-
 /// A generic webhook handler receiving a request and sending back a response.
 ///
 /// This trait is not intended to be implemented by external crates and this
diff --git a/crates/stackable-webhook/src/maintainer.rs b/crates/stackable-webhook/src/maintainer.rs
@@ -10,13 +10,16 @@ use kube::{
     api::{Patch, PatchParams},
 };
 use snafu::{ResultExt, Snafu, ensure};
-use stackable_webhook::x509_cert::{self, Certificate, EncodePem, LineEnding};
 use tokio::sync::{mpsc, oneshot};
+use x509_cert::{
+    Certificate,
+    der::{EncodePem, pem::LineEnding},
+};
 
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display("failed to encode CA certificate as PEM format"))]
-    EncodeCertificateAuthorityAsPem { source: x509_cert::Error },
+    EncodeCertificateAuthorityAsPem { source: x509_cert::der::Error },
 
     #[snafu(display("failed to send initial CRD reconcile heartbeat"))]
     SendInitialReconcileHeartbeat,
@@ -34,17 +37,17 @@ pub enum Error {
 ///
 /// - Apply the CRDs when starting up
 /// - Reconcile the CRDs when the conversion webhook certificate is rotated
-pub struct CustomResourceDefinitionMaintainer {
+pub struct CustomResourceDefinitionMaintainer<'a> {
     client: Client,
     certificate_rx: mpsc::Receiver<Certificate>,
 
     definitions: Vec<CustomResourceDefinition>,
-    options: CustomResourceDefinitionMaintainerOptions,
+    options: CustomResourceDefinitionMaintainerOptions<'a>,
 
     initial_reconcile_tx: oneshot::Sender<()>,
 }
 
-impl CustomResourceDefinitionMaintainer {
+impl<'a> CustomResourceDefinitionMaintainer<'a> {
     /// Creates and returns a new [`CustomResourceDefinitionMaintainer`] which manages one or more
     /// custom resource definitions.
     ///
@@ -117,7 +120,7 @@ impl CustomResourceDefinitionMaintainer {
         client: Client,
         certificate_rx: mpsc::Receiver<Certificate>,
         definitions: impl IntoIterator<Item = CustomResourceDefinition>,
-        options: CustomResourceDefinitionMaintainerOptions,
+        options: CustomResourceDefinitionMaintainerOptions<'a>,
     ) -> (Self, oneshot::Receiver<()>) {
         let (initial_reconcile_tx, initial_reconcile_rx) = oneshot::channel();
 
@@ -139,10 +142,9 @@ impl CustomResourceDefinitionMaintainer {
     /// [`std::task::Poll::Ready`] and thus doesn't consume any resources.
     pub async fn run(mut self) -> Result<(), Error> {
         let CustomResourceDefinitionMaintainerOptions {
-            operator_service_name,
             operator_namespace,
-            field_manager,
-            webhook_https_port: https_port,
+            webhook_https_port,
+            operator_name,
             disabled,
         } = self.options;
 
@@ -173,7 +175,7 @@ impl CustomResourceDefinitionMaintainer {
 
             let crd_api: Api<CustomResourceDefinition> = Api::all(self.client.clone());
 
-            for mut crd in self.definitions.iter().cloned() {
+            for crd in self.definitions.iter_mut() {
                 let crd_kind = &crd.spec.names.kind;
                 let crd_name = crd.name_any();
 
@@ -195,10 +197,10 @@ impl CustomResourceDefinitionMaintainer {
                         conversion_review_versions: vec!["v1".to_owned()],
                         client_config: Some(WebhookClientConfig {
                             service: Some(ServiceReference {
-                                name: operator_service_name.clone(),
-                                namespace: operator_namespace.clone(),
+                                name: operator_name.to_owned(),
+                                namespace: operator_namespace.to_owned(),
                                 path: Some(format!("/convert/{crd_name}")),
-                                port: Some(https_port.into()),
+                                port: Some(webhook_https_port.into()),
                             }),
                             // Here, ByteString takes care of encoding the provided content as
                             // base64.
@@ -210,7 +212,7 @@ impl CustomResourceDefinitionMaintainer {
 
                 // Deploy the updated CRDs using a server-side apply.
                 let patch = Patch::Apply(&crd);
-                let patch_params = PatchParams::apply(&field_manager);
+                let patch_params = PatchParams::apply(operator_name);
                 crd_api
                     .patch(&crd_name, &patch_params, &patch)
                     .await
@@ -233,15 +235,12 @@ impl CustomResourceDefinitionMaintainer {
 
 // TODO (@Techassi): Make this a builder instead
 /// This contains required options to customize a [`CustomResourceDefinitionMaintainer`].
-pub struct CustomResourceDefinitionMaintainerOptions {
-    /// The service name used by the operator/conversion webhook.
-    pub operator_service_name: String,
+pub struct CustomResourceDefinitionMaintainerOptions<'a> {
+    /// The service name used by the operator/conversion webhook and as a field manager.
+    pub operator_name: &'a str,
 
     /// The namespace the operator/conversion webhook runs in.
-    pub operator_namespace: String,
-
-    /// The name of the field manager used for the server-side apply.
-    pub field_manager: String,
+    pub operator_namespace: &'a str,
 
     /// The HTTPS port the conversion webhook listens on.
     pub webhook_https_port: u16,
diff --git a/crates/stackable-webhook/src/servers/conversion.rs b/crates/stackable-webhook/src/servers/conversion.rs
@@ -2,18 +2,22 @@ use std::{fmt::Debug, net::SocketAddr};
 
 use axum::{Json, Router, routing::post};
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
-use kube::ResourceExt;
 // Re-export this type because users of the conversion webhook server require
 // this type to write the handler function. Instead of importing this type from
 // kube directly, consumers can use this type instead. This also eliminates
 // keeping the kube dependency version in sync between here and the operator.
 pub use kube::core::conversion::ConversionReview;
+use kube::{Client, ResourceExt};
 use snafu::{ResultExt, Snafu};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 use tracing::instrument;
 use x509_cert::Certificate;
 
-use crate::{WebhookError, WebhookHandler, WebhookServer, options::WebhookOptions};
+use crate::{
+    WebhookError, WebhookHandler, WebhookServer,
+    maintainer::{CustomResourceDefinitionMaintainer, CustomResourceDefinitionMaintainerOptions},
+    options::WebhookOptions,
+};
 
 #[derive(Debug, Snafu)]
 pub enum ConversionWebhookError {
@@ -53,15 +57,15 @@ where
 
 // TODO: Add a builder, maybe with `bon`.
 #[derive(Debug)]
-pub struct ConversionWebhookOptions {
+pub struct ConversionWebhookOptions<'a> {
     /// The bind address to bind the HTTPS server to.
     pub socket_addr: SocketAddr,
 
     /// The namespace the operator/webhook is running in.
-    pub namespace: String,
+    pub namespace: &'a str,
 
     /// The name of the Kubernetes service which points to the operator/webhook.
-    pub service_name: String,
+    pub service_name: &'a str,
 }
 
 /// A ready-to-use CRD conversion webhook server.
@@ -77,6 +81,8 @@ impl ConversionWebhookServer {
     /// Creates and returns a new [`ConversionWebhookServer`], which expects POST requests being
     /// made to the `/convert/{CRD_NAME}` endpoint.
     ///
+    /// The TLS certificate is automatically generated and rotated.
+    ///
     /// ## Parameters
     ///
     /// This function expects the following parameters:
@@ -134,25 +140,25 @@ impl ConversionWebhookServer {
     #[instrument(name = "create_conversion_webhook_server", skip(crds_and_handlers))]
     pub async fn new<H>(
         crds_and_handlers: impl IntoIterator<Item = (CustomResourceDefinition, H)>,
-        options: ConversionWebhookOptions,
+        options: ConversionWebhookOptions<'_>,
     ) -> Result<(Self, mpsc::Receiver<Certificate>), ConversionWebhookError>
     where
         H: WebhookHandler<ConversionReview, ConversionReview> + Clone + Send + Sync + 'static,
     {
         tracing::debug!("create new conversion webhook server");
 
         let mut router = Router::new();
-        let mut crds = Vec::new();
+
         for (crd, handler) in crds_and_handlers {
             let crd_name = crd.name_any();
             let handler_fn = |Json(review): Json<ConversionReview>| async {
                 let review = handler.call(review);
                 Json(review)
             };
 
+            // TODO (@Techassi): Make this part of the trait mentioned above
             let route = format!("/convert/{crd_name}");
             router = router.route(&route, post(handler_fn));
-            crds.push(crd);
         }
 
         let ConversionWebhookOptions {
@@ -180,6 +186,61 @@ impl ConversionWebhookServer {
         Ok((Self(server), certificate_rx))
     }
 
+    /// Creates and returns a tuple consisting of a [`ConversionWebhookServer`], a [`CustomResourceDefinitionMaintainer`],
+    /// and a [`oneshot::Receiver`].
+    ///
+    /// See the referenced items for more details on usage.
+    pub async fn with_maintainer<'a, H>(
+        // TODO (@Techassi): Use a trait type here which can be used to build all part of the
+        // conversion webhook server and a CRD maintainer.
+        crds_and_handlers: impl IntoIterator<Item = (CustomResourceDefinition, H)> + Clone,
+        operator_name: &'a str,
+        operator_namespace: &'a str,
+        disable_maintainer: bool,
+        client: Client,
+    ) -> Result<
+        (
+            Self,
+            CustomResourceDefinitionMaintainer<'a>,
+            oneshot::Receiver<()>,
+        ),
+        ConversionWebhookError,
+    >
+    where
+        H: WebhookHandler<ConversionReview, ConversionReview> + Clone + Send + Sync + 'static,
+    {
+        let socket_addr = ConversionWebhookServer::DEFAULT_SOCKET_ADDRESS;
+
+        // TODO (@Techassi): These should be moved into a builder
+        let webhook_options = ConversionWebhookOptions {
+            namespace: operator_namespace,
+            service_name: operator_name,
+            socket_addr,
+        };
+
+        let (conversion_webhook_server, certificate_rx) =
+            Self::new(crds_and_handlers.clone(), webhook_options).await?;
+
+        let definitions = crds_and_handlers.into_iter().map(|(crd, _)| crd);
+
+        // TODO (@Techassi): These should be moved into a builder
+        let maintainer_options = CustomResourceDefinitionMaintainerOptions {
+            webhook_https_port: socket_addr.port(),
+            disabled: disable_maintainer,
+            operator_namespace,
+            operator_name,
+        };
+
+        let (maintainer, initial_reconcile_rx) = CustomResourceDefinitionMaintainer::new(
+            client,
+            certificate_rx,
+            definitions,
+            maintainer_options,
+        );
+
+        Ok((conversion_webhook_server, maintainer, initial_reconcile_rx))
+    }
+
     /// Runs the [`ConversionWebhookServer`] asynchronously.
     pub async fn run(self) -> Result<(), ConversionWebhookError> {
         tracing::info!("run conversion webhook server");
diff --git a/crates/stackable-webhook/src/tls/cert_resolver.rs b/crates/stackable-webhook/src/tls/cert_resolver.rs
@@ -59,23 +59,25 @@ pub struct CertificateResolver {
     current_certified_key: ArcSwap<CertifiedKey>,
     subject_alterative_dns_names: Arc<Vec<String>>,
 
-    cert_tx: mpsc::Sender<Certificate>,
+    certificate_tx: mpsc::Sender<Certificate>,
 }
 
 impl CertificateResolver {
     pub async fn new(
         subject_alterative_dns_names: Vec<String>,
-        cert_tx: mpsc::Sender<Certificate>,
+        certificate_tx: mpsc::Sender<Certificate>,
     ) -> Result<Self> {
         let subject_alterative_dns_names = Arc::new(subject_alterative_dns_names);
-        let certified_key =
-            Self::generate_new_certificate_inner(subject_alterative_dns_names.clone(), &cert_tx)
-                .await?;
+        let certified_key = Self::generate_new_certificate_inner(
+            subject_alterative_dns_names.clone(),
+            &certificate_tx,
+        )
+        .await?;
 
         Ok(Self {
             subject_alterative_dns_names,
             current_certified_key: ArcSwap::new(certified_key),
-            cert_tx,
+            certificate_tx,
         })
     }
 
@@ -90,7 +92,8 @@ impl CertificateResolver {
 
     async fn generate_new_certificate(&self) -> Result<Arc<CertifiedKey>> {
         let subject_alterative_dns_names = self.subject_alterative_dns_names.clone();
-        Self::generate_new_certificate_inner(subject_alterative_dns_names, &self.cert_tx).await
+        Self::generate_new_certificate_inner(subject_alterative_dns_names, &self.certificate_tx)
+            .await
     }
 
     /// Creates a new certificate and returns the certified key.
@@ -102,7 +105,7 @@ impl CertificateResolver {
     /// See [the relevant decision](https://github.com/stackabletech/decisions/issues/56)
     async fn generate_new_certificate_inner(
         subject_alterative_dns_names: Arc<Vec<String>>,
-        cert_tx: &mpsc::Sender<Certificate>,
+        certificate_tx: &mpsc::Sender<Certificate>,
     ) -> Result<Arc<CertifiedKey>> {
         // The certificate generations can take a while, so we use `spawn_blocking`
         let (cert, certified_key) = tokio::task::spawn_blocking(move || {
@@ -141,7 +144,7 @@ impl CertificateResolver {
         .await
         .context(TokioSpawnBlockingSnafu)??;
 
-        cert_tx
+        certificate_tx
             .send(cert)
             .await
             .map_err(|_err| CertificateResolverError::SendCertificateToChannel)?;
diff --git a/crates/stackable-webhook/src/tls/mod.rs b/crates/stackable-webhook/src/tls/mod.rs
@@ -88,14 +88,14 @@ impl TlsServer {
         router: Router,
         options: WebhookOptions,
     ) -> Result<(Self, mpsc::Receiver<Certificate>)> {
-        let (cert_tx, cert_rx) = mpsc::channel(1);
+        let (certificate_tx, certificate_rx) = mpsc::channel(1);
 
         let WebhookOptions {
             socket_addr,
             subject_alterative_dns_names,
         } = options;
 
-        let cert_resolver = CertificateResolver::new(subject_alterative_dns_names, cert_tx)
+        let cert_resolver = CertificateResolver::new(subject_alterative_dns_names, certificate_tx)
             .await
             .context(CreateCertificateResolverSnafu)?;
         let cert_resolver = Arc::new(cert_resolver);
@@ -117,7 +117,7 @@ impl TlsServer {
             router,
         };
 
-        Ok((tls_server, cert_rx))
+        Ok((tls_server, certificate_rx))
     }
 
     /// Runs the TLS server by listening for incoming TCP connections on the
