feat: Inject vector aggregator address as env into vector config

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 `NIFI_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 ([#772]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#774]).
+- Fix a bug where changes to ConfigMaps that are referenced in the NifiCluster spec didn't trigger a reconciliation ([#772]).
 
 [#767]: https://github.com/stackabletech/nifi-operator/pull/767
+[#772]: https://github.com/stackabletech/nifi-operator/pull/772
 [#774]: https://github.com/stackabletech/nifi-operator/pull/774
 
 ## [25.3.0] - 2025-03-21

Cargo.toml

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

deny.toml

@@ -9,27 +9,6 @@ targets = [
 
 [advisories]
 yanked = "deny"
-ignore = [
-    # https://rustsec.org/advisories/RUSTSEC-2023-0071
-    # "rsa" crate: Marvin Attack: potential key recovery through timing sidechannel
-    #
-    # No patch is yet available, however work is underway to migrate to a fully constant-time implementation
-    # So we need to accept this, as of SDP 24.11 we are not using the rsa crate to create certificates used in production
-    # setups.
-    #
-    # TODO: Remove after https://github.com/RustCrypto/RSA/pull/394 is merged
-    "RUSTSEC-2023-0071",
-
-    # https://rustsec.org/advisories/RUSTSEC-2024-0384
-    # "instant" is unmaintained
-    #
-    # The upstream "kube" crate also silenced this in https://github.com/kube-rs/kube/commit/4f1e889f265da8f19f03f60683569cae1a154fda
-    # They/we are actively working on migrating kube from backoff to backon, which removes the transitive dependency on
-    # instant, in https://github.com/kube-rs/kube/pull/1652.
-    #
-    # TODO: Remove after https://github.com/kube-rs/kube/pull/1652 is merged
-    "RUSTSEC-2024-0384",
-]
 
 [bans]
 multiple-versions = "allow"

rust/operator-binary/src/controller.rs

@@ -88,7 +88,7 @@ use crate::{
         v1alpha1,
     },
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
-    product_logging::{extend_role_group_config_map, resolve_vector_aggregator_address},
+    product_logging::extend_role_group_config_map,
     reporting_task::{self, build_maybe_reporting_task, build_reporting_task_service_name},
     security::{
         authentication::{
@@ -254,10 +254,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 {
@@ -515,10 +513,6 @@ pub async fn reconcile_nifi(
             .context(SecuritySnafu)?;
     }
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(nifi, client)
-        .await
-        .context(ResolveVectorAggregatorAddressSnafu)?;
-
     let (rbac_sa, rbac_rolebinding) = build_rbac_resources(
         nifi,
         APP_NAME,
@@ -572,7 +566,6 @@ pub async fn reconcile_nifi(
                 &rolegroup,
                 rolegroup_config,
                 &merged_config,
-                vector_aggregator_address.as_deref(),
                 &proxy_hosts,
             )
             .await?;
@@ -747,7 +740,6 @@ async fn build_node_rolegroup_config_map(
     rolegroup: &RoleGroupRef<v1alpha1::NifiCluster>,
     rolegroup_config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
     merged_config: &NifiConfig,
-    vector_aggregator_address: Option<&str>,
     proxy_hosts: &str,
 ) -> Result<ConfigMap> {
     tracing::debug!("building rolegroup configmaps");
@@ -833,15 +825,11 @@ async fn build_node_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()
@@ -1244,21 +1232,29 @@ async fn build_node_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(ConfigureLoggingSnafu)?,
-        );
+        match &nifi.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(ConfigureLoggingSnafu)?,
+                );
+            }
+            None => {
+                VectorAggregatorConfigMapMissingSnafu.fail()?;
+            }
+        }
     }
 
     nifi_auth_config

rust/operator-binary/src/main.rs

@@ -11,6 +11,7 @@ use stackable_operator::{
         core::v1::{ConfigMap, Service},
     },
     kube::{
+        ResourceExt,
         core::DeserializeGuard,
         runtime::{
             Controller,
@@ -139,7 +140,8 @@ async fn main() -> anyhow::Result<()> {
                 watcher::Config::default(),
             );
 
-            let nifi_store_1 = nifi_controller.store();
+            let authentication_class_store = nifi_controller.store();
+            let config_map_store = nifi_controller.store();
 
             nifi_controller
                 .owns(
@@ -159,12 +161,23 @@ async fn main() -> anyhow::Result<()> {
                     client.get_api::<DeserializeGuard<AuthenticationClass>>(&()),
                     watcher::Config::default(),
                     move |_| {
-                        nifi_store_1
+                        authentication_class_store
                             .state()
                             .into_iter()
                             .map(|nifi| ObjectRef::from_obj(&*nifi))
                     },
                 )
+                .watches(
+                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                    watcher::Config::default(),
+                    move |config_map| {
+                        config_map_store
+                            .state()
+                            .into_iter()
+                            .filter(move |nifi| references_config_map(nifi, &config_map))
+                            .map(|nifi| ObjectRef::from_obj(&*nifi))
+                    },
+                )
                 .run(
                     controller::reconcile_nifi,
                     controller::error_policy,
@@ -196,3 +209,14 @@ async fn main() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+fn references_config_map(
+    nifi: &DeserializeGuard<v1alpha1::NifiCluster>,
+    config_map: &DeserializeGuard<ConfigMap>,
+) -> bool {
+    let Ok(nifi) = &nifi.0 else {
+        return false;
+    };
+
+    nifi.spec.cluster_config.zookeeper_config_map_name == config_map.name_any()
+}

rust/operator-binary/src/product_logging.rs

@@ -1,9 +1,6 @@
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::Snafu;
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
-    client::Client,
-    k8s_openapi::api::core::v1::ConfigMap,
-    kube::ResourceExt,
     memory::BinaryMultiple,
     product_logging::{
         self,
@@ -39,7 +36,6 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 pub const LOGBACK_CONFIG_FILE: &str = "logback.xml";
 pub const NIFI_LOG_FILE: &str = "nifi.log4j.xml";
 
-const VECTOR_AGGREGATOR_CM_ENTRY: &str = "ADDRESS";
 const CONSOLE_CONVERSION_PATTERN: &str = "%date %level [%thread] %logger{40} %msg%n";
 // This is required to remove double entries in the nifi.log4j.xml as well as nested
 // console output like: "<timestamp> <loglevel> ... <timestamp> <loglevel> ...
@@ -58,47 +54,9 @@ const ADDITONAL_LOGBACK_CONFIG: &str = r#"  <appender name="PASSTHROUGH" class="
   </logger>
 "#;
 
-/// 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(
-    nifi: &v1alpha1::NifiCluster,
-    client: &Client,
-) -> Result<Option<String>> {
-    let vector_aggregator_address = if let Some(vector_aggregator_config_map_name) = &nifi
-        .spec
-        .cluster_config
-        .vector_aggregator_config_map_name
-        .as_ref()
-    {
-        let vector_aggregator_address = client
-            .get::<ConfigMap>(
-                vector_aggregator_config_map_name,
-                nifi.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::NifiCluster>,
-    vector_aggregator_address: Option<&str>,
     logging: &Logging<Container>,
     cm_builder: &mut ConfigMapBuilder,
 ) -> Result<()> {
@@ -137,11 +95,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),
         );
     }