From 05a3440bb2b11f69f954e2dec18967ff57002240 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 17:51:36 +0200 Subject: [PATCH 1/5] refactor: Introduce build aggregator for SparkApplication --- .../src/spark_k8s_controller.rs | 128 +++++------------- .../src/spark_k8s_controller/build/mod.rs | 100 ++++++++++++++ 2 files changed, 133 insertions(+), 95 deletions(-) diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs index a7a128b7..7b0bf1f0 100644 --- a/rust/operator-binary/src/spark_k8s_controller.rs +++ b/rust/operator-binary/src/spark_k8s_controller.rs @@ -3,6 +3,11 @@ use std::sync::Arc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::{self}, + k8s_openapi::api::{ + batch::v1::Job, + core::v1::{ConfigMap, ServiceAccount}, + rbac::v1::RoleBinding, + }, kube::{ ResourceExt, core::{DeserializeGuard, error_boundary}, @@ -115,6 +120,18 @@ impl ReconcilerError for Error { } } +/// Every Kubernetes resource produced by the build step for a SparkApplication. +/// +/// Built without a Kubernetes client: all references are already dereferenced and validated by +/// this point, so the only errors possible during assembly are resource-construction failures. +pub struct SparkResources { + pub service_account: ServiceAccount, + pub role_binding: RoleBinding, + /// Driver pod-template, executor pod-template, and submit-job ConfigMaps (in that order). + pub config_maps: Vec, + pub job: Job, +} + pub async fn reconcile( spark_application: Arc>, ctx: Arc, @@ -147,117 +164,38 @@ pub async fn reconcile( .context(ValidateSparkApplicationSnafu)?; let spark_application = &validated.spark_application; - let opt_s3conn = &validated.cluster_config.s3_connection; - let logdir = &validated.cluster_config.log_dir; - let resolved_product_image = &validated.resolved_product_image; // This is the final version of the spark app to reconcile. // No more mutating operations after this point (except for status). tracing::debug!("reconciling spark application [{spark_application:?}]"); - let (serviceaccount, rolebinding) = - build::resource::serviceaccount::build_spark_role_serviceaccount(&validated)?; - client - .apply_patch(SPARK_CONTROLLER_NAME, &serviceaccount, &serviceaccount) - .await - .context(ApplyServiceAccountSnafu)?; - client - .apply_patch(SPARK_CONTROLLER_NAME, &rolebinding, &rolebinding) - .await - .context(ApplyRoleBindingSnafu)?; - - let env_vars = spark_application.env(opt_s3conn, logdir); - - let driver_config = spark_application - .driver_config() - .context(FailedToResolveConfigSnafu)?; - - let driver_config_overrides = spark_application - .spec - .driver - .as_ref() - .map(|driver| driver.config_overrides.clone()) - .unwrap_or_default(); - - let driver_pod_template_config_map = build::resource::config_map::pod_template_config_map( - &validated, - SparkApplicationRole::Driver, - &driver_config, - &driver_config_overrides, - &env_vars, - &serviceaccount, - )?; - client - .apply_patch( - SPARK_CONTROLLER_NAME, - &driver_pod_template_config_map, - &driver_pod_template_config_map, - ) - .await - .context(ApplyApplicationSnafu)?; - - let executor_config = spark_application - .executor_config() - .context(FailedToResolveConfigSnafu)?; + let resources = build::build(&validated)?; - let executor_config_overrides = spark_application - .spec - .executor - .as_ref() - .map(|executor| executor.config.config_overrides.clone()) - .unwrap_or_default(); - - let executor_pod_template_config_map = build::resource::config_map::pod_template_config_map( - &validated, - SparkApplicationRole::Executor, - &executor_config, - &executor_config_overrides, - &env_vars, - &serviceaccount, - )?; + // Apply the ServiceAccount and RoleBinding first, then the ConfigMaps, and finally the Job: + // the Job runs under the ServiceAccount and mounts the ConfigMaps, so they must exist first. client .apply_patch( SPARK_CONTROLLER_NAME, - &executor_pod_template_config_map, - &executor_pod_template_config_map, + &resources.service_account, + &resources.service_account, ) .await - .context(ApplyApplicationSnafu)?; - - let job_commands = spark_application - .build_command(opt_s3conn, logdir, &resolved_product_image.image) - .context(BuildCommandSnafu)?; - - let submit_config = spark_application - .submit_config() - .context(SubmitConfigSnafu)?; - - let submit_config_overrides = spark_application - .spec - .job - .as_ref() - .map(|job| job.config_overrides.clone()) - .unwrap_or_default(); - - let submit_job_config_map = - build::resource::config_map::submit_job_config_map(&validated, &submit_config_overrides)?; + .context(ApplyServiceAccountSnafu)?; client .apply_patch( SPARK_CONTROLLER_NAME, - &submit_job_config_map, - &submit_job_config_map, + &resources.role_binding, + &resources.role_binding, ) .await - .context(ApplyApplicationSnafu)?; - - let job = build::resource::job::spark_job( - &validated, - &serviceaccount, - &env_vars, - &job_commands, - &submit_config, - )?; + .context(ApplyRoleBindingSnafu)?; + for config_map in &resources.config_maps { + client + .apply_patch(SPARK_CONTROLLER_NAME, config_map, config_map) + .await + .context(ApplyApplicationSnafu)?; + } client - .apply_patch(SPARK_CONTROLLER_NAME, &job, &job) + .apply_patch(SPARK_CONTROLLER_NAME, &resources.job, &resources.job) .await .context(ApplyApplicationSnafu)?; diff --git a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs index f94ce8fb..0167c4f9 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs @@ -1,2 +1,102 @@ pub mod pod; pub mod resource; + +use snafu::ResultExt; + +use crate::{ + crd::roles::SparkApplicationRole, + spark_k8s_controller::{ + BuildCommandSnafu, FailedToResolveConfigSnafu, Result, SparkResources, SubmitConfigSnafu, + validate::ValidatedSparkApplication, + }, +}; + +/// Builds every Kubernetes resource for the given validated SparkApplication. +pub fn build(validated: &ValidatedSparkApplication) -> Result { + let spark_application = &validated.spark_application; + let opt_s3conn = &validated.cluster_config.s3_connection; + let logdir = &validated.cluster_config.log_dir; + let resolved_product_image = &validated.resolved_product_image; + + let (service_account, role_binding) = + resource::serviceaccount::build_spark_role_serviceaccount(validated)?; + + let env_vars = spark_application.env(opt_s3conn, logdir); + + let driver_config = spark_application + .driver_config() + .context(FailedToResolveConfigSnafu)?; + + let driver_config_overrides = spark_application + .spec + .driver + .as_ref() + .map(|driver| driver.config_overrides.clone()) + .unwrap_or_default(); + + let driver_pod_template_config_map = resource::config_map::pod_template_config_map( + validated, + SparkApplicationRole::Driver, + &driver_config, + &driver_config_overrides, + &env_vars, + &service_account, + )?; + + let executor_config = spark_application + .executor_config() + .context(FailedToResolveConfigSnafu)?; + + let executor_config_overrides = spark_application + .spec + .executor + .as_ref() + .map(|executor| executor.config.config_overrides.clone()) + .unwrap_or_default(); + + let executor_pod_template_config_map = resource::config_map::pod_template_config_map( + validated, + SparkApplicationRole::Executor, + &executor_config, + &executor_config_overrides, + &env_vars, + &service_account, + )?; + + let job_commands = spark_application + .build_command(opt_s3conn, logdir, &resolved_product_image.image) + .context(BuildCommandSnafu)?; + + let submit_config = spark_application + .submit_config() + .context(SubmitConfigSnafu)?; + + let submit_config_overrides = spark_application + .spec + .job + .as_ref() + .map(|job| job.config_overrides.clone()) + .unwrap_or_default(); + + let submit_job_config_map = + resource::config_map::submit_job_config_map(validated, &submit_config_overrides)?; + + let job = resource::job::spark_job( + validated, + &service_account, + &env_vars, + &job_commands, + &submit_config, + )?; + + Ok(SparkResources { + service_account, + role_binding, + config_maps: vec![ + driver_pod_template_config_map, + executor_pod_template_config_map, + submit_job_config_map, + ], + job, + }) +} From 64cac02cd1deae55907a056ad0ba6a57d634be30 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 18:00:02 +0200 Subject: [PATCH 2/5] refactor: Introduce build aggregator for SparkConnectServer --- .../operator-binary/src/connect/controller.rs | 223 +++++------------- .../src/connect/controller/build/mod.rs | 107 +++++++++ 2 files changed, 168 insertions(+), 162 deletions(-) diff --git a/rust/operator-binary/src/connect/controller.rs b/rust/operator-binary/src/connect/controller.rs index cd3467e6..70b67ec4 100644 --- a/rust/operator-binary/src/connect/controller.rs +++ b/rust/operator-binary/src/connect/controller.rs @@ -4,6 +4,12 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ cluster_resources::ClusterResourceApplyStrategy, commons::rbac::build_rbac_resources, + crd::listener, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service, ServiceAccount}, + rbac::v1::RoleBinding, + }, kube::{ ResourceExt, core::{DeserializeGuard, error_boundary}, @@ -20,15 +26,7 @@ use stackable_operator::{ use strum::{EnumDiscriminants, IntoStaticStr}; use super::crd::{CONNECT_APP_NAME, v1alpha1}; -use crate::{ - Ctx, - connect::{ - common, - controller::build::{executor, server, service}, - crd::SparkConnectServerStatus, - }, - crd::constants::OPERATOR_NAME, -}; +use crate::{Ctx, connect::crd::SparkConnectServerStatus, crd::constants::OPERATOR_NAME}; pub mod build; pub mod dereference; @@ -38,57 +36,11 @@ pub mod validate; #[strum_discriminants(derive(IntoStaticStr))] #[allow(clippy::enum_variant_names)] pub enum Error { - #[snafu(display("failed to apply spark connect listener"))] - ApplyListener { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to serialize connect properties"))] - SerializeProperties { source: common::Error }, - - #[snafu(display("failed to build connect executor properties"))] - ExecutorProperties { source: executor::Error }, - - #[snafu(display("failed to build connect server properties"))] - ServerProperties { source: server::Error }, - - #[snafu(display("failed to build spark connect executor config map for {name}"))] - BuildExecutorConfigMap { - source: executor::Error, - name: String, - }, - - #[snafu(display("failed to build spark connect server config map for {name}"))] - BuildServerConfigMap { source: server::Error, name: String }, - - #[snafu(display("failed to build spark connect stateful set"))] - BuildServerStatefulSet { source: server::Error }, - - #[snafu(display("failed to update status of spark connect server {name}"))] - ApplyStatus { - source: stackable_operator::client::Error, - name: String, - }, - - #[snafu(display("failed to update the connect server stateful set"))] - ApplyStatefulSet { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update connect executor config map for {name}"))] - ApplyExecutorConfigMap { - source: stackable_operator::cluster_resources::Error, - name: String, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to update connect server config map for {name}"))] - ApplyServerConfigMap { - source: stackable_operator::cluster_resources::Error, - name: String, - }, - - #[snafu(display("failed to update connect server service"))] - ApplyService { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, }, @@ -102,6 +54,12 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, + #[snafu(display("failed to update status of spark connect server {name}"))] + ApplyStatus { + source: stackable_operator::client::Error, + name: String, + }, + #[snafu(display("failed to delete orphaned resources"))] DeleteOrphanedResources { source: stackable_operator::cluster_resources::Error, @@ -123,17 +81,6 @@ pub enum Error { source: stackable_operator::commons::rbac::Error, }, - #[snafu(display("failed to build connect executor pod template"))] - ExecutorPodTemplate { - source: crate::connect::controller::build::executor::Error, - }, - - #[snafu(display("failed to serialize executor pod template"))] - ExecutorPodTemplateSerde { source: serde_yaml::Error }, - - #[snafu(display("failed to build connect server S3 properties"))] - S3SparkProperties { source: crate::connect::s3::Error }, - #[snafu(display("failed to dereference SparkConnectServer"))] DereferenceSparkConnectServer { source: dereference::Error }, @@ -148,7 +95,22 @@ impl ReconcilerError for Error { ErrorDiscriminants::from(self).into() } } -/// Updates the status of the SparkApplication that started the pod. + +/// Every Kubernetes resource produced by the build step for a SparkConnectServer. +/// +/// Built without a Kubernetes client: all references are already dereferenced and validated by +/// this point, so the only errors possible during assembly are resource-construction failures. +pub struct SparkConnectResources { + pub service_account: ServiceAccount, + pub role_binding: RoleBinding, + /// The headless Service (for executors to reach the driver) and the metrics Service. + pub services: Vec, + /// The executor and server ConfigMaps. + pub config_maps: Vec, + pub listener: listener::v1alpha1::Listener, + pub stateful_set: StatefulSet, +} + pub async fn reconcile( scs: Arc>, ctx: Arc, @@ -177,8 +139,6 @@ pub async fn reconcile( "Validated SparkConnectServer identity" ); - let resolved_s3 = &validated.cluster_config.resolved_s3; - let mut cluster_resources = cluster_resources_new( &validate::product_name(), &validate::operator_name(), @@ -190,7 +150,9 @@ pub async fn reconcile( &scs.spec.object_overrides, ); - // Use a dedicated service account for connect server pods. + // Use a dedicated service account for connect server pods. Building the RBAC resources needs + // the cluster-resource labels, so it stays in the reconcile step; the built objects (whose + // names are deterministic) are handed to the client-free build step. let (service_account, role_binding) = build_rbac_resources( scs, CONNECT_APP_NAME, @@ -200,106 +162,43 @@ pub async fn reconcile( ) .context(BuildRbacResourcesSnafu)?; - let service_account = cluster_resources - .add(client, service_account) - .await - .context(ApplyServiceAccountSnafu)?; - cluster_resources - .add(client, role_binding) - .await - .context(ApplyRoleBindingSnafu)?; - - // Headless service used by executors connect back to the driver - let headless_service = service::build_headless_service(&validated); - - let applied_headless_service = cluster_resources - .add(client, headless_service.clone()) - .await - .context(ApplyServiceSnafu)?; - - // Metrics service used for scraping - let metrics_service = service::build_metrics_service(&validated); + let resources = build::build(&validated, service_account, role_binding, &scs.spec.args) + .context(BuildResourcesSnafu)?; + // Apply order: ServiceAccount and RoleBinding first, then the Services, ConfigMaps and + // Listener, and finally the StatefulSet (it mounts the ConfigMaps and runs under the SA, so + // they must exist first). cluster_resources - .add(client, metrics_service.clone()) + .add(client, resources.service_account) .await - .context(ApplyServiceSnafu)?; - - // ======================================== - // Server config map - - let spark_props = common::spark_properties(&[ - resolved_s3 - .spark_properties() - .context(S3SparkPropertiesSnafu)?, - server::server_properties(&validated, &applied_headless_service, &service_account) - .context(ServerPropertiesSnafu)?, - executor::executor_properties(&validated).context(ExecutorPropertiesSnafu)?, - ]) - .context(SerializePropertiesSnafu)?; - - // ======================================== - // Executor config map and pod template - let executor_config_map = - executor::executor_config_map(&validated).context(BuildExecutorConfigMapSnafu { - name: validated.name_any(), - })?; + .context(ApplyServiceAccountSnafu)?; cluster_resources - .add(client, executor_config_map.clone()) + .add(client, resources.role_binding) .await - .context(ApplyExecutorConfigMapSnafu { - name: validated.name_any(), - })?; - - let executor_pod_template = serde_yaml::to_string( - &executor::executor_pod_template(&validated, &executor_config_map) - .context(ExecutorPodTemplateSnafu)?, - ) - .context(ExecutorPodTemplateSerdeSnafu)?; - - // ======================================== - // Server config map - let server_config_map = - server::server_config_map(&validated, &spark_props, &executor_pod_template).context( - BuildServerConfigMapSnafu { - name: validated.name_any(), - }, - )?; + .context(ApplyRoleBindingSnafu)?; + for service in resources.services { + cluster_resources + .add(client, service) + .await + .context(ApplyResourceSnafu)?; + } + for config_map in resources.config_maps { + cluster_resources + .add(client, config_map) + .await + .context(ApplyResourceSnafu)?; + } cluster_resources - .add(client, server_config_map.clone()) + .add(client, resources.listener) .await - .context(ApplyServerConfigMapSnafu { - name: validated.name_any(), - })?; - - // ======================================== - // Server listener - let listener = server::build_listener(&validated); - - let applied_listener = cluster_resources - .add(client, listener) - .await - .context(ApplyListenerSnafu)?; - - // ======================================== - // Server stateful set - let args = server::command_args(&scs.spec.args); - let stateful_set = server::build_stateful_set( - &validated, - &service_account, - &server_config_map, - &applied_listener.name_any(), - args, - ) - .context(BuildServerStatefulSetSnafu)?; + .context(ApplyResourceSnafu)?; let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - ss_cond_builder.add( cluster_resources - .add(client, stateful_set) + .add(client, resources.stateful_set) .await - .context(ApplyStatefulSetSnafu)?, + .context(ApplyResourceSnafu)?, ); cluster_resources diff --git a/rust/operator-binary/src/connect/controller/build/mod.rs b/rust/operator-binary/src/connect/controller/build/mod.rs index ff9c47cc..da79b9b9 100644 --- a/rust/operator-binary/src/connect/controller/build/mod.rs +++ b/rust/operator-binary/src/connect/controller/build/mod.rs @@ -8,3 +8,110 @@ pub(crate) mod executor; pub(crate) mod server; pub(crate) mod service; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}, + kube::ResourceExt, +}; + +use crate::connect::{ + common, + controller::{SparkConnectResources, validate::ValidatedSparkConnectServer}, +}; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build connect server S3 properties"))] + S3SparkProperties { source: crate::connect::s3::Error }, + + #[snafu(display("failed to build connect server properties"))] + ServerProperties { source: server::Error }, + + #[snafu(display("failed to build connect executor properties"))] + ExecutorProperties { source: executor::Error }, + + #[snafu(display("failed to serialize connect properties"))] + SerializeProperties { source: common::Error }, + + #[snafu(display("failed to build spark connect executor config map for {name}"))] + BuildExecutorConfigMap { + source: executor::Error, + name: String, + }, + + #[snafu(display("failed to build connect executor pod template"))] + ExecutorPodTemplate { source: executor::Error }, + + #[snafu(display("failed to serialize executor pod template"))] + ExecutorPodTemplateSerde { source: serde_yaml::Error }, + + #[snafu(display("failed to build spark connect server config map for {name}"))] + BuildServerConfigMap { source: server::Error, name: String }, + + #[snafu(display("failed to build spark connect stateful set"))] + BuildServerStatefulSet { source: server::Error }, +} + +/// Builds every Kubernetes resource for the given validated SparkConnectServer. +pub(crate) fn build( + validated: &ValidatedSparkConnectServer, + service_account: ServiceAccount, + role_binding: RoleBinding, + user_args: &[String], +) -> Result { + let resolved_s3 = &validated.cluster_config.resolved_s3; + + // Headless service used by executors to connect back to the driver, plus the metrics service. + let headless_service = service::build_headless_service(validated); + let metrics_service = service::build_metrics_service(validated); + + let spark_props = common::spark_properties(&[ + resolved_s3 + .spark_properties() + .context(S3SparkPropertiesSnafu)?, + server::server_properties(validated, &headless_service, &service_account) + .context(ServerPropertiesSnafu)?, + executor::executor_properties(validated).context(ExecutorPropertiesSnafu)?, + ]) + .context(SerializePropertiesSnafu)?; + + let executor_config_map = + executor::executor_config_map(validated).context(BuildExecutorConfigMapSnafu { + name: validated.name_any(), + })?; + + let executor_pod_template = serde_yaml::to_string( + &executor::executor_pod_template(validated, &executor_config_map) + .context(ExecutorPodTemplateSnafu)?, + ) + .context(ExecutorPodTemplateSerdeSnafu)?; + + let server_config_map = + server::server_config_map(validated, &spark_props, &executor_pod_template).context( + BuildServerConfigMapSnafu { + name: validated.name_any(), + }, + )?; + + let listener = server::build_listener(validated); + + let args = server::command_args(user_args); + let stateful_set = server::build_stateful_set( + validated, + &service_account, + &server_config_map, + &listener.name_any(), + args, + ) + .context(BuildServerStatefulSetSnafu)?; + + Ok(SparkConnectResources { + service_account, + role_binding, + services: vec![headless_service, metrics_service], + config_maps: vec![executor_config_map, server_config_map], + listener, + stateful_set, + }) +} From 397471b4bd67cbe7f1ec0431347d18439c243852 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 18:11:10 +0200 Subject: [PATCH 3/5] refactor: Introduce build aggregator for SparkHistoryServer --- .../operator-binary/src/history/controller.rs | 113 ++++++++---------- .../src/history/controller/build/mod.rs | 57 +++++++++ 2 files changed, 110 insertions(+), 60 deletions(-) diff --git a/rust/operator-binary/src/history/controller.rs b/rust/operator-binary/src/history/controller.rs index efe5f2bb..9ebefc78 100644 --- a/rust/operator-binary/src/history/controller.rs +++ b/rust/operator-binary/src/history/controller.rs @@ -5,6 +5,13 @@ use stackable_operator::{ builder::{self}, cluster_resources::ClusterResourceApplyStrategy, commons::rbac::build_rbac_resources, + crd::listener, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service, ServiceAccount}, + policy::v1::PodDisruptionBudget, + rbac::v1::RoleBinding, + }, kube::{ core::{DeserializeGuard, error_boundary}, runtime::controller::Action, @@ -18,13 +25,9 @@ use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ Ctx, crd::{ - constants::{HISTORY_APP_NAME, HISTORY_ROLE_NAME, JVM_SECURITY_PROPERTIES_FILE}, + constants::{HISTORY_APP_NAME, JVM_SECURITY_PROPERTIES_FILE}, history::v1alpha1, }, - history::controller::build::resource::{ - config_map::build_config_map, listener::build_group_listener, pdb::build_pdb, - service::build_rolegroup_metrics_service, statefulset::build_stateful_set, - }, }; pub mod build; @@ -49,18 +52,8 @@ pub enum Error { name: String, }, - #[snafu(display("failed to update the history server stateful set"))] - ApplyStatefulSet { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update history server config map"))] - ApplyConfigMap { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update history server metrics service"))] - ApplyMetricsService { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, }, @@ -94,11 +87,6 @@ pub enum Error { rolegroup: String, }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to get required Labels"))] GetRequiredLabels { source: @@ -123,11 +111,6 @@ pub enum Error { source: Box, }, - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to serialize Spark default properties"))] InvalidSparkDefaults { source: PropertiesWriterError }, } @@ -137,7 +120,22 @@ impl ReconcilerError for Error { ErrorDiscriminants::from(self).into() } } -/// Updates the status of the SparkApplication that started the pod. + +/// Every Kubernetes resource produced by the build step for a SparkHistoryServer. +/// +/// Built without a Kubernetes client: all references are already dereferenced and validated by +/// this point, so the only errors possible during assembly are resource-construction failures. +pub struct SparkHistoryResources { + pub service_account: ServiceAccount, + pub role_binding: RoleBinding, + /// One ConfigMap, metrics Service and StatefulSet per role group. + pub config_maps: Vec, + pub metrics_services: Vec, + pub stateful_sets: Vec, + pub listener: listener::v1alpha1::Listener, + pub pod_disruption_budget: Option, +} + pub async fn reconcile( shs: Arc>, ctx: Arc, @@ -170,9 +168,9 @@ pub async fn reconcile( &shs.spec.object_overrides, ); - let log_dir = &validated.cluster_config.log_dir; - - // Use a dedicated service account for history server pods. + // Use a dedicated service account for history server pods. Building the RBAC resources needs + // the cluster-resource labels, so it stays in the reconcile step; the built objects (whose + // names are deterministic) are handed to the client-free build step. let (service_account, role_binding) = build_rbac_resources( shs, HISTORY_APP_NAME, @@ -181,52 +179,47 @@ pub async fn reconcile( .context(GetRequiredLabelsSnafu)?, ) .context(BuildRbacResourcesSnafu)?; - let service_account = cluster_resources - .add(client, service_account) + + let resources = build::build(&validated, service_account, role_binding)?; + + // Apply order: ServiceAccount and RoleBinding first, then the ConfigMaps, metrics Services, + // Listener and PodDisruptionBudget, and finally the StatefulSets (they mount the ConfigMaps + // and run under the SA, so those must exist first). + cluster_resources + .add(client, resources.service_account) .await .context(ApplyServiceAccountSnafu)?; cluster_resources - .add(client, role_binding) + .add(client, resources.role_binding) .await .context(ApplyRoleBindingSnafu)?; - - for (role_group_name, rg) in &validated.role_groups { - let config_map = build_config_map(&validated, role_group_name, rg)?; - - let metrics_service = build_rolegroup_metrics_service(&validated, role_group_name); - - let sts = build_stateful_set(&validated, role_group_name, rg, log_dir, &service_account)?; - + for config_map in resources.config_maps { cluster_resources .add(client, config_map) .await - .context(ApplyConfigMapSnafu)?; + .context(ApplyResourceSnafu)?; + } + for metrics_service in resources.metrics_services { cluster_resources .add(client, metrics_service) .await - .context(ApplyMetricsServiceSnafu)?; - cluster_resources - .add(client, sts) - .await - .context(ApplyStatefulSetSnafu)?; + .context(ApplyResourceSnafu)?; } - - let rg_group_listener = build_group_listener( - &validated, - HISTORY_ROLE_NAME, - validated.role_config.listener_class.clone(), - ); - cluster_resources - .add(client, rg_group_listener) + .add(client, resources.listener) .await - .context(ApplyGroupListenerSnafu)?; - - if let Some(pdb) = build_pdb(&validated.role_config.pdb, &validated) { + .context(ApplyResourceSnafu)?; + if let Some(pdb) = resources.pod_disruption_budget { cluster_resources .add(client, pdb) .await - .context(ApplyPdbSnafu)?; + .context(ApplyResourceSnafu)?; + } + for stateful_set in resources.stateful_sets { + cluster_resources + .add(client, stateful_set) + .await + .context(ApplyResourceSnafu)?; } cluster_resources diff --git a/rust/operator-binary/src/history/controller/build/mod.rs b/rust/operator-binary/src/history/controller/build/mod.rs index c6bee053..31b0ed49 100644 --- a/rust/operator-binary/src/history/controller/build/mod.rs +++ b/rust/operator-binary/src/history/controller/build/mod.rs @@ -1 +1,58 @@ pub mod resource; + +use stackable_operator::k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}; + +use crate::{ + crd::constants::HISTORY_ROLE_NAME, + history::controller::{ + Error, SparkHistoryResources, + build::resource::{ + config_map::build_config_map, listener::build_group_listener, pdb::build_pdb, + service::build_rolegroup_metrics_service, statefulset::build_stateful_set, + }, + validate::ValidatedSparkHistoryServer, + }, +}; + +/// Builds every Kubernetes resource for the given validated SparkHistoryServer. +pub fn build( + validated: &ValidatedSparkHistoryServer, + service_account: ServiceAccount, + role_binding: RoleBinding, +) -> Result { + let log_dir = &validated.cluster_config.log_dir; + + let mut config_maps = vec![]; + let mut metrics_services = vec![]; + let mut stateful_sets = vec![]; + + for (role_group_name, rg) in &validated.role_groups { + config_maps.push(build_config_map(validated, role_group_name, rg)?); + metrics_services.push(build_rolegroup_metrics_service(validated, role_group_name)); + stateful_sets.push(build_stateful_set( + validated, + role_group_name, + rg, + log_dir, + &service_account, + )?); + } + + let listener = build_group_listener( + validated, + HISTORY_ROLE_NAME, + validated.role_config.listener_class.clone(), + ); + + let pod_disruption_budget = build_pdb(&validated.role_config.pdb, validated); + + Ok(SparkHistoryResources { + service_account, + role_binding, + config_maps, + metrics_services, + stateful_sets, + listener, + pod_disruption_budget, + }) +} From c2ce137f45502f506d04e6aec2811850644feb78 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Mon, 13 Jul 2026 08:53:18 +0200 Subject: [PATCH 4/5] refactor: consolidate error handling --- .../src/spark_k8s_controller.rs | 52 ++----------------- .../src/spark_k8s_controller/build/mod.rs | 42 +++++++++++---- .../src/spark_k8s_controller/build/pod.rs | 30 +++++++++-- .../build/resource/config_map.rs | 36 +++++++++++-- .../build/resource/job.rs | 23 ++++++-- .../build/resource/serviceaccount.rs | 9 ++-- 6 files changed, 114 insertions(+), 78 deletions(-) diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs index 7b0bf1f0..2489f2e4 100644 --- a/rust/operator-binary/src/spark_k8s_controller.rs +++ b/rust/operator-binary/src/spark_k8s_controller.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::{self}, k8s_openapi::api::{ batch::v1::Job, core::v1::{ConfigMap, ServiceAccount}, @@ -15,13 +14,12 @@ use stackable_operator::{ }, logging::controller::ReconcilerError, shared::time::Duration, - v2::config_file_writer::PropertiesWriterError, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ Ctx, - crd::{constants::*, roles::SparkApplicationRole, v1alpha1}, + crd::{constants::*, v1alpha1}, }; pub mod build; @@ -38,8 +36,8 @@ pub enum Error { #[snafu(display("failed to validate SparkApplication"))] ValidateSparkApplication { source: validate::Error }, - #[snafu(display("missing secret lifetime"))] - MissingSecretLifetime, + #[snafu(display("failed to build SparkApplication resources"))] + BuildSparkApplication { source: build::Error }, #[snafu(display("failed to apply role ServiceAccount"))] ApplyServiceAccount { @@ -56,54 +54,12 @@ pub enum Error { source: stackable_operator::client::Error, }, - #[snafu(display("failed to build stark-submit command"))] - BuildCommand { source: crate::crd::Error }, - - #[snafu(display("failed to build the pod template config map"))] - PodTemplateConfigMap { - source: stackable_operator::builder::configmap::Error, - }, - - #[snafu(display("pod template serialization"))] - PodTemplateSerde { source: serde_yaml::Error }, - - #[snafu(display("failed to resolve and merge config"))] - FailedToResolveConfig { source: crate::crd::Error }, - - #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))] - VectorAggregatorConfigMapMissing, - - #[snafu(display("failed to validate the logging configuration"))] - ValidateLoggingConfig { - source: stackable_operator::v2::product_logging::framework::Error, - }, - - #[snafu(display("failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for {}", role))] - JvmSecurityProperties { - source: PropertiesWriterError, - role: SparkApplicationRole, - }, - - #[snafu(display("invalid submit config"))] - SubmitConfig { source: crate::crd::Error }, - - #[snafu(display("failed to create Volumes for SparkApplication"))] - CreateVolumes { source: crate::crd::Error }, - #[snafu(display("Failed to update status for application {name:?}"))] ApplySparkApplicationStatus { source: stackable_operator::client::Error, name: String, }, - #[snafu(display("failed to add needed volume"))] - AddVolume { source: builder::pod::Error }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: builder::pod::container::Error, - }, - #[snafu(display("SparkApplication object is invalid"))] InvalidSparkApplication { // boxed because otherwise Clippy warns about a large enum variant @@ -168,7 +124,7 @@ pub async fn reconcile( // No more mutating operations after this point (except for status). tracing::debug!("reconciling spark application [{spark_application:?}]"); - let resources = build::build(&validated)?; + let resources = build::build(&validated).context(BuildSparkApplicationSnafu)?; // Apply the ServiceAccount and RoleBinding first, then the ConfigMaps, and finally the Job: // the Job runs under the ServiceAccount and mounts the ConfigMaps, so they must exist first. diff --git a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs index 0167c4f9..9d2e136c 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/mod.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/mod.rs @@ -1,16 +1,34 @@ pub mod pod; pub mod resource; -use snafu::ResultExt; +use resource::{config_map, job}; +use snafu::{ResultExt, Snafu}; use crate::{ crd::roles::SparkApplicationRole, - spark_k8s_controller::{ - BuildCommandSnafu, FailedToResolveConfigSnafu, Result, SparkResources, SubmitConfigSnafu, - validate::ValidatedSparkApplication, - }, + spark_k8s_controller::{SparkResources, validate::ValidatedSparkApplication}, }; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to resolve and merge config"))] + FailedToResolveConfig { source: crate::crd::Error }, + + #[snafu(display("failed to build stark-submit command"))] + BuildCommand { source: crate::crd::Error }, + + #[snafu(display("invalid submit config"))] + SubmitConfig { source: crate::crd::Error }, + + #[snafu(display("failed to build ConfigMap"))] + BuildConfigMap { source: config_map::Error }, + + #[snafu(display("failed to build Job"))] + BuildJob { source: job::Error }, +} + +type Result = std::result::Result; + /// Builds every Kubernetes resource for the given validated SparkApplication. pub fn build(validated: &ValidatedSparkApplication) -> Result { let spark_application = &validated.spark_application; @@ -19,7 +37,7 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { let resolved_product_image = &validated.resolved_product_image; let (service_account, role_binding) = - resource::serviceaccount::build_spark_role_serviceaccount(validated)?; + resource::serviceaccount::build_spark_role_serviceaccount(validated); let env_vars = spark_application.env(opt_s3conn, logdir); @@ -41,7 +59,8 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { &driver_config_overrides, &env_vars, &service_account, - )?; + ) + .context(BuildConfigMapSnafu)?; let executor_config = spark_application .executor_config() @@ -61,7 +80,8 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { &executor_config_overrides, &env_vars, &service_account, - )?; + ) + .context(BuildConfigMapSnafu)?; let job_commands = spark_application .build_command(opt_s3conn, logdir, &resolved_product_image.image) @@ -79,7 +99,8 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { .unwrap_or_default(); let submit_job_config_map = - resource::config_map::submit_job_config_map(validated, &submit_config_overrides)?; + resource::config_map::submit_job_config_map(validated, &submit_config_overrides) + .context(BuildConfigMapSnafu)?; let job = resource::job::spark_job( validated, @@ -87,7 +108,8 @@ pub fn build(validated: &ValidatedSparkApplication) -> Result { &env_vars, &job_commands, &submit_config, - )?; + ) + .context(BuildJobSnafu)?; Ok(SparkResources { service_account, diff --git a/rust/operator-binary/src/spark_k8s_controller/build/pod.rs b/rust/operator-binary/src/spark_k8s_controller/build/pod.rs index 9abf83ad..8208f7a9 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/pod.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/pod.rs @@ -1,6 +1,6 @@ use std::str::FromStr; -use snafu::{OptionExt, ResultExt}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{ meta::ObjectMetaBuilder, @@ -38,12 +38,32 @@ use crate::{ roles::{RoleConfig, SparkApplicationRole, SparkContainer}, tlscerts, }, - spark_k8s_controller::{ - AddVolumeMountSnafu, AddVolumeSnafu, Result, ValidateLoggingConfigSnafu, - VectorAggregatorConfigMapMissingSnafu, validate, - }, + spark_k8s_controller::validate, }; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to add needed volumeMount"))] + AddVolumeMount { + source: stackable_operator::builder::pod::container::Error, + }, + + #[snafu(display("failed to add needed volume"))] + AddVolume { + source: stackable_operator::builder::pod::Error, + }, + + #[snafu(display("failed to validate the logging configuration"))] + ValidateLoggingConfig { + source: stackable_operator::v2::product_logging::framework::Error, + }, + + #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))] + VectorAggregatorConfigMapMissing, +} + +type Result = std::result::Result; + fn init_containers( validated: &validate::ValidatedSparkApplication, logging: &Logging, diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs index 3993fee3..002e85fe 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/config_map.rs @@ -1,4 +1,4 @@ -use snafu::{OptionExt, ResultExt}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{configmap::ConfigMapBuilder, pod::volume::VolumeBuilder}, k8s_openapi::api::core::v1::{ConfigMap, EnvVar, ServiceAccount}, @@ -20,12 +20,39 @@ use crate::{ }, product_logging::{self}, spark_k8s_controller::{ - CreateVolumesSnafu, JvmSecurityPropertiesSnafu, MissingSecretLifetimeSnafu, - PodTemplateConfigMapSnafu, PodTemplateSerdeSnafu, Result, build::pod::pod_template, + build::pod::{self, pod_template}, validate, }, }; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build the pod template"))] + BuildPodTemplate { source: pod::Error }, + + #[snafu(display("pod template serialization"))] + PodTemplateSerde { source: serde_yaml::Error }, + + #[snafu(display("failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for {}", role))] + JvmSecurityProperties { + source: stackable_operator::v2::config_file_writer::PropertiesWriterError, + role: SparkApplicationRole, + }, + + #[snafu(display("failed to build the pod template config map"))] + PodTemplateConfigMap { + source: stackable_operator::builder::configmap::Error, + }, + + #[snafu(display("missing secret lifetime"))] + MissingSecretLifetime, + + #[snafu(display("failed to create Volumes for SparkApplication"))] + CreateVolumes { source: crate::crd::Error }, +} + +type Result = std::result::Result; + pub(crate) fn pod_template_config_map( validated: &validate::ValidatedSparkApplication, role: SparkApplicationRole, @@ -75,7 +102,8 @@ pub(crate) fn pod_template_config_map( volumes.as_ref(), env, service_account, - )?; + ) + .context(BuildPodTemplateSnafu)?; let mut cm_builder = ConfigMapBuilder::new(); diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs index d078df55..617a4666 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/job.rs @@ -1,4 +1,4 @@ -use snafu::{OptionExt, ResultExt}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{meta::ObjectMetaBuilder, pod::volume::VolumeBuilder}, k8s_openapi::{ @@ -17,12 +17,25 @@ use crate::{ roles::{SparkApplicationRole, SparkContainer, SubmitConfig}, tlscerts, }, - spark_k8s_controller::{ - AddVolumeMountSnafu, CreateVolumesSnafu, MissingSecretLifetimeSnafu, Result, - build::pod::security_context, validate, - }, + spark_k8s_controller::{build::pod::security_context, validate}, }; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to add needed volumeMount"))] + AddVolumeMount { + source: stackable_operator::builder::pod::container::Error, + }, + + #[snafu(display("missing secret lifetime"))] + MissingSecretLifetime, + + #[snafu(display("failed to create Volumes for SparkApplication"))] + CreateVolumes { source: crate::crd::Error }, +} + +type Result = std::result::Result; + pub(crate) fn spark_job( validated: &validate::ValidatedSparkApplication, serviceaccount: &ServiceAccount, diff --git a/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs b/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs index 746361d1..40ca4c46 100644 --- a/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs +++ b/rust/operator-binary/src/spark_k8s_controller/build/resource/serviceaccount.rs @@ -6,10 +6,7 @@ use stackable_operator::k8s_openapi::{ }, }; -use crate::{ - crd::constants::*, - spark_k8s_controller::{Result, validate}, -}; +use crate::{crd::constants::*, spark_k8s_controller::validate}; /// For a given SparkApplication, we create a ServiceAccount with a RoleBinding to the ClusterRole /// that allows the driver to create pods etc. @@ -17,7 +14,7 @@ use crate::{ /// They are deleted when the job is deleted. pub(crate) fn build_spark_role_serviceaccount( validated: &validate::ValidatedSparkApplication, -) -> Result<(ServiceAccount, RoleBinding)> { +) -> (ServiceAccount, RoleBinding) { let sa_name = validated.name.to_string(); let sa = ServiceAccount { metadata: validated.object_meta(&sa_name, "service-account").build(), @@ -38,5 +35,5 @@ pub(crate) fn build_spark_role_serviceaccount( namespace: sa.metadata.namespace.clone(), }]), }; - Ok((sa, binding)) + (sa, binding) } From 8a5057db26a7b10a41d41271e656e0a9be565f3b Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Mon, 13 Jul 2026 09:15:49 +0200 Subject: [PATCH 5/5] refactor: consolidate error handling --- .../operator-binary/src/history/controller.rs | 44 +++---------------- .../src/history/controller/build/mod.rs | 37 +++++++++++----- .../controller/build/resource/config_map.rs | 35 ++++++++++++--- .../controller/build/resource/statefulset.rs | 28 +++++++++--- 4 files changed, 82 insertions(+), 62 deletions(-) diff --git a/rust/operator-binary/src/history/controller.rs b/rust/operator-binary/src/history/controller.rs index 9ebefc78..250e8d30 100644 --- a/rust/operator-binary/src/history/controller.rs +++ b/rust/operator-binary/src/history/controller.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::{self}, cluster_resources::ClusterResourceApplyStrategy, commons::rbac::build_rbac_resources, crd::listener, @@ -18,16 +17,13 @@ use stackable_operator::{ }, logging::controller::ReconcilerError, shared::time::Duration, - v2::{cluster_resources::cluster_resources_new, config_file_writer::PropertiesWriterError}, + v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ Ctx, - crd::{ - constants::{HISTORY_APP_NAME, JVM_SECURITY_PROPERTIES_FILE}, - history::v1alpha1, - }, + crd::{constants::HISTORY_APP_NAME, history::v1alpha1}, }; pub mod build; @@ -43,14 +39,8 @@ pub enum Error { source: stackable_operator::commons::rbac::Error, }, - #[snafu(display("missing secret lifetime"))] - MissingSecretLifetime, - - #[snafu(display("invalid config map {name}"))] - InvalidConfigMap { - source: stackable_operator::builder::configmap::Error, - name: String, - }, + #[snafu(display("failed to build SparkHistoryServer resources"))] + BuildSparkHistoryServer { source: build::Error }, #[snafu(display("failed to apply Kubernetes resource"))] ApplyResource { @@ -78,41 +68,18 @@ pub enum Error { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display( - "History server : failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for group {}", - rolegroup - ))] - JvmSecurityProperties { - source: PropertiesWriterError, - rolegroup: String, - }, - #[snafu(display("failed to get required Labels"))] GetRequiredLabels { source: stackable_operator::kvp::KeyValuePairError, }, - #[snafu(display("failed to create the log dir volumes specification"))] - CreateLogDirVolumesSpec { source: crate::crd::logdir::Error }, - - #[snafu(display("failed to add needed volume"))] - AddVolume { source: builder::pod::Error }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: builder::pod::container::Error, - }, - #[snafu(display("SparkHistoryServer object is invalid"))] InvalidSparkHistoryServer { // boxed because otherwise Clippy warns about a large enum variant #[snafu(source(from(error_boundary::InvalidObject, Box::new)))] source: Box, }, - - #[snafu(display("failed to serialize Spark default properties"))] - InvalidSparkDefaults { source: PropertiesWriterError }, } impl ReconcilerError for Error { @@ -180,7 +147,8 @@ pub async fn reconcile( ) .context(BuildRbacResourcesSnafu)?; - let resources = build::build(&validated, service_account, role_binding)?; + let resources = build::build(&validated, service_account, role_binding) + .context(BuildSparkHistoryServerSnafu)?; // Apply order: ServiceAccount and RoleBinding first, then the ConfigMaps, metrics Services, // Listener and PodDisruptionBudget, and finally the StatefulSets (they mount the ConfigMaps diff --git a/rust/operator-binary/src/history/controller/build/mod.rs b/rust/operator-binary/src/history/controller/build/mod.rs index 31b0ed49..e23d2609 100644 --- a/rust/operator-binary/src/history/controller/build/mod.rs +++ b/rust/operator-binary/src/history/controller/build/mod.rs @@ -1,25 +1,40 @@ pub mod resource; +use snafu::{ResultExt, Snafu}; use stackable_operator::k8s_openapi::api::{core::v1::ServiceAccount, rbac::v1::RoleBinding}; use crate::{ crd::constants::HISTORY_ROLE_NAME, history::controller::{ - Error, SparkHistoryResources, + SparkHistoryResources, build::resource::{ - config_map::build_config_map, listener::build_group_listener, pdb::build_pdb, - service::build_rolegroup_metrics_service, statefulset::build_stateful_set, + config_map::{self, build_config_map}, + listener::build_group_listener, + pdb::build_pdb, + service::build_rolegroup_metrics_service, + statefulset::{self, build_stateful_set}, }, validate::ValidatedSparkHistoryServer, }, }; +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build ConfigMap"))] + BuildConfigMap { source: config_map::Error }, + + #[snafu(display("failed to build StatefulSet"))] + BuildStatefulSet { source: statefulset::Error }, +} + +type Result = std::result::Result; + /// Builds every Kubernetes resource for the given validated SparkHistoryServer. pub fn build( validated: &ValidatedSparkHistoryServer, service_account: ServiceAccount, role_binding: RoleBinding, -) -> Result { +) -> Result { let log_dir = &validated.cluster_config.log_dir; let mut config_maps = vec![]; @@ -27,15 +42,13 @@ pub fn build( let mut stateful_sets = vec![]; for (role_group_name, rg) in &validated.role_groups { - config_maps.push(build_config_map(validated, role_group_name, rg)?); + config_maps + .push(build_config_map(validated, role_group_name, rg).context(BuildConfigMapSnafu)?); metrics_services.push(build_rolegroup_metrics_service(validated, role_group_name)); - stateful_sets.push(build_stateful_set( - validated, - role_group_name, - rg, - log_dir, - &service_account, - )?); + stateful_sets.push( + build_stateful_set(validated, role_group_name, rg, log_dir, &service_account) + .context(BuildStatefulSetSnafu)?, + ); } let listener = build_group_listener( diff --git a/rust/operator-binary/src/history/controller/build/resource/config_map.rs b/rust/operator-binary/src/history/controller/build/resource/config_map.rs index ea89bd2e..c19f18b2 100644 --- a/rust/operator-binary/src/history/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/history/controller/build/resource/config_map.rs @@ -1,6 +1,6 @@ use std::collections::BTreeMap; -use snafu::ResultExt; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder}, k8s_openapi::api::core::v1::ConfigMap, @@ -20,19 +20,40 @@ use crate::{ history::SparkHistoryServerContainer, to_spark_env_sh_string, }, - history::controller::{ - Error, InvalidConfigMapSnafu, InvalidSparkDefaultsSnafu, JvmSecurityPropertiesSnafu, - validate::{self, ValidatedHistoryRoleGroup}, - }, + history::controller::validate::{self, ValidatedHistoryRoleGroup}, product_logging::{self}, }; -#[allow(clippy::result_large_err)] +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("invalid config map {name}"))] + InvalidConfigMap { + source: stackable_operator::builder::configmap::Error, + name: String, + }, + + #[snafu(display( + "History server : failed to serialize [{JVM_SECURITY_PROPERTIES_FILE}] for group {}", + rolegroup + ))] + JvmSecurityProperties { + source: stackable_operator::v2::config_file_writer::PropertiesWriterError, + rolegroup: String, + }, + + #[snafu(display("failed to serialize Spark default properties"))] + InvalidSparkDefaults { + source: stackable_operator::v2::config_file_writer::PropertiesWriterError, + }, +} + +type Result = std::result::Result; + pub(crate) fn build_config_map( validated: &validate::ValidatedSparkHistoryServer, role_group_name: &RoleGroupName, rg: &ValidatedHistoryRoleGroup, -) -> Result { +) -> Result { let cm_name = validated .resource_names(role_group_name) .role_group_config_map() diff --git a/rust/operator-binary/src/history/controller/build/resource/statefulset.rs b/rust/operator-binary/src/history/controller/build/resource/statefulset.rs index 079a7d7b..8ef1f5b2 100644 --- a/rust/operator-binary/src/history/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/history/controller/build/resource/statefulset.rs @@ -1,6 +1,6 @@ use std::str::FromStr; -use snafu::{OptionExt, ResultExt}; +use snafu::{OptionExt, ResultExt, Snafu}; use stackable_operator::{ builder::{ meta::ObjectMetaBuilder, @@ -52,22 +52,40 @@ use crate::{ history::{ config::jvm::construct_history_jvm_args, controller::{ - AddVolumeMountSnafu, AddVolumeSnafu, CreateLogDirVolumesSpecSnafu, Error, - MissingSecretLifetimeSnafu, build::resource::listener::group_listener_name, validate::{self, ValidatedHistoryRoleGroup}, }, }, }; -#[allow(clippy::result_large_err)] +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("missing secret lifetime"))] + MissingSecretLifetime, + + #[snafu(display("failed to create the log dir volumes specification"))] + CreateLogDirVolumesSpec { source: crate::crd::logdir::Error }, + + #[snafu(display("failed to add needed volume"))] + AddVolume { + source: stackable_operator::builder::pod::Error, + }, + + #[snafu(display("failed to add needed volumeMount"))] + AddVolumeMount { + source: stackable_operator::builder::pod::container::Error, + }, +} + +type Result = std::result::Result; + pub(crate) fn build_stateful_set( validated: &validate::ValidatedSparkHistoryServer, role_group_name: &RoleGroupName, rg: &ValidatedHistoryRoleGroup, log_dir: &ResolvedLogDir, serviceaccount: &ServiceAccount, -) -> Result { +) -> Result { let resolved_product_image = &validated.resolved_product_image; let resource_names = validated.resource_names(role_group_name);