feat: Add ConversionWebhookServer::with_maintainer

Author: Techassi

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

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`)

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" }

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

crates/stackable-operator/src/crd/maintainer.rs renamed to 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.
     ///
@@ -79,10 +82,10 @@ impl CustomResourceDefinitionMaintainer {
     ///
     /// ```no_run
     /// # use stackable_operator::crd::s3::{S3Connection, S3ConnectionVersion, S3Bucket, S3BucketVersion};
-    /// # use stackable_webhook::x509_cert::Certificate;
     /// # use tokio::sync::mpsc::channel;
+    /// # use x509_cert::Certificate;
     /// # use kube::Client;
-    /// use stackable_operator::crd::maintainer::{
+    /// use stackable_webhook::maintainer::{
     ///     CustomResourceDefinitionMaintainerOptions,
     ///     CustomResourceDefinitionMaintainer,
     /// };
@@ -91,9 +94,8 @@ impl CustomResourceDefinitionMaintainer {
     /// # async fn main() {
     /// # let (certificate_tx, certificate_rx) = channel(1);
     /// let options = CustomResourceDefinitionMaintainerOptions {
-    ///     operator_service_name: "my-service-name".to_owned(),
-    ///     operator_namespace: "my-namespace".to_owned(),
-    ///     field_manager: "my-operator".to_owned(),
+    ///     operator_name: "my-service-name",
+    ///     operator_namespace: "my-namespace",
     ///     webhook_https_port: 8443,
     ///     disabled: true,
     /// };
@@ -117,7 +119,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 +141,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 +174,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 +196,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 +211,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 +234,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,

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:
@@ -119,8 +125,8 @@ impl ConversionWebhookServer {
     ///
     /// let options = ConversionWebhookOptions {
     ///     socket_addr: ConversionWebhookServer::DEFAULT_SOCKET_ADDRESS,
-    ///     namespace: "stackable-operators".to_owned(),
-    ///     service_name: "product-operator".to_owned(),
+    ///     namespace: "stackable-operators",
+    ///     service_name: "product-operator",
     /// };
     ///
     /// let (conversion_webhook_server, _certificate_rx) =
@@ -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");