feat!: Add working conversion webhook with cert rotation

Cargo.toml

@@ -11,6 +11,7 @@ repository = "https://github.com/stackabletech/operator-rs"
 [workspace.dependencies]
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
 
+arc-swap = "1.7"
 axum = { version = "0.8.1", features = ["http2"] }
 chrono = { version = "0.4.38", default-features = false }
 clap = { version = "4.5.17", features = ["derive", "cargo", "env"] }

crates/stackable-certs/src/ca/consts.rs

@@ -1,6 +1,6 @@
 use stackable_operator::time::Duration;
 
-/// The default CA validity time span of one hour (3600 seconds).
+/// The default CA validity time span
 pub const DEFAULT_CA_VALIDITY: Duration = Duration::from_hours_unchecked(1);
 
 /// The root CA subject name containing only the common name.

crates/stackable-certs/src/ca/mod.rs

@@ -38,7 +38,7 @@ pub enum Error {
     #[snafu(display("failed to generate RSA signing key"))]
     GenerateRsaSigningKey { source: rsa::Error },
 
-    #[snafu(display("failed to generate ECDSA signign key"))]
+    #[snafu(display("failed to generate ECDSA signing key"))]
     GenerateEcdsaSigningKey { source: ecdsa::Error },
 
     #[snafu(display("failed to parse {subject:?} as subject"))]

crates/stackable-operator/CHANGELOG.md

@@ -34,7 +34,12 @@ All notable changes to this project will be documented in this file.
 ### Changed
 
 - Update `kube` to `1.1.0` ([#1049]).
-- BREAKING: Return type for `ListenerOperatorVolumeSourceBuilder::new()` is no onger a `Result` ([#1058]).
+- BREAKING: Return type for `ListenerOperatorVolumeSourceBuilder::new()` is no longer a `Result` ([#1058]).
+- BREAKING: Require two new CLI arguments: `--operator-namespace` and `-operator-service-name`.
+  These are required, so that the operator knows what Service it needs to enter as CRD conversion webhook ([#1066]).
+- BREAKING: The `ProductOperatorRun` used for CLI arguments has some field renamed for consistency ([#1066]):
+  - `telemetry_arguments` -> `telemetry`
+  - `cluster_info_opts` -> `cluster_info`
 
 ### Fixed
 
@@ -50,6 +55,7 @@ All notable changes to this project will be documented in this file.
 [#1058]: https://github.com/stackabletech/operator-rs/pull/1058
 [#1060]: https://github.com/stackabletech/operator-rs/pull/1060
 [#1064]: https://github.com/stackabletech/operator-rs/pull/1064
+[#1066]: https://github.com/stackabletech/operator-rs/pull/1066
 [#1068]: https://github.com/stackabletech/operator-rs/pull/1068
 [#1071]: https://github.com/stackabletech/operator-rs/pull/1071
 

crates/stackable-operator/src/cli.rs

@@ -116,7 +116,7 @@ use product_config::ProductConfigManager;
 use snafu::{ResultExt, Snafu};
 use stackable_telemetry::tracing::TelemetryOptions;
 
-use crate::{namespace::WatchNamespace, utils::cluster_info::KubernetesClusterInfoOpts};
+use crate::{namespace::WatchNamespace, utils::cluster_info::KubernetesClusterInfoOptions};
 
 pub const AUTHOR: &str = "Stackable GmbH - [email protected]";
 
@@ -163,10 +163,10 @@ pub enum Command<Run: Args = ProductOperatorRun> {
 /// Can be embedded into an extended argument set:
 ///
 /// ```rust
-/// # use stackable_operator::cli::{Command, ProductOperatorRun, ProductConfigPath};
+/// # use stackable_operator::cli::{Command, OperatorEnvironmentOptions, ProductOperatorRun, ProductConfigPath};
+/// # use stackable_operator::{namespace::WatchNamespace, utils::cluster_info::KubernetesClusterInfoOptions};
+/// # use stackable_telemetry::tracing::TelemetryOptions;
 /// use clap::Parser;
-/// use stackable_operator::{namespace::WatchNamespace, utils::cluster_info::KubernetesClusterInfoOpts};
-/// use stackable_telemetry::tracing::TelemetryOptions;
 ///
 /// #[derive(clap::Parser, Debug, PartialEq, Eq)]
 /// struct Run {
@@ -176,17 +176,36 @@ pub enum Command<Run: Args = ProductOperatorRun> {
 ///     common: ProductOperatorRun,
 /// }
 ///
-/// let opts = Command::<Run>::parse_from(["foobar-operator", "run", "--name", "foo", "--product-config", "bar", "--watch-namespace", "foobar", "--kubernetes-node-name", "baz"]);
+/// let opts = Command::<Run>::parse_from([
+///     "foobar-operator",
+///     "run",
+///     "--name",
+///     "foo",
+///     "--product-config",
+///     "bar",
+///     "--watch-namespace",
+///     "foobar",
+///     "--operator-namespace",
+///     "stackable-operators",
+///     "--operator-service-name",
+///     "foo-operator",
+///     "--kubernetes-node-name",
+///     "baz",
+/// ]);
 /// assert_eq!(opts, Command::Run(Run {
 ///     name: "foo".to_string(),
 ///     common: ProductOperatorRun {
 ///         product_config: ProductConfigPath::from("bar".as_ref()),
 ///         watch_namespace: WatchNamespace::One("foobar".to_string()),
-///         telemetry_arguments: TelemetryOptions::default(),
-///         cluster_info_opts: KubernetesClusterInfoOpts {
+///         telemetry: TelemetryOptions::default(),
+///         cluster_info: KubernetesClusterInfoOptions {
 ///             kubernetes_cluster_domain: None,
 ///             kubernetes_node_name: "baz".to_string(),
 ///         },
+///         operator_environment: OperatorEnvironmentOptions {
+///             operator_namespace: "stackable-operators".to_string(),
+///             operator_service_name: "foo-operator".to_string(),
+///         },
 ///     },
 /// }));
 /// ```
@@ -220,10 +239,13 @@ pub struct ProductOperatorRun {
     pub watch_namespace: WatchNamespace,
 
     #[command(flatten)]
-    pub telemetry_arguments: TelemetryOptions,
+    pub operator_environment: OperatorEnvironmentOptions,
+
+    #[command(flatten)]
+    pub telemetry: TelemetryOptions,
 
     #[command(flatten)]
-    pub cluster_info_opts: KubernetesClusterInfoOpts,
+    pub cluster_info: KubernetesClusterInfoOptions,
 }
 
 /// A path to a [`ProductConfigManager`] spec file
@@ -281,9 +303,25 @@ impl ProductConfigPath {
     }
 }
 
+#[derive(clap::Parser, Debug, PartialEq, Eq)]
+pub struct OperatorEnvironmentOptions {
+    /// The namespace the operator is running in, usually `stackable-operators`.
+    ///
+    /// Note that when running the operator on Kubernetes we recommend to use the
+    /// [downward API](https://kubernetes.io/docs/concepts/workloads/pods/downward-api/)
+    /// to let Kubernetes project the namespace as the `OPERATOR_NAMESPACE` env variable.
+    #[arg(long, env)]
+    pub operator_namespace: String,
+
+    /// The name of the service the operator is reachable at, usually
+    /// something like `<product>-operator`.
+    #[arg(long, env)]
+    pub operator_service_name: String,
+}
+
 #[cfg(test)]
 mod tests {
-    use std::{env, fs::File};
+    use std::fs::File;
 
     use clap::Parser;
     use rstest::*;
@@ -294,7 +332,6 @@ mod tests {
     const USER_PROVIDED_PATH: &str = "user_provided_path_properties.yaml";
     const DEPLOY_FILE_PATH: &str = "deploy_config_spec_properties.yaml";
     const DEFAULT_FILE_PATH: &str = "default_file_path_properties.yaml";
-    const WATCH_NAMESPACE: &str = "WATCH_NAMESPACE";
 
     #[test]
     fn verify_cli() {
@@ -381,16 +418,17 @@ mod tests {
 
     #[test]
     fn product_operator_run_watch_namespace() {
-        // clean env var to not interfere if already set
-        unsafe { env::remove_var(WATCH_NAMESPACE) };
-
         // cli with namespace
         let opts = ProductOperatorRun::parse_from([
             "run",
             "--product-config",
             "bar",
             "--watch-namespace",
             "foo",
+            "--operator-namespace",
+            "stackable-operators",
+            "--operator-service-name",
+            "foo-operator",
             "--kubernetes-node-name",
             "baz",
         ]);
@@ -399,11 +437,15 @@ mod tests {
             ProductOperatorRun {
                 product_config: ProductConfigPath::from("bar".as_ref()),
                 watch_namespace: WatchNamespace::One("foo".to_string()),
-                cluster_info_opts: KubernetesClusterInfoOpts {
+                cluster_info: KubernetesClusterInfoOptions {
                     kubernetes_cluster_domain: None,
                     kubernetes_node_name: "baz".to_string()
                 },
-                telemetry_arguments: Default::default(),
+                telemetry: Default::default(),
+                operator_environment: OperatorEnvironmentOptions {
+                    operator_namespace: "stackable-operators".to_string(),
+                    operator_service_name: "foo-operator".to_string(),
+                }
             }
         );
 
@@ -412,6 +454,10 @@ mod tests {
             "run",
             "--product-config",
             "bar",
+            "--operator-namespace",
+            "stackable-operators",
+            "--operator-service-name",
+            "foo-operator",
             "--kubernetes-node-name",
             "baz",
         ]);
@@ -420,33 +466,15 @@ mod tests {
             ProductOperatorRun {
                 product_config: ProductConfigPath::from("bar".as_ref()),
                 watch_namespace: WatchNamespace::All,
-                cluster_info_opts: KubernetesClusterInfoOpts {
-                    kubernetes_cluster_domain: None,
-                    kubernetes_node_name: "baz".to_string()
-                },
-                telemetry_arguments: Default::default(),
-            }
-        );
-
-        // env with namespace
-        unsafe { env::set_var(WATCH_NAMESPACE, "foo") };
-        let opts = ProductOperatorRun::parse_from([
-            "run",
-            "--product-config",
-            "bar",
-            "--kubernetes-node-name",
-            "baz",
-        ]);
-        assert_eq!(
-            opts,
-            ProductOperatorRun {
-                product_config: ProductConfigPath::from("bar".as_ref()),
-                watch_namespace: WatchNamespace::One("foo".to_string()),
-                cluster_info_opts: KubernetesClusterInfoOpts {
+                cluster_info: KubernetesClusterInfoOptions {
                     kubernetes_cluster_domain: None,
                     kubernetes_node_name: "baz".to_string()
                 },
-                telemetry_arguments: Default::default(),
+                telemetry: Default::default(),
+                operator_environment: OperatorEnvironmentOptions {
+                    operator_namespace: "stackable-operators".to_string(),
+                    operator_service_name: "foo-operator".to_string(),
+                }
             }
         );
     }

crates/stackable-operator/src/client.rs

@@ -21,7 +21,7 @@ use tracing::trace;
 
 use crate::{
     kvp::LabelSelectorExt,
-    utils::cluster_info::{KubernetesClusterInfo, KubernetesClusterInfoOpts},
+    utils::cluster_info::{KubernetesClusterInfo, KubernetesClusterInfoOptions},
 };
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -529,13 +529,13 @@ impl Client {
     /// use k8s_openapi::api::core::v1::Pod;
     /// use stackable_operator::{
     ///     client::{Client, initialize_operator},
-    ///     utils::cluster_info::KubernetesClusterInfoOpts,
+    ///     utils::cluster_info::KubernetesClusterInfoOptions,
     /// };
     ///
     /// #[tokio::main]
     /// async fn main() {
-    /// let cluster_info_opts = KubernetesClusterInfoOpts::parse();
-    /// let client = initialize_operator(None, &cluster_info_opts)
+    /// let cluster_info_options = KubernetesClusterInfoOptions::parse();
+    /// let client = initialize_operator(None, &cluster_info_options)
     ///     .await
     ///     .expect("Unable to construct client.");
     /// let watcher_config: watcher::Config =
@@ -652,7 +652,7 @@ where
 
 pub async fn initialize_operator(
     field_manager: Option<String>,
-    cluster_info_opts: &KubernetesClusterInfoOpts,
+    cluster_info_opts: &KubernetesClusterInfoOptions,
 ) -> Result<Client> {
     let kubeconfig: Config = kube::Config::infer()
         .await
@@ -687,10 +687,10 @@ mod tests {
     };
     use tokio::time::error::Elapsed;
 
-    use crate::utils::cluster_info::KubernetesClusterInfoOpts;
+    use crate::utils::cluster_info::KubernetesClusterInfoOptions;
 
-    async fn test_cluster_info_opts() -> KubernetesClusterInfoOpts {
-        KubernetesClusterInfoOpts {
+    async fn test_cluster_info_opts() -> KubernetesClusterInfoOptions {
+        KubernetesClusterInfoOptions {
             // We have to hard-code a made-up cluster domain,
             // since kubernetes_node_name (probably) won't be a valid Node that we can query.
             kubernetes_cluster_domain: Some(

crates/stackable-operator/src/utils/cluster_info.rs

@@ -16,32 +16,36 @@ pub struct KubernetesClusterInfo {
 }
 
 #[derive(clap::Parser, Debug, PartialEq, Eq)]
-pub struct KubernetesClusterInfoOpts {
+pub struct KubernetesClusterInfoOptions {
     /// Kubernetes cluster domain, usually this is `cluster.local`.
     // We are not using a default value here, as we query the cluster if it is not specified.
     #[arg(long, env)]
     pub kubernetes_cluster_domain: Option<DomainName>,
 
     /// Name of the Kubernetes Node that the operator is running on.
+    ///
+    /// Note that when running the operator on Kubernetes we recommend to use the
+    /// [downward API](https://kubernetes.io/docs/concepts/workloads/pods/downward-api/)
+    /// to let Kubernetes project the namespace as the `KUBERNETES_NODE_NAME` env variable.
     #[arg(long, env)]
     pub kubernetes_node_name: String,
 }
 
 impl KubernetesClusterInfo {
     pub async fn new(
         client: &Client,
-        cluster_info_opts: &KubernetesClusterInfoOpts,
+        cluster_info_opts: &KubernetesClusterInfoOptions,
     ) -> Result<Self, Error> {
         let cluster_domain = match cluster_info_opts {
-            KubernetesClusterInfoOpts {
+            KubernetesClusterInfoOptions {
                 kubernetes_cluster_domain: Some(cluster_domain),
                 ..
             } => {
                 tracing::info!(%cluster_domain, "Using configured Kubernetes cluster domain");
 
                 cluster_domain.clone()
             }
-            KubernetesClusterInfoOpts {
+            KubernetesClusterInfoOptions {
                 kubernetes_node_name: node_name,
                 ..
             } => {