chore: Wrap SparkApplicationSpec in SparkApplicationTemplateSpec (#685)

Author: sbernauer

rust/operator-binary/src/crd/template_merger.rs

@@ -1209,8 +1209,8 @@ mod tests {
 
         // Convert templates to SparkApplication and merge left-to-right:
         // template-a has the lowest priority, the spark application has the highest.
-        let app_from_a = crate::crd::v1alpha1::SparkApplication::from(&template_a);
-        let app_from_b = crate::crd::v1alpha1::SparkApplication::from(&template_b);
+        let app_from_a = crate::crd::v1alpha1::SparkApplication::from(template_a);
+        let app_from_b = crate::crd::v1alpha1::SparkApplication::from(template_b);
         // This how `merge_application_templates()` does it
         let template_apps = [app_from_a, app_from_b, spark_app];
         let merged = template_apps
@@ -1391,7 +1391,7 @@ mod tests {
         "#})
         .unwrap();
 
-        let app_from_template = crate::crd::v1alpha1::SparkApplication::from(&template);
+        let app_from_template = crate::crd::v1alpha1::SparkApplication::from(template);
         let merged = deep_merge(&app_from_template, &spark_app);
 
         let submit_affinity = merged.submit_config().unwrap().affinity;

rust/operator-binary/src/crd/template_spec.rs

@@ -1,25 +1,17 @@
 //! This module provides the SparkApplicationTemplateSpec CRD definition.
 
-use std::{collections::HashMap, num::ParseIntError, str::ParseBoolError};
+use std::{num::ParseIntError, str::ParseBoolError};
 
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
-    commons::product_image_selection::ProductImage,
-    crd::s3,
-    k8s_openapi::api::core::v1::{EnvVar, Volume},
     kube::{Api, CustomResource, ResourceExt, api::ListParams},
     schemars::{self, JsonSchema},
-    utils::crds::raw_object_list_schema,
     versioned::versioned,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
-use super::{
-    SparkApplicationDriverRoleType, SparkApplicationExecutorRoleType, SparkApplicationJobRoleType,
-    history::LogFileDirectorySpec, job_dependencies::JobDependencies, roles::SparkMode,
-};
 use crate::crd::template_merger::deep_merge;
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -77,85 +69,11 @@ pub mod versioned {
     #[derive(Clone, CustomResource, Debug, Deserialize, JsonSchema, Serialize)]
     #[serde(rename_all = "camelCase")]
     pub struct SparkApplicationTemplateSpec {
-        /// Mode: cluster or client. Currently only cluster is supported.
-        pub mode: SparkMode,
-
-        /// The main class - i.e. entry point - for JVM artifacts.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub main_class: Option<String>,
-
-        /// The actual application file that will be called by `spark-submit`.
-        pub main_application_file: String,
-
-        /// User-supplied image containing spark-job dependencies that will be copied to the specified volume mount.
-        /// See the [examples](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/examples).
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub image: Option<String>,
-
-        // no doc - docs in ProductImage struct.
-        pub spark_image: ProductImage,
-
-        /// Name of the Vector aggregator [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/concepts/service_discovery).
-        /// It must contain the key `ADDRESS` with the address of the Vector aggregator.
-        /// Follow the [logging tutorial](DOCS_BASE_URL_PLACEHOLDER/tutorials/logging-vector-aggregator)
-        /// to learn how to configure log aggregation with Vector.
-        #[serde(skip_serializing_if = "Option::is_none")]
-        pub vector_aggregator_config_map_name: Option<String>,
-
-        /// The job builds a spark-submit command, complete with arguments and referenced dependencies
-        /// such as templates, and passes it on to Spark.
-        /// The reason this property uses its own type (SubmitConfigFragment) is because logging is not
-        /// supported for spark-submit processes.
-        //
-        // IMPORTANT: Please note that the jvmArgumentOverrides have no effect here!
-        // However, due to product-config things I wasn't able to remove them.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub job: Option<SparkApplicationJobRoleType>,
-
-        /// The driver role specifies the configuration that, together with the driver pod template, is used by
-        /// Spark to create driver pods.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub driver: Option<SparkApplicationDriverRoleType>,
-
-        /// The executor role specifies the configuration that, together with the driver pod template, is used by
-        /// Spark to create the executor pods.
-        /// This is RoleGroup instead of plain CommonConfiguration because it needs to allow for the number of replicas.
-        /// to be specified.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub executor: Option<SparkApplicationExecutorRoleType>,
-
-        /// A map of key/value strings that will be passed directly to spark-submit.
-        #[serde(default)]
-        pub spark_conf: HashMap<String, String>,
-
-        /// Job dependencies: a list of python packages that will be installed via pip, a list of packages
-        /// or repositories that is passed directly to spark-submit, or a list of excluded packages
-        /// (also passed directly to spark-submit).
-        #[serde(default)]
-        pub deps: JobDependencies,
-
-        /// Configure an S3 connection that the SparkApplication has access to.
-        /// Read more in the [Spark S3 usage guide](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/s3).
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub s3connection: Option<s3::v1alpha1::InlineConnectionOrReference>,
-
-        /// Arguments passed directly to the job artifact.
-        #[serde(default)]
-        pub args: Vec<String>,
-
-        /// A list of volumes that can be made available to the job, driver or executors via their volume mounts.
-        #[serde(default)]
-        #[schemars(schema_with = "raw_object_list_schema")]
-        pub volumes: Vec<Volume>,
-
-        /// A list of environment variables that will be set in the job pod and the driver and executor
-        /// pod templates.
-        #[serde(default)]
-        pub env: Vec<EnvVar>,
-
-        /// The log file directory definition used by the Spark history server.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub log_file_directory: Option<LogFileDirectorySpec>,
+        /// The template body mirrors a [`SparkApplication`](super::super::v1alpha1::SparkApplication)
+        /// spec exactly — it is merged into a `SparkApplication` at reconcile time, see
+        /// [`merge_application_templates`].
+        #[serde(flatten)]
+        pub spec: crate::crd::v1alpha1::SparkApplicationSpec,
     }
 }
 
@@ -170,33 +88,11 @@ impl From<&v1alpha1::SparkApplicationTemplate>
     }
 }
 
-impl From<&v1alpha1::SparkApplicationTemplate> for super::v1alpha1::SparkApplication {
-    fn from(template: &v1alpha1::SparkApplicationTemplate) -> super::v1alpha1::SparkApplication {
-        let spec = super::v1alpha1::SparkApplicationSpec {
-            mode: template.spec.mode.clone(),
-            main_class: template.spec.main_class.clone(),
-            main_application_file: template.spec.main_application_file.clone(),
-            image: template.spec.image.clone(),
-            spark_image: template.spec.spark_image.clone(),
-            vector_aggregator_config_map_name: template
-                .spec
-                .vector_aggregator_config_map_name
-                .clone(),
-            job: template.spec.job.clone(),
-            driver: template.spec.driver.clone(),
-            executor: template.spec.executor.clone(),
-            spark_conf: template.spec.spark_conf.clone(),
-            deps: template.spec.deps.clone(),
-            s3connection: template.spec.s3connection.clone(),
-            args: template.spec.args.clone(),
-            volumes: template.spec.volumes.clone(),
-            env: template.spec.env.clone(),
-            log_file_directory: template.spec.log_file_directory.clone(),
-        };
-
+impl From<v1alpha1::SparkApplicationTemplate> for super::v1alpha1::SparkApplication {
+    fn from(template: v1alpha1::SparkApplicationTemplate) -> super::v1alpha1::SparkApplication {
         super::v1alpha1::SparkApplication {
-            metadata: template.metadata.clone(),
-            spec,
+            metadata: template.metadata,
+            spec: template.spec.spec,
             status: None,
         }
     }
@@ -379,6 +275,7 @@ pub(crate) async fn merge_application_templates(
             // which has the highest priority during merging.
             let mut template_apps: Vec<super::v1alpha1::SparkApplication> = templates
                 .iter()
+                .cloned()
                 .map(super::v1alpha1::SparkApplication::from)
                 .collect::<Vec<super::v1alpha1::SparkApplication>>();
             template_apps.push(spark_application.clone());