move merge-env logic to history.rs and improve unit test

Author: adwk67

rust/crd/src/constants.rs

@@ -77,3 +77,4 @@ pub const SPARK_DEFAULTS_FILE_NAME: &str = "spark-defaults.conf";
 
 pub const SPARK_CLUSTER_ROLE: &str = "spark-k8s-clusterrole";
 pub const SPARK_UID: i64 = 1000;
+pub const METRICS_PORT: u16 = 18081;

rust/crd/src/history.rs

@@ -1,8 +1,11 @@
+use crate::s3logdir::S3LogDir;
+use crate::tlscerts;
 use crate::{affinity::history_affinity, constants::*};
 
 use product_config::{types::PropertyNameKind, ProductConfigManager};
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
+use stackable_operator::k8s_openapi::api::core::v1::EnvVar;
 use stackable_operator::role_utils::RoleGroup;
 use stackable_operator::{
     commons::{
@@ -232,6 +235,93 @@ impl SparkHistoryServer {
         )
         .context(InvalidProductConfigSnafu)
     }
+
+    pub fn merged_env(
+        &self,
+        s3logdir: &S3LogDir,
+        role_group_env_overrides: HashMap<String, String>,
+    ) -> Vec<EnvVar> {
+        // Maps env var name to env var object. This allows env_overrides to work
+        // as expected (i.e. users can override the env var value).
+        let mut vars: BTreeMap<String, EnvVar> = BTreeMap::new();
+        let role_env_overrides = &self.role().config.env_overrides;
+
+        // This env var prevents the history server from detaching itself from the
+        // start script because this leads to the Pod terminating immediately.
+        vars.insert(
+            "SPARK_NO_DAEMONIZE".to_string(),
+            EnvVar {
+                name: "SPARK_NO_DAEMONIZE".to_string(),
+                value: Some("true".into()),
+                value_from: None,
+            },
+        );
+        vars.insert(
+            "SPARK_DAEMON_CLASSPATH".to_string(),
+            EnvVar {
+                name: "SPARK_DAEMON_CLASSPATH".to_string(),
+                value: Some("/stackable/spark/extra-jars/*".into()),
+                value_from: None,
+            },
+        );
+
+        let mut history_opts = vec![
+        format!("-Dlog4j.configurationFile={VOLUME_MOUNT_PATH_LOG_CONFIG}/{LOG4J2_CONFIG_FILE}"),
+        format!(
+            "-Djava.security.properties={VOLUME_MOUNT_PATH_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}"
+        ),
+        format!("-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/config.yaml")
+    ];
+
+        // if TLS is enabled build truststore
+        if tlscerts::tls_secret_name(&s3logdir.bucket.connection).is_some() {
+            history_opts.extend(vec![
+                format!("-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12"),
+                format!("-Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD}"),
+                format!("-Djavax.net.ssl.trustStoreType=pkcs12"),
+            ]);
+        }
+
+        vars.insert(
+            "SPARK_HISTORY_OPTS".to_string(),
+            EnvVar {
+                name: "SPARK_HISTORY_OPTS".to_string(),
+                value: Some(history_opts.join(" ")),
+                value_from: None,
+            },
+        );
+
+        // apply the role overrides
+        let mut role_envs = role_env_overrides.iter().map(|env_var| {
+            (
+                env_var.0.clone(),
+                EnvVar {
+                    name: env_var.0.clone(),
+                    value: Some(env_var.1.to_owned()),
+                    value_from: None,
+                },
+            )
+        });
+
+        vars.extend(&mut role_envs);
+
+        // apply the role-group overrides
+        let mut role_group_envs = role_group_env_overrides.into_iter().map(|env_var| {
+            (
+                env_var.0.clone(),
+                EnvVar {
+                    name: env_var.0.clone(),
+                    value: Some(env_var.1),
+                    value_from: None,
+                },
+            )
+        });
+
+        vars.extend(&mut role_group_envs);
+
+        // convert to Vec
+        vars.into_values().collect()
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, JsonSchema, Serialize, Display)]
@@ -368,6 +458,7 @@ impl Configuration for HistoryConfigFragment {
 mod test {
     use super::*;
     use indoc::indoc;
+    use stackable_operator::commons::s3::InlinedS3BucketSpec;
 
     #[test]
     pub fn test_env() {
@@ -401,17 +492,16 @@ mod test {
         let history: SparkHistoryServer =
             serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
 
-        assert_eq!(
-            Some(&"ROLE".to_string()),
-            history
-                .spec
-                .nodes
-                .config
-                .env_overrides
-                .get("TEST_SPARK_HIST_VAR")
-        );
-        assert_eq!(
-            Some(&"ROLEGROUP".to_string()),
+        let s3_log_dir: S3LogDir = S3LogDir {
+            bucket: InlinedS3BucketSpec {
+                bucket_name: None,
+                connection: None,
+            },
+            prefix: "prefix".to_string(),
+        };
+
+        let merged_env = history.merged_env(
+            &s3_log_dir,
             history
                 .spec
                 .nodes
@@ -420,7 +510,17 @@ mod test {
                 .unwrap()
                 .config
                 .env_overrides
-                .get("TEST_SPARK_HIST_VAR")
+                .clone(),
+        );
+
+        let env_map: BTreeMap<&str, Option<String>> = merged_env
+            .iter()
+            .map(|env_var| (env_var.name.as_str(), env_var.value.clone()))
+            .collect();
+
+        assert_eq!(
+            Some(&Some("ROLEGROUP".to_string())),
+            env_map.get("TEST_SPARK_HIST_VAR")
         );
     }
 }

rust/operator-binary/src/history/history_controller.rs

@@ -15,8 +15,7 @@ use stackable_operator::{
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
-                ConfigMap, EnvVar, PodSecurityContext, Service, ServiceAccount, ServicePort,
-                ServiceSpec,
+                ConfigMap, PodSecurityContext, Service, ServiceAccount, ServicePort, ServiceSpec,
             },
             rbac::v1::{ClusterRole, RoleBinding, RoleRef, Subject},
         },
@@ -36,14 +35,15 @@ use stackable_operator::{
     role_utils::RoleGroupRef,
     time::Duration,
 };
+use stackable_spark_k8s_crd::constants::METRICS_PORT;
 use stackable_spark_k8s_crd::{
     constants::{
         ACCESS_KEY_ID, APP_NAME, HISTORY_CONTROLLER_NAME, HISTORY_ROLE_NAME,
-        JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE, OPERATOR_NAME,
-        SECRET_ACCESS_KEY, SPARK_CLUSTER_ROLE, SPARK_DEFAULTS_FILE_NAME, SPARK_IMAGE_BASE_NAME,
-        SPARK_UID, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE, VOLUME_MOUNT_NAME_CONFIG,
-        VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG,
-        VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
+        JVM_SECURITY_PROPERTIES_FILE, MAX_SPARK_LOG_FILES_SIZE, OPERATOR_NAME, SECRET_ACCESS_KEY,
+        SPARK_CLUSTER_ROLE, SPARK_DEFAULTS_FILE_NAME, SPARK_IMAGE_BASE_NAME, SPARK_UID,
+        STACKABLE_TRUST_STORE, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
+        VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG,
+        VOLUME_MOUNT_PATH_LOG_CONFIG,
     },
     history,
     history::{HistoryConfig, SparkHistoryServer, SparkHistoryServerContainer},
@@ -59,8 +59,6 @@ use stackable_operator::k8s_openapi::DeepMerge;
 use stackable_operator::logging::controller::ReconcilerError;
 use strum::{EnumDiscriminants, IntoStaticStr};
 
-const METRICS_PORT: u16 = 18081;
-
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
@@ -444,11 +442,7 @@ fn build_stateful_set(
         .rolegroup(rolegroupref)
         .with_context(|_| CannotRetrieveRoleGroupSnafu)?;
 
-    let merged_env_vars = env_vars(
-        s3_log_dir,
-        shs.role().config.env_overrides.clone(),
-        role_group.config.env_overrides,
-    );
+    let merged_env = shs.merged_env(s3_log_dir, role_group.config.env_overrides);
 
     let container_name = "spark-history";
     let container = ContainerBuilder::new(container_name)
@@ -459,7 +453,7 @@ fn build_stateful_set(
         .args(command_args(s3_log_dir))
         .add_container_port("http", 18080)
         .add_container_port("metrics", METRICS_PORT.into())
-        .add_env_vars(merged_env_vars)
+        .add_env_vars(merged_env)
         .add_volume_mounts(s3_log_dir.volume_mounts())
         .add_volume_mount(VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_PATH_CONFIG)
         .add_volume_mount(VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_LOG_CONFIG)
@@ -487,10 +481,6 @@ fn build_stateful_set(
 
     let mut pod_template = pb.build_template();
     pod_template.merge_from(shs.role().config.pod_overrides.clone());
-    let role_group = shs
-        .rolegroup(rolegroupref)
-        .with_context(|_| CannotRetrieveRoleGroupSnafu)?;
-
     pod_template.merge_from(role_group.config.pod_overrides);
 
     Ok(StatefulSet {
@@ -680,98 +670,6 @@ fn command_args(s3logdir: &S3LogDir) -> Vec<String> {
     vec![String::from("-c"), command.join(" && ")]
 }
 
-fn env_vars(
-    s3logdir: &S3LogDir,
-    role_env_overrides: HashMap<String, String>,
-    role_group_env_overrides: HashMap<String, String>,
-) -> Vec<EnvVar> {
-    // Maps env var name to env var object. This allows env_overrides to work
-    // as expected (i.e. users can override the env var value).
-    let mut vars: BTreeMap<String, EnvVar> = BTreeMap::new();
-
-    // This env var prevents the history server from detaching itself from the
-    // start script because this leads to the Pod terminating immediately.
-    vars.insert(
-        "SPARK_NO_DAEMONIZE".to_string(),
-        EnvVar {
-            name: "SPARK_NO_DAEMONIZE".to_string(),
-            value: Some("true".into()),
-            value_from: None,
-        },
-    );
-    vars.insert(
-        "SPARK_DAEMON_CLASSPATH".to_string(),
-        EnvVar {
-            name: "SPARK_DAEMON_CLASSPATH".to_string(),
-            value: Some("/stackable/spark/extra-jars/*".into()),
-            value_from: None,
-        },
-    );
-
-    let mut history_opts = vec![
-        format!("-Dlog4j.configurationFile={VOLUME_MOUNT_PATH_LOG_CONFIG}/{LOG4J2_CONFIG_FILE}"),
-        format!(
-            "-Djava.security.properties={VOLUME_MOUNT_PATH_CONFIG}/{JVM_SECURITY_PROPERTIES_FILE}"
-        ),
-        format!("-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/config.yaml")
-    ];
-
-    // if TLS is enabled build truststore
-    if tlscerts::tls_secret_name(&s3logdir.bucket.connection).is_some() {
-        history_opts.extend(vec![
-            format!("-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12"),
-            format!("-Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD}"),
-            format!("-Djavax.net.ssl.trustStoreType=pkcs12"),
-        ]);
-    }
-
-    vars.insert(
-        "SPARK_HISTORY_OPTS".to_string(),
-        EnvVar {
-            name: "SPARK_HISTORY_OPTS".to_string(),
-            value: Some(history_opts.join(" ")),
-            value_from: None,
-        },
-    );
-
-    // apply the role overrides
-    let mut role_envs = role_env_overrides
-        .into_iter()
-        .map(|env_var| {
-            (
-                env_var.0.clone(),
-                EnvVar {
-                    name: env_var.0.clone(),
-                    value: Some(env_var.1),
-                    value_from: None,
-                },
-            )
-        })
-        .collect();
-
-    vars.append(&mut role_envs);
-
-    // apply the role-group overrides
-    let mut role_group_envs = role_group_env_overrides
-        .into_iter()
-        .map(|env_var| {
-            (
-                env_var.0.clone(),
-                EnvVar {
-                    name: env_var.0.clone(),
-                    value: Some(env_var.1),
-                    value_from: None,
-                },
-            )
-        })
-        .collect();
-
-    vars.append(&mut role_group_envs);
-
-    // convert to Vec
-    vars.into_values().collect()
-}
-
 fn labels<'a, T>(
     shs: &'a T,
     app_version_label: &'a str,