Fix Spark driver not getting imagePullPolicy set

Reported in https://stackable-workspace.slack.com/archives/C037Q9DPF4P/p1682326403038869

Author: sbernauer

CHANGELOG.md

@@ -18,7 +18,8 @@ All notable changes to this project will be documented in this file.
 
 ### Fixed
 
-- Fix quoting issues when spark config values contain spaces ([#243])
+- Fix quoting issues when spark config values contain spaces ([#243]).
+- Fix Spark driver not getting imagePullPolicy set ([#243]).
 
 [#235]: https://github.com/stackabletech/spark-k8s-operator/pull/235
 [#236]: https://github.com/stackabletech/spark-k8s-operator/pull/236

rust/crd/src/lib.rs

@@ -171,10 +171,10 @@ pub struct SparkApplicationSpec {
     pub image: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub spark_image: Option<String>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub spark_image_pull_policy: Option<ImagePullPolicy>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub spark_image_pull_secrets: Option<Vec<LocalObjectReference>>,
+    #[serde(default)]
+    pub spark_image_pull_policy: ImagePullPolicy,
+    #[serde(default)]
+    pub spark_image_pull_secrets: Vec<LocalObjectReference>,
     /// Name of the Vector aggregator discovery ConfigMap.
     /// It must contain the key `ADDRESS` with the address of the Vector aggregator.
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -205,9 +205,12 @@ pub struct SparkApplicationSpec {
     pub log_file_directory: Option<LogFileDirectorySpec>,
 }
 
-#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize, Display, EnumString)]
+#[derive(
+    Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize, Display, EnumString,
+)]
 pub enum ImagePullPolicy {
     Always,
+    #[default]
     IfNotPresent,
     Never,
 }
@@ -250,11 +253,11 @@ impl SparkApplication {
         self.spec.image.as_deref()
     }
 
-    pub fn spark_image_pull_policy(&self) -> Option<ImagePullPolicy> {
+    pub fn spark_image_pull_policy(&self) -> ImagePullPolicy {
         self.spec.spark_image_pull_policy.clone()
     }
 
-    pub fn spark_image_pull_secrets(&self) -> Option<Vec<LocalObjectReference>> {
+    pub fn spark_image_pull_secrets(&self) -> Vec<LocalObjectReference> {
         self.spec.spark_image_pull_secrets.clone()
     }
 
@@ -466,6 +469,7 @@ impl SparkApplication {
             format!("--conf spark.kubernetes.executor.podTemplateContainerName={container_name}", container_name = SparkContainer::Spark),
             format!("--conf spark.kubernetes.namespace={}", self.metadata.namespace.as_ref().context(NoNamespaceSnafu)?),
             format!("--conf spark.kubernetes.driver.container.image={}", self.spec.spark_image.as_ref().context(NoSparkImageSnafu)?),
+            format!("--conf spark.kubernetes.container.image.pullPolicy={}", self.spark_image_pull_policy()),
             format!("--conf spark.kubernetes.executor.container.image={}", self.spec.spark_image.as_ref().context(NoSparkImageSnafu)?),
             format!("--conf spark.kubernetes.authenticate.driver.serviceAccountName={}", serviceaccount_name),
             format!("--conf spark.driver.defaultJavaOptions=-Dlog4j.configurationFile={VOLUME_MOUNT_PATH_LOG_CONFIG}/{LOG4J2_CONFIG_FILE}"),
@@ -1129,13 +1133,13 @@ spec:
         .unwrap();
 
         assert_eq!(
-            Some(vec![LocalObjectReference {
+            vec![LocalObjectReference {
                 name: Some("myregistrykey".to_string())
-            }]),
+            }],
             spark_application.spark_image_pull_secrets()
         );
         assert_eq!(
-            Some(ImagePullPolicy::Always),
+            ImagePullPolicy::Always,
             spark_application.spark_image_pull_policy()
         );
     }

rust/operator-binary/src/spark_k8s_controller.rs

@@ -365,9 +365,7 @@ fn init_containers(
             .args(vec![args.join(" && ")])
             .add_volume_mount(VOLUME_MOUNT_NAME_REQ, VOLUME_MOUNT_PATH_REQ)
             .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG);
-        if let Some(image_pull_policy) = spark_application.spark_image_pull_policy() {
-            rcb.image_pull_policy(image_pull_policy.to_string());
-        }
+        rcb.image_pull_policy(spark_application.spark_image_pull_policy().to_string());
         rcb.build()
     });
 
@@ -402,10 +400,7 @@ fn pod_template(
             .join("; "),
         );
     }
-
-    if let Some(image_pull_policy) = spark_application.spark_image_pull_policy() {
-        cb.image_pull_policy(image_pull_policy.to_string());
-    }
+    cb.image_pull_policy(spark_application.spark_image_pull_policy().to_string());
 
     let mut pb = PodBuilder::new();
     pb.metadata(
@@ -430,13 +425,12 @@ fn pod_template(
         pb.add_init_container(init_container.clone());
     }
 
-    if let Some(image_pull_secrets) = spark_application.spark_image_pull_secrets() {
-        pb.image_pull_secrets(
-            image_pull_secrets
-                .iter()
-                .flat_map(|secret| secret.name.clone()),
-        );
-    }
+    pb.image_pull_secrets(
+        spark_application
+            .spark_image_pull_secrets()
+            .iter()
+            .flat_map(|secret| secret.name.clone()),
+    );
 
     if config.logging.enable_vector_agent {
         pb.add_container(vector_container(
@@ -620,9 +614,7 @@ fn spark_job(
         // TODO: move this to the image
         .add_env_var("SPARK_CONF_DIR", "/stackable/spark/conf");
 
-    if let Some(image_pull_policy) = spark_application.spark_image_pull_policy() {
-        cb.image_pull_policy(image_pull_policy.to_string());
-    }
+    cb.image_pull_policy(spark_application.spark_image_pull_policy().to_string());
 
     let mut volumes = vec![
         VolumeBuilder::new(VOLUME_MOUNT_NAME_CONFIG)
@@ -679,7 +671,7 @@ fn spark_job(
             restart_policy: Some("Never".to_string()),
             service_account_name: serviceaccount.metadata.name.clone(),
             volumes: Some(volumes),
-            image_pull_secrets: spark_application.spark_image_pull_secrets(),
+            image_pull_secrets: Some(spark_application.spark_image_pull_secrets()),
             security_context: Some(security_context()),
             ..PodSpec::default()
         }),