feat: Inject vector aggregator address as env into vector config (#551)

* inject vector aggregator address as env into config

* chore: Use borrows

* chore: Add todo comments on unwraps

---------

Co-authored-by: Nick Larsen <[email protected]>

Author: labrenbe

CHANGELOG.md

@@ -10,12 +10,15 @@ All notable changes to this project will be documented in this file.
   - BREAKING: The file log directory was set by `SPARK_K8S_OPERATOR_LOG_DIRECTORY`, and is now set by `ROLLING_LOGS`
     (or via `--rolling-logs <DIRECTORY>`).
   - Replace stackable-operator `print_startup_string` with `tracing::info!` with fields.
+- BREAKING: Inject the vector aggregator address into the vector config using the env var `VECTOR_AGGREGATOR_ADDRESS` instead
+    of having the operator write it to the vector config ([#551]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#553]).
 
 [#547]: https://github.com/stackabletech/spark-k8s-operator/pull/547
+[#551]: https://github.com/stackabletech/spark-k8s-operator/pull/551
 [#553]: https://github.com/stackabletech/spark-k8s-operator/pull/553
 
 ## [25.3.0] - 2025-03-21

Cargo.lock

@@ -11,7 +11,7 @@ repository = "https://github.com/stackabletech/spark-k8s-operator"
 
 [workspace.dependencies]
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.89.1" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.90.0" }
 stackable-telemetry = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-telemetry-0.4.0" }
 stackable-versioned = { git = "https://github.com/stackabletech/operator-rs.git", features = ["k8s"], tag = "stackable-versioned-0.7.1" }
 

Cargo.nix

@@ -155,6 +155,7 @@ impl S3LogDir {
         let bucket = log_file_dir
             .bucket
             .clone()
+            // TODO (@NickLarsenNZ): Explain this unwrap. Either convert to expect, or gracefully handle the error.
             .resolve(client, namespace.unwrap().as_str())
             .await
             .context(ConfigureS3Snafu)?;

Cargo.toml

@@ -838,6 +838,7 @@ impl v1alpha1::SparkApplication {
         product_config: &ProductConfigManager,
     ) -> Result<ValidatedRoleConfigByPropertyKind, Error> {
         let submit_conf = if self.spec.job.is_some() {
+            // TODO (@NickLarsenNZ): Explain this unwrap. Either convert to expect, or gracefully handle the error.
             self.spec.job.as_ref().unwrap().clone()
         } else {
             CommonConfiguration {
@@ -847,6 +848,7 @@ impl v1alpha1::SparkApplication {
         };
 
         let driver_conf = if self.spec.driver.is_some() {
+            // TODO (@NickLarsenNZ): Explain this unwrap. Either convert to expect, or gracefully handle the error.
             self.spec.driver.as_ref().unwrap().clone()
         } else {
             CommonConfiguration {
@@ -857,6 +859,7 @@ impl v1alpha1::SparkApplication {
 
         let executor_conf: RoleGroup<RoleConfigFragment, JavaCommonConfig> =
             if self.spec.executor.is_some() {
+                // TODO (@NickLarsenNZ): Explain this unwrap. Either convert to expect, or gracefully handle the error.
                 self.spec.executor.as_ref().unwrap().clone()
             } else {
                 RoleGroup {

crate-hashes.json

@@ -63,7 +63,7 @@ use crate::{
         tlscerts, to_spark_env_sh_string,
     },
     history::operations::pdb::add_pdbs,
-    product_logging::{self, resolve_vector_aggregator_address},
+    product_logging::{self},
 };
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -142,8 +142,8 @@ pub enum Error {
         source: stackable_operator::cluster_resources::Error,
     },
 
-    #[snafu(display("failed to resolve the Vector aggregator address"))]
-    ResolveVectorAggregatorAddress { source: product_logging::Error },
+    #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))]
+    VectorAggregatorConfigMapMissing,
 
     #[snafu(display("failed to add the logging configuration to the ConfigMap [{cm_name}]"))]
     InvalidLoggingConfig {
@@ -250,16 +250,6 @@ pub async fn reconcile(
     .await
     .context(LogDirSnafu)?;
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(
-        client,
-        shs.namespace()
-            .as_deref()
-            .context(ObjectHasNoNamespaceSnafu)?,
-        shs.spec.vector_aggregator_config_map_name.as_deref(),
-    )
-    .await
-    .context(ResolveVectorAggregatorAddressSnafu)?;
-
     // Use a dedicated service account for history server pods.
     let (serviceaccount, rolebinding) =
         build_history_role_serviceaccount(shs, &resolved_product_image.app_version_label)?;
@@ -318,7 +308,6 @@ pub async fn reconcile(
                 &resolved_product_image.app_version_label,
                 &rgr,
                 &log_dir,
-                vector_aggregator_address.as_deref(),
             )?;
             cluster_resources
                 .add(client, config_map)
@@ -377,7 +366,6 @@ fn build_config_map(
     app_version_label: &str,
     rolegroupref: &RoleGroupRef<v1alpha1::SparkHistoryServer>,
     log_dir: &ResolvedLogDir,
-    vector_aggregator_address: Option<&str>,
 ) -> Result<ConfigMap, Error> {
     let cm_name = rolegroupref.object_name();
 
@@ -428,7 +416,6 @@ fn build_config_map(
 
     product_logging::extend_config_map(
         rolegroupref,
-        vector_aggregator_address,
         &merged_config.logging,
         SparkHistoryServerContainer::SparkHistory,
         SparkHistoryServerContainer::Vector,
@@ -556,24 +543,32 @@ fn build_stateful_set(
     pb.add_container(container);
 
     if merged_config.logging.enable_vector_agent {
-        pb.add_container(
-            vector_container(
-                resolved_product_image,
-                VOLUME_MOUNT_NAME_CONFIG,
-                VOLUME_MOUNT_NAME_LOG,
-                merged_config
-                    .logging
-                    .containers
-                    .get(&SparkHistoryServerContainer::Vector),
-                ResourceRequirementsBuilder::new()
-                    .with_cpu_request("250m")
-                    .with_cpu_limit("500m")
-                    .with_memory_request("128Mi")
-                    .with_memory_limit("128Mi")
-                    .build(),
-            )
-            .context(ConfigureLoggingSnafu)?,
-        );
+        match shs.spec.vector_aggregator_config_map_name.to_owned() {
+            Some(vector_aggregator_config_map_name) => {
+                pb.add_container(
+                    vector_container(
+                        resolved_product_image,
+                        VOLUME_MOUNT_NAME_CONFIG,
+                        VOLUME_MOUNT_NAME_LOG,
+                        merged_config
+                            .logging
+                            .containers
+                            .get(&SparkHistoryServerContainer::Vector),
+                        ResourceRequirementsBuilder::new()
+                            .with_cpu_request("250m")
+                            .with_cpu_limit("500m")
+                            .with_memory_request("128Mi")
+                            .with_memory_limit("128Mi")
+                            .build(),
+                        &vector_aggregator_config_map_name,
+                    )
+                    .context(ConfigureLoggingSnafu)?,
+                );
+            }
+            None => {
+                VectorAggregatorConfigMapMissingSnafu.fail()?;
+            }
+        }
     }
 
     let mut pod_template = pb.build_template();
@@ -634,6 +629,7 @@ fn build_service(
             Some("None".to_string()),
         ),
         None => (
+            // TODO (@NickLarsenNZ): Explain this unwrap. Either convert to expect, or gracefully handle the error.
             format!("{}-{}", shs.metadata.name.as_ref().unwrap(), role),
             shs.spec.cluster_config.listener_class.k8s_service_type(),
             None,

rust/operator-binary/src/crd/logdir.rs

@@ -1,10 +1,8 @@
 use std::fmt::Display;
 
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::Snafu;
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
-    client::Client,
-    k8s_openapi::api::core::v1::ConfigMap,
     kube::Resource,
     memory::BinaryMultiple,
     product_logging::{
@@ -29,51 +27,17 @@ pub enum Error {
         entry: &'static str,
         cm_name: String,
     },
-
-    #[snafu(display("vectorAggregatorConfigMapName must be set"))]
-    MissingVectorAggregatorAddress,
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
 pub const LOG_FILE: &str = "spark.log4j2.xml";
 
-const VECTOR_AGGREGATOR_CM_ENTRY: &str = "ADDRESS";
 const CONSOLE_CONVERSION_PATTERN: &str = "%d{ISO8601} %p [%t] %c - %m%n";
 
-/// Return the address of the Vector aggregator if the corresponding ConfigMap name is given in the
-/// cluster spec
-pub async fn resolve_vector_aggregator_address(
-    client: &Client,
-    namespace: &str,
-    vector_aggregator_config_map_name: Option<&str>,
-) -> Result<Option<String>> {
-    let vector_aggregator_address =
-        if let Some(vector_aggregator_config_map_name) = vector_aggregator_config_map_name {
-            let vector_aggregator_address = client
-                .get::<ConfigMap>(vector_aggregator_config_map_name, namespace)
-                .await
-                .context(ConfigMapNotFoundSnafu {
-                    cm_name: vector_aggregator_config_map_name.to_string(),
-                })?
-                .data
-                .and_then(|mut data| data.remove(VECTOR_AGGREGATOR_CM_ENTRY))
-                .context(MissingConfigMapEntrySnafu {
-                    entry: VECTOR_AGGREGATOR_CM_ENTRY,
-                    cm_name: vector_aggregator_config_map_name.to_string(),
-                })?;
-            Some(vector_aggregator_address)
-        } else {
-            None
-        };
-
-    Ok(vector_aggregator_address)
-}
-
 /// Extend a ConfigMap with logging and Vector configurations
 pub fn extend_config_map<C, K>(
     role_group: &RoleGroupRef<K>,
-    vector_aggregator_address: Option<&str>,
     logging: &Logging<C>,
     main_container: C,
     vector_container: C,
@@ -114,11 +78,7 @@ where
     if logging.enable_vector_agent {
         cm_builder.add_data(
             product_logging::framework::VECTOR_CONFIG_FILE,
-            product_logging::framework::create_vector_config(
-                role_group,
-                vector_aggregator_address.context(MissingVectorAggregatorAddressSnafu)?,
-                vector_log_config,
-            ),
+            product_logging::framework::create_vector_config(role_group, vector_log_config),
         );
     }