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

* watch referenced configmaps and inject vector aggregator discovery cm as env

* wip: use local stackable-operator version

* add changelog entry

* bump operator-rs

* remove local patch for operator-rs

* chore: Update nix files

* chore: Generated commit to update templated files since the last template run up to stackabletech/operator-templating@8ad0568 (#646)

Reference-to: stackabletech/operator-templating@8ad0568 (Update nixpkgs)

* fix: Use `json` file extension for log files (#647)

* fix: Use `json` file extension for log files

* chore: Update changelog

* fix formatting in changelog

* fix typo in changelog

* chore: Rename store variable

* chore: Use borrows

* chore: Formatting

* chore: Formatting

* noop

---------

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

Author: 3 people

CHANGELOG.md

@@ -8,12 +8,20 @@
   - BREAKING: The file log directory was set by `HBASE_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 ([#645]).
+
+### Fixed
+
+- Use `json` file extension for log files ([#647]).
+- Fix a bug where changes to ConfigMaps that are referenced in the HbaseCluster spec didn't trigger a reconciliation ([#645]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#647]).
 
 [#640]: https://github.com/stackabletech/hbase-operator/pull/640
+[#645]: https://github.com/stackabletech/hbase-operator/pull/645
 [#647]: https://github.com/stackabletech/hbase-operator/pull/647
 
 ## [25.3.0] - 2025-03-21

Cargo.lock

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

@@ -82,7 +82,6 @@ use crate::{
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::{
         CONTAINERDEBUG_LOG_DIRECTORY, STACKABLE_LOG_DIR, extend_role_group_config_map,
-        resolve_vector_aggregator_address,
     },
     security::{self, opa::HbaseOpaConfig},
     zookeeper::{self, ZookeeperConnectionInformation},
@@ -235,10 +234,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 add the logging configuration to the ConfigMap [{cm_name}]"))]
     InvalidLoggingConfig {
@@ -350,10 +347,6 @@ pub async fn reconcile_hbase(
         .await
         .context(RetrieveZookeeperConnectionInformationSnafu)?;
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(hbase, client)
-        .await
-        .context(ResolveVectorAggregatorAddressSnafu)?;
-
     let roles = hbase.build_role_properties().context(RolePropertiesSnafu)?;
 
     let validated_config = validate_all_roles_and_groups_config(
@@ -448,7 +441,6 @@ pub async fn reconcile_hbase(
                 &merged_config,
                 &resolved_product_image,
                 hbase_opa_config.as_ref(),
-                vector_aggregator_address.as_deref(),
             )?;
             let rg_statefulset = build_rolegroup_statefulset(
                 hbase,
@@ -576,7 +568,6 @@ fn build_rolegroup_config_map(
     merged_config: &AnyServiceConfig,
     resolved_product_image: &ResolvedProductImage,
     hbase_opa_config: Option<&HbaseOpaConfig>,
-    vector_aggregator_address: Option<&str>,
 ) -> Result<ConfigMap, Error> {
     let mut hbase_site_xml = String::new();
     let mut hbase_env_sh = String::new();
@@ -703,7 +694,6 @@ fn build_rolegroup_config_map(
 
     extend_role_group_config_map(
         rolegroup,
-        vector_aggregator_address,
         merged_config.logging(),
         &mut builder,
         &resolved_product_image.product_version,
@@ -1003,21 +993,28 @@ fn build_rolegroup_statefulset(
 
     // Vector sidecar shall be the last container in the list
     if merged_config.logging().enable_vector_agent {
-        pod_builder.add_container(
-            product_logging::framework::vector_container(
-                resolved_product_image,
-                "hbase-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(ConfigureLoggingSnafu)?,
-        );
+        if let Some(vector_aggregator_config_map_name) =
+            &hbase.spec.cluster_config.vector_aggregator_config_map_name
+        {
+            pod_builder.add_container(
+                product_logging::framework::vector_container(
+                    resolved_product_image,
+                    "hbase-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(ConfigureLoggingSnafu)?,
+            );
+        } else {
+            VectorAggregatorConfigMapMissingSnafu.fail()?;
+        }
     }
 
     let mut pod_template = pod_builder.build_template();

Cargo.toml

@@ -6,12 +6,17 @@ use hbase_controller::FULL_HBASE_CONTROLLER_NAME;
 use stackable_operator::{
     YamlSchema,
     cli::{Command, ProductOperatorRun, RollingPeriod},
-    k8s_openapi::api::{apps::v1::StatefulSet, core::v1::Service},
+    k8s_openapi::api::{
+        apps::v1::StatefulSet,
+        core::v1::{ConfigMap, Service},
+    },
     kube::{
+        ResourceExt,
         core::DeserializeGuard,
         runtime::{
             Controller,
             events::{Recorder, Reporter},
+            reflector::ObjectRef,
             watcher,
         },
     },
@@ -128,46 +133,77 @@ async fn main() -> anyhow::Result<()> {
                 instance: None,
             }));
 
-            Controller::new(
+            let hbase_controller = Controller::new(
                 watch_namespace.get_api::<DeserializeGuard<v1alpha1::HbaseCluster>>(&client),
                 watcher::Config::default(),
-            )
-            .owns(
-                watch_namespace.get_api::<Service>(&client),
-                watcher::Config::default(),
-            )
-            .owns(
-                watch_namespace.get_api::<StatefulSet>(&client),
-                watcher::Config::default(),
-            )
-            .shutdown_on_signal()
-            .run(
-                hbase_controller::reconcile_hbase,
-                hbase_controller::error_policy,
-                Arc::new(hbase_controller::Ctx {
-                    client: client.clone(),
-                    product_config,
-                }),
-            )
-            .for_each_concurrent(
-                16, // concurrency limit
-                |result| {
-                    // The event_recorder needs to be shared across all invocations, so that
-                    // events are correctly aggregated
-                    let event_recorder = event_recorder.clone();
-                    async move {
-                        report_controller_reconciled(
-                            &event_recorder,
-                            FULL_HBASE_CONTROLLER_NAME,
-                            &result,
-                        )
-                        .await;
-                    }
-                },
-            )
-            .await;
+            );
+            let config_map_store = hbase_controller.store();
+            hbase_controller
+                .owns(
+                    watch_namespace.get_api::<Service>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<StatefulSet>(&client),
+                    watcher::Config::default(),
+                )
+                .shutdown_on_signal()
+                .watches(
+                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                    watcher::Config::default(),
+                    move |config_map| {
+                        config_map_store
+                            .state()
+                            .into_iter()
+                            .filter(move |hbase| references_config_map(hbase, &config_map))
+                            .map(|hbase| ObjectRef::from_obj(&*hbase))
+                    },
+                )
+                .run(
+                    hbase_controller::reconcile_hbase,
+                    hbase_controller::error_policy,
+                    Arc::new(hbase_controller::Ctx {
+                        client: client.clone(),
+                        product_config,
+                    }),
+                )
+                .for_each_concurrent(
+                    16, // concurrency limit
+                    |result| {
+                        // The event_recorder needs to be shared across all invocations, so that
+                        // events are correctly aggregated
+                        let event_recorder = event_recorder.clone();
+                        async move {
+                            report_controller_reconciled(
+                                &event_recorder,
+                                FULL_HBASE_CONTROLLER_NAME,
+                                &result,
+                            )
+                            .await;
+                        }
+                    },
+                )
+                .await;
         }
     }
 
     Ok(())
 }
+
+fn references_config_map(
+    hbase: &DeserializeGuard<v1alpha1::HbaseCluster>,
+    config_map: &DeserializeGuard<ConfigMap>,
+) -> bool {
+    let Ok(hbase) = &hbase.0 else {
+        return false;
+    };
+
+    hbase.spec.cluster_config.zookeeper_config_map_name == config_map.name_any()
+        || hbase.spec.cluster_config.hdfs_config_map_name == config_map.name_any()
+        || match &hbase.spec.cluster_config.authorization {
+            Some(hbase_authorization) => {
+                hbase_authorization.opa.config_map_name == config_map.name_any()
+            }
+            None => false,
+        }
+}