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

* start watching configmaps that are referenced in superset spec

* format code

* rename function

* add changelog entry

* minor refactor

* wip: Inject the vector aggregator address into the vector config using an env var

* regenerate nix

* rename stores

* chore: Rename store variables

* chore: Use borrows

---------

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

Author: labrenbe

CHANGELOG.md

@@ -8,11 +8,15 @@
   - BREAKING: The file log directory was set by `SUPERSET_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 ([#609]).
 
 ### Fixed
 
 - Use `json` file extension for log files ([#615]).
+- Fix a bug where changes to ConfigMaps that are referenced in the SupersetCluster spec didn't trigger a reconciliation ([#609]).
 
+[#609]: https://github.com/stackabletech/superset-operator/pull/609
 [#610]: https://github.com/stackabletech/superset-operator/pull/610
 [#615]: https://github.com/stackabletech/superset-operator/pull/615
 

Cargo.lock

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

@@ -150,7 +150,8 @@ async fn main() -> anyhow::Result<()> {
                 watch_namespace.get_api::<DeserializeGuard<v1alpha1::SupersetCluster>>(&client),
                 watcher::Config::default(),
             );
-            let superset_store_1 = superset_controller.store();
+            let authentication_class_store = superset_controller.store();
+            let config_map_store = superset_controller.store();
             let superset_controller = superset_controller
                 .owns(
                     watch_namespace.get_api::<DeserializeGuard<Service>>(&client),
@@ -165,7 +166,7 @@ async fn main() -> anyhow::Result<()> {
                     client.get_api::<DeserializeGuard<AuthenticationClass>>(&()),
                     watcher::Config::default(),
                     move |authentication_class| {
-                        superset_store_1
+                        authentication_class_store
                             .state()
                             .into_iter()
                             .filter(move |superset| {
@@ -174,6 +175,17 @@ async fn main() -> anyhow::Result<()> {
                             .map(|superset| ObjectRef::from_obj(&*superset))
                     },
                 )
+                .watches(
+                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                    watcher::Config::default(),
+                    move |config_map| {
+                        config_map_store
+                            .state()
+                            .into_iter()
+                            .filter(move |superset| references_config_map(superset, &config_map))
+                            .map(|superset| ObjectRef::from_obj(&*superset))
+                    },
+                )
                 .run(
                     superset_controller::reconcile_superset,
                     superset_controller::error_policy,
@@ -212,16 +224,16 @@ async fn main() -> anyhow::Result<()> {
                     ),
                 watcher::Config::default(),
             );
-            let druid_connection_store_1 = druid_connection_controller.store();
-            let druid_connection_store_2 = druid_connection_controller.store();
-            let druid_connection_store_3 = druid_connection_controller.store();
+            let superset_cluster_store = druid_connection_controller.store();
+            let job_store = druid_connection_controller.store();
+            let config_map_store = druid_connection_controller.store();
             let druid_connection_controller = druid_connection_controller
                 .shutdown_on_signal()
                 .watches(
                     watch_namespace.get_api::<DeserializeGuard<v1alpha1::SupersetCluster>>(&client),
                     watcher::Config::default(),
                     move |superset_cluster| {
-                        druid_connection_store_1
+                        superset_cluster_store
                             .state()
                             .into_iter()
                             .filter(move |druid_connection| {
@@ -234,7 +246,7 @@ async fn main() -> anyhow::Result<()> {
                     watch_namespace.get_api::<DeserializeGuard<Job>>(&client),
                     watcher::Config::default(),
                     move |job| {
-                        druid_connection_store_2
+                        job_store
                             .state()
                             .into_iter()
                             .filter(move |druid_connection| valid_druid_job(druid_connection, &job))
@@ -245,7 +257,7 @@ async fn main() -> anyhow::Result<()> {
                     watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
                     watcher::Config::default(),
                     move |config_map| {
-                        druid_connection_store_3
+                        config_map_store
                             .state()
                             .into_iter()
                             .filter(move |druid_connection| {
@@ -305,6 +317,26 @@ fn references_authentication_class(
         .any(|c| c.common.authentication_class_name() == &authentication_class_name)
 }
 
+fn references_config_map(
+    superset: &DeserializeGuard<v1alpha1::SupersetCluster>,
+    config_map: &DeserializeGuard<ConfigMap>,
+) -> bool {
+    let Ok(superset) = &superset.0 else {
+        return false;
+    };
+
+    match &superset.spec.cluster_config.authorization {
+        Some(superset_authorization) => {
+            superset_authorization
+                .role_mapping_from_opa
+                .opa
+                .config_map_name
+                == config_map.name_any()
+        }
+        None => false,
+    }
+}
+
 fn valid_druid_connection(
     superset_cluster: &DeserializeGuard<v1alpha1::SupersetCluster>,
     druid_connection: &DeserializeGuard<druidconnection::v1alpha1::DruidConnection>,

Cargo.toml

@@ -1,10 +1,8 @@
 use std::fmt::{Display, Write};
 
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::Snafu;
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
-    client::Client,
-    k8s_openapi::api::core::v1::ConfigMap,
     kube::Resource,
     product_logging::{
         self,
@@ -31,54 +29,17 @@ 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>;
 
 pub const LOG_CONFIG_FILE: &str = "log_config.py";
 
-const VECTOR_AGGREGATOR_CM_ENTRY: &str = "ADDRESS";
 const LOG_FILE: &str = "superset.py.json";
 
-/// 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<T: Resource>(
-    client: &Client,
-    cluster: &T,
-    vector_aggregator_config_map_name: Option<&str>,
-) -> Result<Option<String>> {
-    let vector_aggregator_address =
-        if let Some(vector_aggregator_config_map_name) = vector_aggregator_config_map_name {
-            let namespace = cluster
-                .meta()
-                .namespace
-                .as_deref()
-                .context(ObjectHasNoNamespaceSnafu)?;
-            let vector_aggregator_address = client
-                .get::<ConfigMap>(vector_aggregator_config_map_name, namespace)
-                .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 ConfigMap with logging and Vector configurations
 pub fn extend_config_map_with_log_config<C, K>(
     rolegroup: &RoleGroupRef<K>,
-    vector_aggregator_address: Option<&str>,
     logging: &Logging<C>,
     main_container: &C,
     vector_container: &C,
@@ -111,11 +72,7 @@ where
     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),
         );
     }