Merge remote-tracking branch 'origin/main' into feat/conversion-webhook

Author: sbernauer

Cargo.lock

@@ -26,6 +26,7 @@ educe =  { version = "0.6.0", default-features = false, features = ["Clone", "De
 either = "1.13.0"
 futures = "0.3.30"
 futures-util = "0.3.30"
+http = "1.3.1"
 indexmap = "2.5.0"
 indoc = "2.0.6"
 insta = { version= "1.40", features = ["glob"] }
@@ -37,11 +38,11 @@ k8s-openapi = { version = "0.25.0", default-features = false, features = ["schem
 # We use rustls instead of openssl for easier portability, e.g. so that we can build stackablectl without the need to vendor (build from source) openssl
 # We use ring instead of aws-lc-rs, as this currently fails to build in "make run-dev"
 kube = { version = "1.1.0", default-features = false, features = ["client", "jsonpatch", "runtime", "derive", "rustls-tls", "ring"] }
-opentelemetry = "0.29.1"
-opentelemetry_sdk = { version = "0.29.0", features = ["rt-tokio"] }
-opentelemetry-appender-tracing = "0.29.1"
-opentelemetry-otlp = "0.29.0"
-opentelemetry-semantic-conventions = "0.29.0"
+opentelemetry = "0.30.0"
+opentelemetry_sdk = { version = "0.30.0", features = ["rt-tokio"] }
+opentelemetry-appender-tracing = "0.30.1"
+opentelemetry-otlp = "0.30.0"
+opentelemetry-semantic-conventions = "0.30.0"
 p256 = { version = "0.13.2", features = ["ecdsa"] }
 paste = "1.0.15"
 pin-project = "1.1.5"
@@ -75,7 +76,7 @@ tower = { version = "0.5.1", features = ["util"] }
 tower-http = { version = "0.6.1", features = ["trace"] }
 tracing = "0.1.40"
 tracing-appender = "0.2.3"
-tracing-opentelemetry = "0.30.0"
+tracing-opentelemetry = "0.31.0"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter", "json"] }
 trybuild = "1.0.99"
 url = { version = "2.5.2", features = ["serde"] }

Cargo.toml

@@ -4,6 +4,33 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- The default Kubernetes cluster domain name is now fetched from the kubelet API unless explicitly configured ([#1068], [#1071])
+  This requires operators to have the RBAC permission to `get` `nodes/proxy` in the apiGroup "", an example RBAC rule could look like:
+
+  ```yaml
+  ---
+  apiVersion: rbac.authorization.k8s.io/v1
+  kind: ClusterRole
+  metadata:
+    name: operator-cluster-role
+  rules:
+    - apiGroups: [""]
+      resources: [nodes/proxy]
+      verbs: [get]
+  ```
+
+  In addition, they must be provided the environment variable `KUBERNETES_NODE_NAME` like this:
+
+  ```yaml
+  env:
+  - name: KUBERNETES_NODE_NAME
+    valueFrom:
+      fieldRef:
+        fieldPath: spec.nodeName
+  ```
+
 ### Changed
 
 - Update `kube` to `1.1.0` ([#1049]).
@@ -26,6 +53,8 @@ All notable changes to this project will be documented in this file.
 [#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
 
 ## [0.93.2] - 2025-05-26
 
@@ -151,7 +180,7 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Add Deployments to `ClusterResource`s ([#992]).
-- Add `DeploymentConditionBuilder`  ([#993]).
+- Add `DeploymentConditionBuilder` ([#993]).
 
 ### Changed
 
@@ -372,7 +401,7 @@ All notable changes to this project will be documented in this file.
 ### Fixed
 
 - BREAKING: `KeyValuePairs::insert` (as well as `Labels::`/`Annotations::` via it) now overwrites
-  the old value if the key already exists. Previously, `iter()` would return *both* values in
+  the old value if the key already exists. Previously, `iter()` would return _both_ values in
   lexicographical order (causing further conversions like `Into<BTreeMap>` to prefer the maximum
   value) ([#888]).
 
@@ -637,7 +666,7 @@ All notable changes to this project will be documented in this file.
 
 ### Changed
 
-- Implement `PartialEq` for most *Snafu* Error enums ([#757]).
+- Implement `PartialEq` for most _Snafu_ Error enums ([#757]).
 - Update Rust to 1.77 ([#759])
 
 ### Fixed
@@ -1388,7 +1417,7 @@ This is a rerelease of 0.25.1 which some last-minute incompatible API changes to
 ### Changed
 
 - Objects are now streamed rather than polled when waiting for them to be deleted ([#452]).
-- serde\_yaml 0.8.26 -> 0.9.9 ([#450])
+- serde_yaml 0.8.26 -> 0.9.9 ([#450])
 
 [#450]: https://github.com/stackabletech/operator-rs/pull/450
 [#452]: https://github.com/stackabletech/operator-rs/pull/452

crates/stackable-operator/CHANGELOG.md

@@ -28,6 +28,7 @@ dockerfile-parser.workspace = true
 either.workspace = true
 educe.workspace = true
 futures.workspace = true
+http.workspace = true
 indexmap.workspace = true
 json-patch.workspace = true
 k8s-openapi.workspace = true

crates/stackable-operator/Cargo.toml

@@ -165,7 +165,7 @@ pub enum Command<Run: Args = ProductOperatorRun> {
 /// ```rust
 /// # use stackable_operator::cli::{Command, OperatorEnvironmentOpts, ProductOperatorRun, ProductConfigPath};
 /// use clap::Parser;
-/// use stackable_operator::namespace::WatchNamespace;
+/// use stackable_operator::{namespace::WatchNamespace, utils::cluster_info::KubernetesClusterInfoOpts};
 /// use stackable_telemetry::tracing::TelemetryOptions;
 ///
 /// #[derive(clap::Parser, Debug, PartialEq, Eq)]
@@ -189,14 +189,19 @@ pub enum Command<Run: Args = ProductOperatorRun> {
 ///     "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: Default::default(),
+///         cluster_info_opts: KubernetesClusterInfoOpts {
+///             kubernetes_cluster_domain: None,
+///             kubernetes_node_name: "baz".to_string(),
+///         },
 ///         operator_environment: OperatorEnvironmentOpts {
 ///             operator_namespace: "stackable-operators".to_string(),
 ///             operator_service_name: "foo-operator".to_string(),
@@ -326,6 +331,7 @@ mod tests {
     const WATCH_NAMESPACE: &str = "WATCH_NAMESPACE";
     const OPERATOR_NAMESPACE: &str = "OPERATOR_NAMESPACE";
     const OPERATOR_SERVICE_NAME: &str = "OPERATOR_SERVICE_NAME";
+    const KUBERNETES_NODE_NAME: &str = "KUBERNETES_NODE_NAME";
 
     #[test]
     fn verify_cli() {
@@ -426,13 +432,18 @@ mod tests {
             "stackable-operators",
             "--operator-service-name",
             "foo-operator",
+            "--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: Default::default(),
+                cluster_info_opts: KubernetesClusterInfoOpts {
+                    kubernetes_cluster_domain: None,
+                    kubernetes_node_name: "baz".to_string()
+                },
                 telemetry_arguments: Default::default(),
                 operator_environment: OperatorEnvironmentOpts {
                     operator_namespace: "stackable-operators".to_string(),
@@ -450,13 +461,18 @@ mod tests {
             "stackable-operators",
             "--operator-service-name",
             "foo-operator",
+            "--kubernetes-node-name",
+            "baz",
         ]);
         assert_eq!(
             opts,
             ProductOperatorRun {
                 product_config: ProductConfigPath::from("bar".as_ref()),
                 watch_namespace: WatchNamespace::All,
-                cluster_info_opts: Default::default(),
+                cluster_info_opts: KubernetesClusterInfoOpts {
+                    kubernetes_cluster_domain: None,
+                    kubernetes_node_name: "baz".to_string()
+                },
                 telemetry_arguments: Default::default(),
                 operator_environment: OperatorEnvironmentOpts {
                     operator_namespace: "stackable-operators".to_string(),
@@ -469,14 +485,18 @@ mod tests {
         unsafe { env::set_var(WATCH_NAMESPACE, "foo") };
         unsafe { env::set_var(OPERATOR_SERVICE_NAME, "foo-operator") };
         unsafe { env::set_var(OPERATOR_NAMESPACE, "stackable-operators") };
+        unsafe { env::set_var(KUBERNETES_NODE_NAME, "baz") };
 
         let opts = ProductOperatorRun::parse_from(["run", "--product-config", "bar"]);
         assert_eq!(
             opts,
             ProductOperatorRun {
                 product_config: ProductConfigPath::from("bar".as_ref()),
                 watch_namespace: WatchNamespace::One("foo".to_string()),
-                cluster_info_opts: Default::default(),
+                cluster_info_opts: KubernetesClusterInfoOpts {
+                    kubernetes_cluster_domain: None,
+                    kubernetes_node_name: "baz".to_string()
+                },
                 telemetry_arguments: Default::default(),
                 operator_environment: OperatorEnvironmentOpts {
                     operator_namespace: "stackable-operators".to_string(),

crates/stackable-operator/src/cli.rs

@@ -84,6 +84,11 @@ pub enum Error {
 
     #[snafu(display("unable to create kubernetes client"))]
     CreateKubeClient { source: kube::Error },
+
+    #[snafu(display("unable to fetch cluster information from kubelet"))]
+    NewKubeletClusterInfo {
+        source: crate::utils::cluster_info::Error,
+    },
 }
 
 /// This `Client` can be used to access Kubernetes.
@@ -518,15 +523,19 @@ impl Client {
     ///
     /// ```no_run
     /// use std::time::Duration;
+    /// use clap::Parser;
     /// use tokio::time::error::Elapsed;
     /// use kube::runtime::watcher;
     /// use k8s_openapi::api::core::v1::Pod;
-    /// use stackable_operator::client::{Client, initialize_operator};
+    /// use stackable_operator::{
+    ///     client::{Client, initialize_operator},
+    ///     utils::cluster_info::KubernetesClusterInfoOpts,
+    /// };
     ///
     /// #[tokio::main]
     /// async fn main() {
-    ///
-    /// let client = initialize_operator(None, &Default::default())
+    /// let cluster_info_opts = KubernetesClusterInfoOpts::parse();
+    /// let client = initialize_operator(None, &cluster_info_opts)
     ///     .await
     ///     .expect("Unable to construct client.");
     /// let watcher_config: watcher::Config =
@@ -651,7 +660,9 @@ pub async fn initialize_operator(
         .context(InferKubeConfigSnafu)?;
     let default_namespace = kubeconfig.default_namespace.clone();
     let client = kube::Client::try_from(kubeconfig).context(CreateKubeClientSnafu)?;
-    let cluster_info = KubernetesClusterInfo::new(cluster_info_opts);
+    let cluster_info = KubernetesClusterInfo::new(&client, cluster_info_opts)
+        .await
+        .context(NewKubeletClusterInfoSnafu)?;
 
     Ok(Client::new(
         client,
@@ -676,10 +687,26 @@ mod tests {
     };
     use tokio::time::error::Elapsed;
 
+    use crate::utils::cluster_info::KubernetesClusterInfoOpts;
+
+    async fn test_cluster_info_opts() -> KubernetesClusterInfoOpts {
+        KubernetesClusterInfoOpts {
+            // 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(
+                "fake-cluster.local"
+                    .parse()
+                    .expect("hard-coded cluster domain must be valid"),
+            ),
+            // Tests aren't running in a kubelet, so make up a name of one.
+            kubernetes_node_name: "fake-node-name".to_string(),
+        }
+    }
+
     #[tokio::test]
     #[ignore = "Tests depending on Kubernetes are not ran by default"]
     async fn k8s_test_wait_created() {
-        let client = super::initialize_operator(None, &Default::default())
+        let client = super::initialize_operator(None, &test_cluster_info_opts().await)
             .await
             .expect("KUBECONFIG variable must be configured.");
 
@@ -757,7 +784,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "Tests depending on Kubernetes are not ran by default"]
     async fn k8s_test_wait_created_timeout() {
-        let client = super::initialize_operator(None, &Default::default())
+        let client = super::initialize_operator(None, &test_cluster_info_opts().await)
             .await
             .expect("KUBECONFIG variable must be configured.");
 
@@ -777,7 +804,7 @@ mod tests {
     #[tokio::test]
     #[ignore = "Tests depending on Kubernetes are not ran by default"]
     async fn k8s_test_list_with_label_selector() {
-        let client = super::initialize_operator(None, &Default::default())
+        let client = super::initialize_operator(None, &test_cluster_info_opts().await)
             .await
             .expect("KUBECONFIG variable must be configured.");
 

crates/stackable-operator/src/client.rs

@@ -1,45 +1,62 @@
-use std::str::FromStr;
+use kube::Client;
+use snafu::{ResultExt, Snafu};
 
-use crate::commons::networking::DomainName;
+use crate::{commons::networking::DomainName, utils::kubelet};
 
-const KUBERNETES_CLUSTER_DOMAIN_DEFAULT: &str = "cluster.local";
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("unable to fetch kubelet config"))]
+    KubeletConfig { source: kubelet::Error },
+}
 
-/// Some information that we know about the Kubernetes cluster.
 #[derive(Debug, Clone)]
 pub struct KubernetesClusterInfo {
     /// The Kubernetes cluster domain, typically `cluster.local`.
     pub cluster_domain: DomainName,
 }
 
-#[derive(clap::Parser, Debug, Default, PartialEq, Eq)]
+#[derive(clap::Parser, Debug, PartialEq, Eq)]
 pub struct KubernetesClusterInfoOpts {
     /// Kubernetes cluster domain, usually this is `cluster.local`.
-    // We are not using a default value here, as operators will probably do an more advanced
-    // auto-detection of the cluster domain in case it is not specified in the future.
+    // 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.
+    #[arg(long, env)]
+    pub kubernetes_node_name: String,
 }
 
 impl KubernetesClusterInfo {
-    pub fn new(cluster_info_opts: &KubernetesClusterInfoOpts) -> Self {
-        let cluster_domain = match &cluster_info_opts.kubernetes_cluster_domain {
-            Some(cluster_domain) => {
+    pub async fn new(
+        client: &Client,
+        cluster_info_opts: &KubernetesClusterInfoOpts,
+    ) -> Result<Self, Error> {
+        let cluster_domain = match cluster_info_opts {
+            KubernetesClusterInfoOpts {
+                kubernetes_cluster_domain: Some(cluster_domain),
+                ..
+            } => {
                 tracing::info!(%cluster_domain, "Using configured Kubernetes cluster domain");
 
                 cluster_domain.clone()
             }
-            None => {
-                // TODO(sbernauer): Do some sort of advanced auto-detection, see https://github.com/stackabletech/issues/issues/436.
-                // There have been attempts of parsing the `/etc/resolv.conf`, but they have been
-                // reverted. Please read on the linked issue for details.
-                let cluster_domain = DomainName::from_str(KUBERNETES_CLUSTER_DOMAIN_DEFAULT)
-                    .expect("KUBERNETES_CLUSTER_DOMAIN_DEFAULT constant must a valid domain");
-                tracing::info!(%cluster_domain, "Defaulting Kubernetes cluster domain as it has not been configured");
+            KubernetesClusterInfoOpts {
+                kubernetes_node_name: node_name,
+                ..
+            } => {
+                tracing::info!(%node_name, "Fetching Kubernetes cluster domain from the local kubelet");
+                let kubelet_config = kubelet::KubeletConfig::fetch(client, node_name)
+                    .await
+                    .context(KubeletConfigSnafu)?;
+
+                let cluster_domain = kubelet_config.cluster_domain;
+                tracing::info!(%cluster_domain, "Using Kubernetes cluster domain from the kubelet config");
 
                 cluster_domain
             }
         };
 
-        Self { cluster_domain }
+        Ok(Self { cluster_domain })
     }
 }