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

* inject vector aggregator address as env into config file

* add changelog entry

* fix rustdoc

* run cargo fmt

* fix doctest

* Update crates/stackable-operator/CHANGELOG.md

Co-authored-by: Siegfried Weber <[email protected]>

* move env name into const

* Update crates/stackable-operator/src/product_logging/framework.rs

Co-authored-by: Siegfried Weber <[email protected]>

* move env name into const

---------

Co-authored-by: Siegfried Weber <[email protected]>

Author: labrenbe

Cargo.lock

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- BREAKING: Inject vector aggregator address into vector config file using an environment variable ([#1000]).
+
+[#1000]: https://github.com/stackabletech/operator-rs/pull/1000
+
 ## [0.89.1] - 2025-04-02
 
 ### Changed

crates/stackable-operator/CHANGELOG.md

@@ -31,6 +31,10 @@ const SHUTDOWN_FILE: &str = "shutdown";
 
 /// File name of the Vector config file
 pub const VECTOR_CONFIG_FILE: &str = "vector.yaml";
+/// Key in the discovery ConfigMap that holds the vector aggregator address
+const VECTOR_AGGREGATOR_CM_KEY: &str = "ADDRESS";
+/// Name of the env var in the vector container that holds the vector aggregator address
+const VECTOR_AGGREGATOR_ENV_NAME: &str = "VECTOR_AGGREGATOR_ADDRESS";
 
 #[derive(Debug, Snafu)]
 pub enum LoggingError {
@@ -678,7 +682,6 @@ pub fn create_logback_config(
 /// # }
 /// #
 /// # let logging = fragment::validate::<Logging<Container>>(default_logging()).unwrap();
-/// # let vector_aggregator_address = "vector-aggregator:6000";
 /// # let role_group = RoleGroupRef {
 /// #     cluster: ObjectRef::<Pod>::new("test-cluster"),
 /// #     role: "role".into(),
@@ -702,7 +705,6 @@ pub fn create_logback_config(
 ///         product_logging::framework::VECTOR_CONFIG_FILE,
 ///         product_logging::framework::create_vector_config(
 ///             &role_group,
-///             vector_aggregator_address,
 ///             vector_log_config,
 ///         ),
 ///     );
@@ -712,7 +714,6 @@ pub fn create_logback_config(
 /// ```
 pub fn create_vector_config<T>(
     role_group: &RoleGroupRef<T>,
-    vector_aggregator_address: &str,
     config: Option<&AutomaticContainerLogConfig>,
 ) -> String
 where
@@ -1330,7 +1331,7 @@ sinks:
     inputs:
       - extended_logs
     type: vector
-    address: {vector_aggregator_address}
+    address: ${VECTOR_AGGREGATOR_ENV_NAME}
 "#,
         namespace = role_group.cluster.namespace.clone().unwrap_or_default(),
         cluster_name = role_group.cluster.name,
@@ -1386,6 +1387,7 @@ sinks:
 /// #     image_pull_policy: "Always".into(),
 /// #     pull_secrets: None,
 /// # };
+/// # let vector_aggregator_config_map_name = "vector-aggregator-discovery";
 ///
 /// let mut pod_builder = PodBuilder::new();
 /// pod_builder.metadata(ObjectMetaBuilder::default().build());
@@ -1425,6 +1427,7 @@ sinks:
 ///         "log",
 ///         logging.containers.get(&Container::Vector),
 ///         resources,
+///         vector_aggregator_config_map_name,
 ///     ).unwrap());
 /// }
 ///
@@ -1436,6 +1439,7 @@ pub fn vector_container(
     log_volume_name: &str,
     log_config: Option<&ContainerLogConfig>,
     resources: ResourceRequirements,
+    vector_aggregator_config_map_name: &str,
 ) -> Result<Container, LoggingError> {
     let log_level = if let Some(ContainerLogConfig {
         choice: Some(ContainerLogConfigChoice::Automatic(automatic_log_config)),
@@ -1473,6 +1477,11 @@ kill $vector_pid
 "
         )])
         .add_env_var("VECTOR_LOG", log_level.to_vector_literal())
+        .add_env_var_from_config_map(
+            VECTOR_AGGREGATOR_ENV_NAME,
+            vector_aggregator_config_map_name,
+            VECTOR_AGGREGATOR_CM_KEY,
+        )
         .add_volume_mount(config_volume_name, STACKABLE_CONFIG_DIR)
         .context(AddVolumeMountsSnafu)?
         .add_volume_mount(log_volume_name, STACKABLE_LOG_DIR)