minor refactor

Author: labrenbe

rust/stackable-cockpit/src/platform/release/spec.rs

@@ -2,7 +2,7 @@ use futures::{StreamExt as _, TryStreamExt};
 use indexmap::IndexMap;
 use reqwest::StatusCode;
 use serde::{Deserialize, Serialize};
-use serde_yaml::{Mapping, Value};
+use serde_yaml::Mapping;
 use snafu::{ResultExt, Snafu};
 use tokio::task::JoinError;
 use tracing::{Instrument, Span, debug, info, instrument};
@@ -19,7 +19,7 @@ use crate::{
     utils::{
         k8s::{self, Client},
         path::{IntoPathOrUrl as _, PathOrUrlParseError},
-        yaml::deep_merge,
+        yaml::merged_values_for_operator,
     },
     xfer::{self, processor::Text},
 };
@@ -104,24 +104,14 @@ impl ReleaseSpec {
         Span::current().pb_set_length(operators.len() as u64);
 
         let namespace = namespace.to_string();
-        let common_values = operator_values
-            .get("common")
-            .and_then(Value::as_mapping)
-            .cloned()
-            .unwrap_or_default();
         futures::stream::iter(operators)
             .map(|(product_name, product)| {
                 let task_span =
                     tracing::info_span!("install_operator", product_name = tracing::field::Empty);
 
                 let namespace = namespace.clone();
                 let chart_source = chart_source.clone();
-                let operator_specific = operator_values
-                    .get(format!("{product_name}-operator"))
-                    .and_then(Value::as_mapping)
-                    .cloned()
-                    .unwrap_or_default();
-                let merged_values = deep_merge(common_values.clone(), operator_specific);
+                let merged_values = merged_values_for_operator(operator_values, &product_name);
                 // Helm installs currently `block_in_place`, so we need to spawn each job onto a separate task to
                 // get useful parallelism.
                 tokio::spawn(

rust/stackable-cockpit/src/utils/yaml.rs

@@ -1,5 +1,23 @@
 use serde_yaml::{Mapping, Value};
 
+/// Extracts the merged Helm values for a specific operator from the operator values mapping.
+///
+/// Looks up the `common` key and the operator-specific key (e.g. `airflow-operator`),
+/// then deep-merges them with operator-specific values taking precedence.
+pub fn merged_values_for_operator(operator_values: &Mapping, operator_name: &str) -> Mapping {
+    let common = operator_values
+        .get("common")
+        .and_then(Value::as_mapping)
+        .cloned()
+        .unwrap_or_default();
+    let operator_specific = operator_values
+        .get(format!("{operator_name}-operator"))
+        .and_then(Value::as_mapping)
+        .cloned()
+        .unwrap_or_default();
+    deep_merge(common, operator_specific)
+}
+
 /// Deep merges `overlay` into `base`. Overlay values take precedence.
 /// When both values are mappings, their contents are merged recursively.
 /// Non-mapping values (including sequences) are replaced entirely, not merged.

rust/stackablectl/src/cmds/operator.rs

@@ -8,7 +8,6 @@ use comfy_table::{
 use indexmap::IndexMap;
 use semver::Version;
 use serde::Serialize;
-use serde_yaml::Value;
 use snafu::{ResultExt, Snafu};
 use stackable_cockpit::{
     constants::{
@@ -25,7 +24,7 @@ use stackable_cockpit::{
         chartsource::ChartSourceMetadata,
         k8s::{self, Client},
         path::PathOrUrlParseError,
-        yaml::deep_merge,
+        yaml::merged_values_for_operator,
     },
     xfer,
 };
@@ -351,18 +350,8 @@ async fn install_cmd(
         .await
         .context(FileTransferSnafu)?;
 
-    let common_values = operator_values
-        .get("common")
-        .and_then(Value::as_mapping)
-        .cloned()
-        .unwrap_or_default();
     for operator in &args.operators {
-        let operator_specific = operator_values
-            .get(format!("{}-operator", &operator.name))
-            .and_then(Value::as_mapping)
-            .cloned()
-            .unwrap_or_default();
-        let merged_values = deep_merge(common_values.clone(), operator_specific);
+        let merged_values = merged_values_for_operator(&operator_values, &operator.name);
 
         operator
             .install(