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

* inject vector aggregator address as env into config & add watch for referenced cms

* add pr number to changelog

* regenerate nix

* fix changelog

* chore: Use borrows

---------

Co-authored-by: Nick Larsen <nick.larsen@stackable.tech>

Author: labrenbe

CHANGELOG.md

@@ -18,12 +18,15 @@ All notable changes to this project will be documented in this file.
   - BREAKING: user-info-fetcher: The file log directory was set by `OPA_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 ([#707]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#709]).
 
 [#703]: https://github.com/stackabletech/opa-operator/pull/703
+[#707]: https://github.com/stackabletech/opa-operator/pull/707
 [#709]: https://github.com/stackabletech/opa-operator/pull/709
 
 ## [25.3.0] - 2025-03-21

Cargo.lock

@@ -11,7 +11,7 @@ repository = "https://github.com/stackabletech/opa-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.6.0" }
 

Cargo.nix

@@ -76,9 +76,7 @@ use strum::{EnumDiscriminants, IntoStaticStr};
 use crate::{
     discovery::{self, build_discovery_configmaps},
     operations::graceful_shutdown::add_graceful_shutdown_config,
-    product_logging::{
-        BundleBuilderLogLevel, extend_role_group_config_map, resolve_vector_aggregator_address,
-    },
+    product_logging::{BundleBuilderLogLevel, extend_role_group_config_map},
 };
 
 pub const OPA_CONTROLLER_NAME: &str = "opacluster";
@@ -251,10 +249,8 @@ pub enum Error {
         source: stackable_operator::builder::pod::container::Error,
     },
 
-    #[snafu(display("failed to resolve the Vector aggregator address"))]
-    ResolveVectorAggregatorAddress {
-        source: crate::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 {
@@ -443,10 +439,6 @@ pub async fn reconcile_opa(
         .map(Cow::Borrowed)
         .unwrap_or_default();
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(opa, client)
-        .await
-        .context(ResolveVectorAggregatorAddressSnafu)?;
-
     let server_role_service = build_server_role_service(opa, &resolved_product_image)?;
     // required for discovery config map later
     let server_role_service = cluster_resources
@@ -488,7 +480,6 @@ pub async fn reconcile_opa(
             &resolved_product_image,
             &rolegroup,
             &merged_config,
-            vector_aggregator_address.as_deref(),
         )?;
         let rg_service = build_rolegroup_service(opa, &resolved_product_image, &rolegroup)?;
         let rg_daemonset = build_server_rolegroup_daemonset(
@@ -682,7 +673,6 @@ fn build_server_rolegroup_config_map(
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha1::OpaCluster>,
     merged_config: &v1alpha1::OpaConfig,
-    vector_aggregator_address: Option<&str>,
 ) -> Result<ConfigMap> {
     let mut cm_builder = ConfigMapBuilder::new();
 
@@ -711,15 +701,11 @@ fn build_server_rolegroup_config_map(
         );
     }
 
-    extend_role_group_config_map(
-        rolegroup,
-        vector_aggregator_address,
-        &merged_config.logging,
-        &mut cm_builder,
-    )
-    .context(InvalidLoggingConfigSnafu {
-        cm_name: rolegroup.object_name(),
-    })?;
+    extend_role_group_config_map(rolegroup, &merged_config.logging, &mut cm_builder).context(
+        InvalidLoggingConfigSnafu {
+            cm_name: rolegroup.object_name(),
+        },
+    )?;
 
     cm_builder
         .build()
@@ -1034,24 +1020,32 @@ fn build_server_rolegroup_daemonset(
     }
 
     if merged_config.logging.enable_vector_agent {
-        pb.add_container(
-            product_logging::framework::vector_container(
-                resolved_product_image,
-                CONFIG_VOLUME_NAME,
-                LOG_VOLUME_NAME,
-                merged_config
-                    .logging
-                    .containers
-                    .get(&v1alpha1::Container::Vector),
-                ResourceRequirementsBuilder::new()
-                    .with_cpu_request("250m")
-                    .with_cpu_limit("500m")
-                    .with_memory_request("128Mi")
-                    .with_memory_limit("128Mi")
-                    .build(),
-            )
-            .context(ConfigureLoggingSnafu)?,
-        );
+        match &opa.spec.cluster_config.vector_aggregator_config_map_name {
+            Some(vector_aggregator_config_map_name) => {
+                pb.add_container(
+                    product_logging::framework::vector_container(
+                        resolved_product_image,
+                        CONFIG_VOLUME_NAME,
+                        LOG_VOLUME_NAME,
+                        merged_config
+                            .logging
+                            .containers
+                            .get(&v1alpha1::Container::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()?;
+            }
+        }
     }
 
     add_graceful_shutdown_config(merged_config, &mut pb).context(GracefulShutdownSnafu)?;

Cargo.toml

@@ -1,10 +1,7 @@
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::Snafu;
 use stackable_opa_operator::crd::v1alpha1;
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
-    client::Client,
-    k8s_openapi::api::core::v1::ConfigMap,
-    kube::ResourceExt,
     product_logging::{
         self,
         spec::{ContainerLogConfig, ContainerLogConfigChoice, LogLevel, Logging},
@@ -32,8 +29,6 @@ pub enum Error {
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
-const VECTOR_AGGREGATOR_CM_ENTRY: &str = "ADDRESS";
-
 #[derive(strum::Display)]
 #[strum(serialize_all = "lowercase")]
 pub enum OpaLogLevel {
@@ -74,44 +69,9 @@ impl From<LogLevel> for BundleBuilderLogLevel {
     }
 }
 
-/// 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(
-    opa: &v1alpha1::OpaCluster,
-    client: &Client,
-) -> Result<Option<String>> {
-    let vector_aggregator_address = if let Some(vector_aggregator_config_map_name) =
-        &opa.spec.cluster_config.vector_aggregator_config_map_name
-    {
-        let vector_aggregator_address = client
-            .get::<ConfigMap>(
-                vector_aggregator_config_map_name,
-                opa.namespace()
-                    .as_deref()
-                    .context(ObjectHasNoNamespaceSnafu)?,
-            )
-            .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 the role group ConfigMap with logging and Vector configurations
 pub fn extend_role_group_config_map(
     rolegroup: &RoleGroupRef<v1alpha1::OpaCluster>,
-    vector_aggregator_address: Option<&str>,
     logging: &Logging<v1alpha1::Container>,
     cm_builder: &mut ConfigMapBuilder,
 ) -> Result<()> {
@@ -127,11 +87,7 @@ pub fn extend_role_group_config_map(
     if logging.enable_vector_agent {
         cm_builder.add_data(
             product_logging::framework::VECTOR_CONFIG_FILE,
-            product_logging::framework::create_vector_config(
-                rolegroup,
-                vector_aggregator_address.context(MissingVectorAggregatorAddressSnafu)?,
-                vector_log_config,
-            ),
+            product_logging::framework::create_vector_config(rolegroup, vector_log_config),
         );
     }