WIP

Author: sbernauer

Cargo.lock

@@ -25,6 +25,7 @@ ecdsa = { version = "0.16.9", features = ["digest", "pem"] }
 either = "1.13.0"
 futures = "0.3.30"
 futures-util = "0.3.30"
+indexmap = "2.5"
 hyper = { version = "1.4.1", features = ["full"] }
 hyper-util = "0.1.8"
 itertools = "0.13.0"

Cargo.toml

@@ -21,6 +21,7 @@ derivative.workspace = true
 dockerfile-parser.workspace = true
 either.workspace = true
 futures.workspace = true
+indexmap.workspace = true
 json-patch.workspace = true
 k8s-openapi.workspace = true
 kube.workspace = true

crates/stackable-operator/Cargo.toml

@@ -1,5 +1,6 @@
 use std::fmt;
 
+use indexmap::IndexMap;
 use k8s_openapi::api::core::v1::{
     ConfigMapKeySelector, Container, ContainerPort, EnvVar, EnvVarSource, Lifecycle,
     LifecycleHandler, ObjectFieldSelector, Probe, ResourceRequirements, SecretKeySelector,
@@ -26,6 +27,11 @@ pub enum Error {
 /// A builder to build [`Container`] objects.
 ///
 /// This will automatically create the necessary volumes and mounts for each `ConfigMap` which is added.
+///
+/// This struct is often times using an [`IndexMap`] to have consistent ordering (so we don't produce reconcile loops).
+/// We are also choosing it over an [`std::collections::BTreeMap`], as it's easier to debug for users, as logically
+/// grouped volumeMounts (e.g. all volumeMounts related to S3) are near each other in the list instead of "just" being
+/// sorted alphabetically.
 #[derive(Clone, Default)]
 pub struct ContainerBuilder {
     args: Option<Vec<String>>,
@@ -36,7 +42,9 @@ pub struct ContainerBuilder {
     image_pull_policy: Option<String>,
     name: String,
     resources: Option<ResourceRequirements>,
-    volume_mounts: Option<Vec<VolumeMount>>,
+
+    /// The key is the volumeMount mountPath.
+    volume_mounts: IndexMap<String, VolumeMount>,
     readiness_probe: Option<Probe>,
     liveness_probe: Option<Probe>,
     startup_probe: Option<Probe>,
@@ -188,28 +196,55 @@ impl ContainerBuilder {
         self
     }
 
+    /// This function only adds the [`VolumeMount`] in case there is no volumeMount with the same mountPath already.
+    /// In case there already was a volumeMount with the same path already, an [`tracing::error`] is raised in case the
+    /// contents of the volumeMounts differ.
+    ///
+    /// Historically this function unconditionally added volumeMounts, which resulted in invalid
+    /// [`k8s_openapi::api::core::v1::PodSpec`]s, as volumeMounts where added multiple times - think of Trino using the same [`crate::commons::s3::S3Connection`]
+    /// two times, resulting in e.g. the s3 credentials being mounted twice as the same volumeMount.
+    ///
+    /// We could have made this function fallible, but decided not to do so for now, as it would be a bigger breaking
+    /// change.
+    pub fn add_volume_mount_struct(&mut self, volume_mount: VolumeMount) -> &mut Self {
+        if let Some(existing_volume_mount) = self.volume_mounts.get(&volume_mount.mount_path) {
+            if !existing_volume_mount.eq(&volume_mount) {
+                tracing::error!(
+                    ?existing_volume_mount,
+                    new_volume_mount = ?volume_mount,
+                    "The operator tried to add a volumeMount to Container, which was already added with a different content! \
+                        The new volumeMount will be ignored."
+                );
+            }
+        } else {
+            self.volume_mounts
+                .insert(volume_mount.mount_path.clone(), volume_mount);
+        }
+
+        self
+    }
+
+    /// See [`Self::add_volume_mount_struct`] for details
     pub fn add_volume_mount(
         &mut self,
         name: impl Into<String>,
         path: impl Into<String>,
     ) -> &mut Self {
-        self.volume_mounts
-            .get_or_insert_with(Vec::new)
-            .push(VolumeMount {
-                name: name.into(),
-                mount_path: path.into(),
-                ..VolumeMount::default()
-            });
-        self
+        self.add_volume_mount_struct(VolumeMount {
+            name: name.into(),
+            mount_path: path.into(),
+            ..VolumeMount::default()
+        })
     }
 
+    /// See [`Self::add_volume_mount_struct`] for details
     pub fn add_volume_mounts(
         &mut self,
         volume_mounts: impl IntoIterator<Item = VolumeMount>,
     ) -> &mut Self {
-        self.volume_mounts
-            .get_or_insert_with(Vec::new)
-            .extend(volume_mounts);
+        for volume_mount in volume_mounts {
+            self.add_volume_mount_struct(volume_mount);
+        }
         self
     }
 
@@ -265,7 +300,11 @@ impl ContainerBuilder {
             resources: self.resources.clone(),
             name: self.name.clone(),
             ports: self.container_ports.clone(),
-            volume_mounts: self.volume_mounts.clone(),
+            volume_mounts: if self.volume_mounts.is_empty() {
+                None
+            } else {
+                Some(self.volume_mounts.clone().into_values().collect())
+            },
             readiness_probe: self.readiness_probe.clone(),
             liveness_probe: self.liveness_probe.clone(),
             startup_probe: self.startup_probe.clone(),

crates/stackable-operator/src/builder/pod/container.rs

@@ -1,5 +1,6 @@
 use std::{collections::BTreeMap, num::TryFromIntError};
 
+use indexmap::IndexMap;
 use k8s_openapi::{
     api::core::v1::{
         Affinity, Container, LocalObjectReference, NodeAffinity, Pod, PodAffinity, PodAntiAffinity,
@@ -53,6 +54,10 @@ pub enum Error {
 }
 
 /// A builder to build [`Pod`] or [`PodTemplateSpec`] objects.
+///
+/// This struct is often times using an [`IndexMap`] to have consistent ordering (so we don't produce reconcile loops).
+/// We are also choosing it over an [`BTreeMap`], as it's easier to debug for users, as logically grouped volumes
+/// (e.g. all volumes related to S3) are near each other in the list instead of "just" being sorted alphabetically.
 #[derive(Clone, Debug, Default, PartialEq)]
 pub struct PodBuilder {
     containers: Vec<Container>,
@@ -67,7 +72,9 @@ pub struct PodBuilder {
     status: Option<PodStatus>,
     security_context: Option<PodSecurityContext>,
     tolerations: Option<Vec<Toleration>>,
-    volumes: Option<Vec<Volume>>,
+
+    /// The key is the volume name.
+    volumes: IndexMap<String, Volume>,
     service_account_name: Option<String>,
     image_pull_secrets: Option<Vec<LocalObjectReference>>,
     restart_policy: Option<String>,
@@ -254,11 +261,6 @@ impl PodBuilder {
         self
     }
 
-    pub fn add_volume(&mut self, volume: Volume) -> &mut Self {
-        self.volumes.get_or_insert_with(Vec::new).push(volume);
-        self
-    }
-
     /// Utility function for the common case of adding an emptyDir Volume
     /// with the given name and no medium and no quantity.
     pub fn add_empty_dir_volume(
@@ -273,8 +275,39 @@ impl PodBuilder {
         )
     }
 
+    /// This function only adds the [`Volume`] in case there is no volume with the same name already.
+    /// In case there already was a volume with the same name already, an [`tracing::error`] is raised in case the
+    /// contents of the volumes differ.
+    ///
+    /// Historically this function unconditionally added volumes, which resulted in invalid [`PodSpec`]s, as volumes
+    /// where added multiple times - think of Trino using the same [`crate::commons::s3::S3Connection`] two times,
+    /// resulting in e.g. the s3 credentials being mounted twice as the sake volume.
+    ///
+    /// We could have made this function fallible, but decided not to do so for now, as it would be a bigger breaking
+    /// change.
+    pub fn add_volume(&mut self, volume: Volume) -> &mut Self {
+        if let Some(existing_volume) = self.volumes.get(&volume.name) {
+            if !existing_volume.eq(&volume) {
+                tracing::error!(
+                    ?existing_volume,
+                    new_volume = ?volume,
+                    "The operator tried to add a volume to Pod, which was already added with a different content! \
+                        The new volume will be ignored."
+                );
+            }
+        } else {
+            self.volumes.insert(volume.name.clone(), volume);
+        }
+
+        self
+    }
+
+    /// See [`Self::add_volume`] for details
     pub fn add_volumes(&mut self, volumes: Vec<Volume>) -> &mut Self {
-        self.volumes.get_or_insert_with(Vec::new).extend(volumes);
+        for volume in volumes {
+            self.add_volume(volume);
+        }
+
         self
     }
 
@@ -533,7 +566,11 @@ impl PodBuilder {
             }),
             security_context: self.security_context.clone(),
             tolerations: self.tolerations.clone(),
-            volumes: self.volumes.clone(),
+            volumes: if self.volumes.is_empty() {
+                None
+            } else {
+                Some(self.volumes.clone().into_values().collect())
+            },
             // Legacy feature for ancient Docker images
             // In practice, this just causes a bunch of unused environment variables that may conflict with other uses,
             // such as https://github.com/stackabletech/spark-operator/pull/256.

crates/stackable-operator/src/builder/pod/mod.rs

@@ -79,16 +79,12 @@ impl ResolvedS3Connection {
     ///
     /// * Credentials needed to connect to S3
     /// * Needed TLS volumes
-    ///
-    /// `unique_identifier` needs to be a unique identifier (e.g. in case of trino-operator the name of the catalog),
-    /// so that multiple mounts of the same SecretClass do not produce clashing volumes and volumeMounts.
     pub fn add_volumes_and_mounts(
         &self,
-        unique_identifier: &str,
         pod_builder: &mut PodBuilder,
         container_builders: Vec<&mut ContainerBuilder>,
     ) -> Result<(), S3Error> {
-        let (volumes, mounts) = self.volumes_and_mounts(unique_identifier)?;
+        let (volumes, mounts) = self.volumes_and_mounts()?;
         pod_builder.add_volumes(volumes);
         for cb in container_builders {
             cb.add_volume_mounts(mounts.clone());
@@ -99,28 +95,22 @@ impl ResolvedS3Connection {
 
     /// It is recommended to use [`Self::add_volumes_and_mounts`], this function returns you the
     /// volumes and mounts in case you need to add them by yourself.
-    pub fn volumes_and_mounts(
-        &self,
-        unique_identifier: &str,
-    ) -> Result<(Vec<Volume>, Vec<VolumeMount>), S3Error> {
+    pub fn volumes_and_mounts(&self) -> Result<(Vec<Volume>, Vec<VolumeMount>), S3Error> {
         let mut volumes = Vec::new();
         let mut mounts = Vec::new();
 
         if let Some(credentials) = &self.credentials {
             let secret_class = &credentials.secret_class;
-            let volume_name = format!("{unique_identifier}-{secret_class}-s3-credentials");
+            let volume_name = format!("{secret_class}-s3-credentials");
 
             volumes.push(
                 credentials
                     .to_volume(&volume_name)
                     .context(AddS3CredentialVolumesSnafu)?,
             );
             mounts.push(
-                VolumeMountBuilder::new(
-                    volume_name,
-                    format!("{SECRET_BASE_PATH}/{unique_identifier}-{secret_class}"),
-                )
-                .build(),
+                VolumeMountBuilder::new(volume_name, format!("{SECRET_BASE_PATH}/{secret_class}"))
+                    .build(),
             );
         }
 
@@ -137,15 +127,12 @@ impl ResolvedS3Connection {
 
     /// Returns the path of the files containing bind user and password.
     /// This will be None if there are no credentials for this LDAP connection.
-    ///
-    /// `unique_identifier` needs to be a unique identifier (e.g. in case of trino-operator the name of the catalog),
-    /// so that multiple mounts of the same SecretClass do not produce clashing volumes and volumeMounts.
-    pub fn credentials_mount_paths(&self, unique_identifier: &str) -> Option<(String, String)> {
+    pub fn credentials_mount_paths(&self) -> Option<(String, String)> {
         self.credentials.as_ref().map(|bind_credentials| {
             let secret_class = &bind_credentials.secret_class;
             (
-                format!("{SECRET_BASE_PATH}/{unique_identifier}-{secret_class}/accessKey"),
-                format!("{SECRET_BASE_PATH}/{unique_identifier}-{secret_class}/secretKey"),
+                format!("{SECRET_BASE_PATH}/{secret_class}/accessKey"),
+                format!("{SECRET_BASE_PATH}/{secret_class}/secretKey"),
             )
         })
     }
@@ -214,7 +201,7 @@ mod test {
             credentials: None,
             tls: TlsClientDetails { tls: None },
         };
-        let (volumes, mounts) = s3.volumes_and_mounts("lakehouse").unwrap();
+        let (volumes, mounts) = s3.volumes_and_mounts().unwrap();
 
         assert_eq!(s3.endpoint().unwrap(), Url::parse("http://minio").unwrap());
         assert_eq!(volumes, vec![]);
@@ -239,7 +226,7 @@ mod test {
                 }),
             },
         };
-        let (mut volumes, mut mounts) = s3.volumes_and_mounts("lakehouse").unwrap();
+        let (mut volumes, mut mounts) = s3.volumes_and_mounts().unwrap();
 
         assert_eq!(
             s3.endpoint().unwrap(),
@@ -250,10 +237,7 @@ mod test {
         assert_eq!(mounts.len(), 1);
         let mount = mounts.remove(0);
 
-        assert_eq!(
-            &volume.name,
-            "lakehouse-ionos-s3-credentials-s3-credentials"
-        );
+        assert_eq!(&volume.name, "ionos-s3-credentials-s3-credentials");
         assert_eq!(
             &volume
                 .ephemeral
@@ -271,15 +255,12 @@ mod test {
         );
 
         assert_eq!(mount.name, volume.name);
+        assert_eq!(mount.mount_path, "/stackable/secrets/ionos-s3-credentials");
         assert_eq!(
-            mount.mount_path,
-            "/stackable/secrets/lakehouse-ionos-s3-credentials"
-        );
-        assert_eq!(
-            s3.credentials_mount_paths("lakehouse"),
+            s3.credentials_mount_paths(),
             Some((
-                "/stackable/secrets/lakehouse-ionos-s3-credentials/accessKey".to_string(),
-                "/stackable/secrets/lakehouse-ionos-s3-credentials/secretKey".to_string()
+                "/stackable/secrets/ionos-s3-credentials/accessKey".to_string(),
+                "/stackable/secrets/ionos-s3-credentials/secretKey".to_string()
             ))
         );
     }
@@ -297,7 +278,7 @@ mod test {
                 }),
             },
         };
-        let (volumes, mounts) = s3.volumes_and_mounts("lakehouse").unwrap();
+        let (volumes, mounts) = s3.volumes_and_mounts().unwrap();
 
         assert_eq!(
             s3.endpoint().unwrap(),