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

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

* add pr number to changelog

* chore: Use borrows

* chore: Remove Arc

---------

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

Author: labrenbe

CHANGELOG.md

@@ -10,12 +10,16 @@ All notable changes to this project will be documented in this file.
   - BREAKING: The file log directory was set by `TRINO_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 ([#734]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#733]).
+- Fix a bug where changes to ConfigMaps that are referenced in the TrinoCluster spec didn't trigger a reconciliation ([#734]).
 
 [#728]: https://github.com/stackabletech/trino-operator/pull/728
+[#734]: https://github.com/stackabletech/trino-operator/pull/734
 [#733]: https://github.com/stackabletech/trino-operator/pull/733
 
 ## [25.3.0] - 2025-03-21

Cargo.lock

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

@@ -93,7 +93,7 @@ use crate::{
     operations::{
         add_graceful_shutdown_config, graceful_shutdown_config_properties, pdb::add_pdbs,
     },
-    product_logging::{get_log_properties, get_vector_toml, resolve_vector_aggregator_address},
+    product_logging::{get_log_properties, get_vector_toml},
 };
 
 pub struct Ctx {
@@ -230,10 +230,8 @@ pub enum Error {
     #[snafu(display("failed to resolve and merge config for role and role group"))]
     FailedToResolveConfig { source: crate::crd::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 build vector container"))]
     BuildVectorContainer { source: LoggingError },
@@ -470,10 +468,6 @@ pub async fn reconcile_trino(
 
     create_shared_internal_secret(trino, client).await?;
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(trino, client)
-        .await
-        .context(ResolveVectorAggregatorAddressSnafu)?;
-
     let mut sts_cond_builder = StatefulSetConditionBuilder::default();
 
     for (trino_role_str, role_config) in validated_config {
@@ -498,7 +492,6 @@ pub async fn reconcile_trino(
                 &merged_config,
                 &trino_authentication_config,
                 &trino_opa_config,
-                vector_aggregator_address.as_deref(),
                 &client.kubernetes_cluster_info,
             )?;
             let rg_catalog_configmap = build_rolegroup_catalog_config_map(
@@ -634,7 +627,6 @@ fn build_rolegroup_config_map(
     merged_config: &v1alpha1::TrinoConfig,
     trino_authentication_config: &TrinoAuthenticationConfig,
     trino_opa_config: &Option<TrinoOpaConfig>,
-    vector_aggregator_address: Option<&str>,
     cluster_info: &KubernetesClusterInfo,
 ) -> Result<ConfigMap> {
     let mut cm_conf_data = BTreeMap::new();
@@ -725,14 +717,11 @@ fn build_rolegroup_config_map(
                     cm_conf_data.insert(file_name.to_string(), log_properties);
                 }
 
-                if let Some(vector_toml) = get_vector_toml(
-                    rolegroup_ref,
-                    vector_aggregator_address,
-                    &merged_config.logging,
-                )
-                .context(InvalidLoggingConfigSnafu {
-                    cm_name: rolegroup_ref.object_name(),
-                })? {
+                if let Some(vector_toml) = get_vector_toml(rolegroup_ref, &merged_config.logging)
+                    .context(InvalidLoggingConfigSnafu {
+                        cm_name: rolegroup_ref.object_name(),
+                    })?
+                {
                     cm_conf_data.insert(
                         product_logging::framework::VECTOR_CONFIG_FILE.to_string(),
                         vector_toml,
@@ -1079,21 +1068,29 @@ fn build_rolegroup_statefulset(
     }
 
     if merged_config.logging.enable_vector_agent {
-        pod_builder.add_container(
-            product_logging::framework::vector_container(
-                resolved_product_image,
-                "config",
-                "log",
-                merged_config.logging.containers.get(&Container::Vector),
-                ResourceRequirementsBuilder::new()
-                    .with_cpu_request("250m")
-                    .with_cpu_limit("500m")
-                    .with_memory_request("128Mi")
-                    .with_memory_limit("128Mi")
-                    .build(),
-            )
-            .context(BuildVectorContainerSnafu)?,
-        );
+        match &trino.spec.cluster_config.vector_aggregator_config_map_name {
+            Some(vector_aggregator_config_map_name) => {
+                pod_builder.add_container(
+                    product_logging::framework::vector_container(
+                        resolved_product_image,
+                        "config",
+                        "log",
+                        merged_config.logging.containers.get(&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(BuildVectorContainerSnafu)?,
+                );
+            }
+            None => {
+                VectorAggregatorConfigMapMissingSnafu.fail()?;
+            }
+        }
     }
 
     let metadata = ObjectMetaBuilder::new()
@@ -1812,7 +1809,6 @@ mod tests {
             &merged_config,
             &trino_authentication_config,
             &trino_opa_config,
-            None,
             &cluster_info,
         )
         .unwrap()

Cargo.toml

@@ -137,8 +137,9 @@ async fn main() -> anyhow::Result<()> {
                 watch_namespace.get_api::<DeserializeGuard<v1alpha1::TrinoCluster>>(&client),
                 watcher::Config::default(),
             );
-            let catalog_cluster_store = Arc::new(cluster_controller.store());
-            let authentication_class_cluster_store = catalog_cluster_store.clone();
+            let catalog_cluster_store = cluster_controller.store();
+            let authentication_class_cluster_store = cluster_controller.store();
+            let config_map_cluster_store = cluster_controller.store();
 
             cluster_controller
                 .owns(
@@ -181,6 +182,17 @@ async fn main() -> anyhow::Result<()> {
                             .map(|trino| ObjectRef::from_obj(&*trino))
                     },
                 )
+                .watches(
+                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                    watcher::Config::default(),
+                    move |config_map| {
+                        config_map_cluster_store
+                            .state()
+                            .into_iter()
+                            .filter(move |druid| references_config_map(druid, &config_map))
+                            .map(|druid| ObjectRef::from_obj(&*druid))
+                    },
+                )
                 .run(
                     controller::reconcile_trino,
                     controller::error_policy,
@@ -229,3 +241,20 @@ fn references_authentication_class(
         .iter()
         .any(|c| c.authentication_class_name() == &authentication_class_name)
 }
+
+fn references_config_map(
+    trino: &DeserializeGuard<v1alpha1::TrinoCluster>,
+    config_map: &DeserializeGuard<ConfigMap>,
+) -> bool {
+    let Ok(trino) = &trino.0 else {
+        return false;
+    };
+
+    match &trino.spec.cluster_config.authorization {
+        Some(trino_authorization) => match &trino_authorization.opa {
+            Some(opa_config) => opa_config.config_map_name == config_map.name_any(),
+            None => false,
+        },
+        None => false,
+    }
+}

crate-hashes.json

@@ -1,8 +1,5 @@
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::Snafu;
 use stackable_operator::{
-    client::Client,
-    k8s_openapi::api::core::v1::ConfigMap,
-    kube::ResourceExt,
     product_logging::{
         framework::create_vector_config,
         spec::{
@@ -32,15 +29,10 @@ 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>;
 
-const VECTOR_AGGREGATOR_CM_ENTRY: &str = "ADDRESS";
-
 #[derive(Display)]
 #[strum(serialize_all = "lowercase")]
 pub enum TrinoLogLevel {
@@ -63,44 +55,6 @@ impl From<LogLevel> for TrinoLogLevel {
     }
 }
 
-/// 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(
-    trino: &v1alpha1::TrinoCluster,
-    client: &Client,
-) -> Result<Option<String>> {
-    let vector_aggregator_address = if let Some(vector_aggregator_config_map_name) = &trino
-        .spec
-        .cluster_config
-        .vector_aggregator_config_map_name
-        .as_ref()
-    {
-        let vector_aggregator_address = client
-            .get::<ConfigMap>(
-                vector_aggregator_config_map_name,
-                trino
-                    .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)
-}
-
 /// Return the `log.properties` configuration
 pub fn get_log_properties(logging: &Logging<Container>) -> Option<String> {
     if let Some(ContainerLogConfig {
@@ -116,7 +70,6 @@ pub fn get_log_properties(logging: &Logging<Container>) -> Option<String> {
 /// Return the vector toml configuration
 pub fn get_vector_toml(
     rolegroup: &RoleGroupRef<v1alpha1::TrinoCluster>,
-    vector_aggregator_address: Option<&str>,
     logging: &Logging<Container>,
 ) -> Result<Option<String>> {
     let vector_log_config = if let Some(ContainerLogConfig {
@@ -129,11 +82,7 @@ pub fn get_vector_toml(
     };
 
     if logging.enable_vector_agent {
-        Ok(Some(create_vector_config(
-            rolegroup,
-            vector_aggregator_address.context(MissingVectorAggregatorAddressSnafu)?,
-            vector_log_config,
-        )))
+        Ok(Some(create_vector_config(rolegroup, vector_log_config)))
     } else {
         Ok(None)
     }