diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c324d46..e4839503 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,7 @@ - Environment variable overrides (`envOverrides`) are now applied after all environment variables set by the operator. In particular, `CONTAINERDEBUG_LOG_DIRECTORY` can now be overridden, whereas previously the operator's value always took precedence ([#838]). +- Make operations infallible where appropriate ([#852]). ### Fixed @@ -58,6 +59,7 @@ [#844]: https://github.com/stackabletech/airflow-operator/pull/844 [#847]: https://github.com/stackabletech/airflow-operator/pull/847 [#849]: https://github.com/stackabletech/airflow-operator/pull/849 +[#852]: https://github.com/stackabletech/airflow-operator/pull/852 ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index af374e3b..e8b3410e 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -693,4 +693,49 @@ mod tests { assert_eq!(env.get("AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX"), None); assert_eq!(env.get("AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR"), None); } + + /// A cluster with the given `spec.clusterConfig.volumeMounts` (as standalone YAML). + fn cluster_with_user_volume_mounts( + executor_key: &str, + executor_config: &str, + volume_mounts: &str, + ) -> ValidatedCluster { + validated_cluster_with(executor_key, executor_config, |cluster| { + cluster["spec"]["clusterConfig"] + .as_mapping_mut() + .expect("clusterConfig is a mapping") + .insert( + "volumeMounts".into(), + serde_yaml::from_str(volume_mounts).expect("valid volumeMounts YAML"), + ); + }) + } + + /// A user-supplied volumeMount that collides with an operator-managed mount path must be + /// reported as an error (the operator's own mounts are added first and are infallible). + #[test] + fn user_volume_mount_colliding_with_config_path_is_an_error() { + let cluster = cluster_with_user_volume_mounts( + "kubernetesExecutors", + "{config: {}}", + "[{name: user-volume, mountPath: /stackable/app/config}]", + ); + + let Err(error) = build(&cluster) else { + panic!("the colliding mount must be rejected"); + }; + assert!( + matches!( + error, + super::Error::StatefulSet { + source: + crate::controller::build::resource::statefulset::Error::AddVolumeMount { .. }, + .. + } | super::Error::ExecutorTemplate { + source: crate::controller::build::resource::executor::Error::AddVolumeMount { .. }, + } + ), + "unexpected error: {error:?}" + ); + } } diff --git a/rust/operator-binary/src/controller/build/properties/env_vars.rs b/rust/operator-binary/src/controller/build/properties/env_vars.rs index ca5e8f63..02361145 100644 --- a/rust/operator-binary/src/controller/build/properties/env_vars.rs +++ b/rust/operator-binary/src/controller/build/properties/env_vars.rs @@ -1,6 +1,5 @@ -use std::{collections::BTreeSet, path::PathBuf, str::FromStr}; +use std::{collections::BTreeSet, str::FromStr}; -use snafu::Snafu; use stackable_operator::{ constant, crd::{authentication::oidc, git_sync}, @@ -81,14 +80,6 @@ constant!(PYTHONPATH: EnvVarName = "PYTHONPATH"); constant!(CONTAINERDEBUG_LOG_DIRECTORY: EnvVarName = "CONTAINERDEBUG_LOG_DIRECTORY"); constant!(STACKABLE_POST_HOOK: EnvVarName = "_STACKABLE_POST_HOOK"); -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display( - "failed to construct Git DAG folder - Is the git folder a valid path?: {dag_folder:?}" - ))] - ConstructGitDagFolder { dag_folder: PathBuf }, -} - /// Return environment variables to be applied to the statefulsets for the scheduler, webserver (and worker, /// for clusters utilizing `celeryExecutor`: for clusters using `kubernetesExecutor` a different set will be /// used which is defined in [`build_airflow_template_envs`]). diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index fc0d15c4..061458c5 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -37,12 +37,6 @@ pub enum Error { source: webserver_config::Error, role_group: RoleGroupName, }, - - #[snafu(display("failed to build ConfigMap for role group {role_group}"))] - BuildConfigMap { - source: stackable_operator::builder::configmap::Error, - role_group: RoleGroupName, - }, } /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator @@ -98,7 +92,7 @@ pub fn build_rolegroup_config_map( cm_builder.add_data(VECTOR_CONFIG_FILE, vector_config_file_content()); } - cm_builder.build().with_context(|_| BuildConfigMapSnafu { - role_group: role_group_name.clone(), - }) + Ok(cm_builder + .build() + .expect("The ConfigMap metadata is set in this function.")) } diff --git a/rust/operator-binary/src/controller/build/resource/executor.rs b/rust/operator-binary/src/controller/build/resource/executor.rs index 34fd8535..54f1f7ac 100644 --- a/rust/operator-binary/src/controller/build/resource/executor.rs +++ b/rust/operator-binary/src/controller/build/resource/executor.rs @@ -8,11 +8,11 @@ use stackable_operator::{ meta::ObjectMetaBuilder, pod::{PodBuilder, security::PodSecurityContextBuilder}, }, + constants::RESTART_CONTROLLER_ENABLED_LABEL, k8s_openapi::{ DeepMerge, api::core::v1::{ConfigMap, PodTemplateSpec}, }, - kvp::{Label, LabelError}, v2::{ builder::pod::container::{EnvVarSet, new_container_builder}, product_logging::framework::STACKABLE_LOG_DIR, @@ -55,17 +55,9 @@ pub enum Error { source: stackable_operator::builder::pod::container::Error, }, - #[snafu(display("failed to build label"))] - BuildLabel { source: LabelError }, - #[snafu(display("pod template serialization"))] PodTemplateSerde { source: serde_yaml::Error }, - #[snafu(display("failed to build the pod template config map"))] - PodTemplateConfigMap { - source: stackable_operator::builder::configmap::Error, - }, - #[snafu(display("failed to build shared pod resources"))] Pod { source: crate::controller::build::resource::pod::Error, @@ -133,13 +125,16 @@ pub fn build_executor_template_config_map( &executor_config.logging, git_sync_resources, )) - .add_volume_mounts(cluster.volume_mounts()) - .context(AddVolumeMountSnafu)? + // Operator-managed mounts first: their names and paths are constants, so they cannot + // collide with each other. .add_volume_mount(&*CONFIG_VOLUME_NAME, CONFIG_PATH) - .context(AddVolumeMountSnafu)? + .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount(&*LOG_CONFIG_VOLUME_NAME, LOG_CONFIG_DIR) - .context(AddVolumeMountSnafu)? + .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount(&*LOG_VOLUME_NAME, STACKABLE_LOG_DIR) + .expect("The mount paths are statically defined and there should be no duplicates.") + // User-supplied mounts last: these can collide with the ones above, so this stays fallible. + .add_volume_mounts(cluster.volume_mounts()) .context(AddVolumeMountSnafu)?; add_git_sync_resources( @@ -157,8 +152,7 @@ pub fn build_executor_template_config_map( .add_to_container(&mut airflow_container); pb.add_container(airflow_container.build()); - pb.add_volumes(cluster.volumes().clone()) - .context(AddVolumeSnafu)?; + // Operator-managed volumes first (static names), user-supplied volumes last (fallible). pb.add_volumes(volumes::create_volumes( cluster .role_group_resource_names(&EXECUTOR_ROLE_NAME, &EXECUTOR_ROLE_GROUP_NAME) @@ -166,7 +160,9 @@ pub fn build_executor_template_config_map( .as_ref(), &executor_config.logging.product_container, )) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); + pb.add_volumes(cluster.volumes().clone()) + .context(AddVolumeSnafu)?; if let Some(vector_log_config) = &executor_config.logging.vector_container { pb.add_container(build_logging_container( @@ -182,9 +178,6 @@ pub fn build_executor_template_config_map( let mut cm_builder = ConfigMapBuilder::new(); - let restarter_label = - Label::try_from(("restarter.stackable.tech/enabled", "true")).context(BuildLabelSnafu)?; - cm_builder .metadata( object_meta( @@ -196,7 +189,7 @@ pub fn build_executor_template_config_map( &EXECUTOR_TEMPLATE_ROLE_GROUP_NAME, ), ) - .with_label(restarter_label) + .with_label(RESTART_CONTROLLER_ENABLED_LABEL.clone()) .build(), ) .add_data( @@ -204,5 +197,7 @@ pub fn build_executor_template_config_map( serde_yaml::to_string(&pod_template).context(PodTemplateSerdeSnafu)?, ); - cm_builder.build().context(PodTemplateConfigMapSnafu) + Ok(cm_builder + .build() + .expect("The ConfigMap metadata is set in this function.")) } diff --git a/rust/operator-binary/src/controller/build/resource/pod.rs b/rust/operator-binary/src/controller/build/resource/pod.rs index 7c48e1a4..7a7930b7 100644 --- a/rust/operator-binary/src/controller/build/resource/pod.rs +++ b/rust/operator-binary/src/controller/build/resource/pod.rs @@ -148,3 +148,14 @@ pub(crate) fn build_logging_container( EnvVarSet::new(), ) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_constants() { + // Test that dereferencing the constants does not panic. + let _ = *VECTOR_CONTAINER_NAME; + } +} diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 700e3a70..8afbe37a 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -7,6 +7,7 @@ use stackable_operator::{ security::PodSecurityContextBuilder, volume::VolumeBuilder, }, }, + constants::RESTART_CONTROLLER_ENABLED_LABEL, k8s_openapi::{ DeepMerge, api::{ @@ -16,7 +17,7 @@ use stackable_operator::{ apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, }, kube::api::ObjectMeta, - kvp::{Annotation, Label, LabelError}, + kvp::Annotation, utils::COMMON_BASH_TRAP_FUNCTIONS, v2::{ builder::pod::{ @@ -63,9 +64,6 @@ pub enum Error { source: crate::controller::build::graceful_shutdown::Error, }, - #[snafu(display("failed to build label"))] - BuildLabel { source: LabelError }, - #[snafu(display("failed to add needed volume"))] AddVolume { source: stackable_operator::builder::pod::Error, @@ -88,7 +86,6 @@ fn build_rolegroup_metadata( cluster: &ValidatedCluster, role: &AirflowRole, role_group_name: &RoleGroupName, - prometheus_label: Label, name: String, ) -> ObjectMeta { object_meta( @@ -96,7 +93,7 @@ fn build_rolegroup_metadata( name, recommended_labels_for_role_group_resources(cluster, role, role_group_name), ) - .with_label(prometheus_label) + .with_label(RESTART_CONTROLLER_ENABLED_LABEL.clone()) .build() } @@ -194,24 +191,20 @@ pub fn build_server_rolegroup_statefulset( git_sync_resources, )); - let volume_mounts = validated_cluster.volume_mounts(); - airflow_container - .add_volume_mounts(volume_mounts) - .context(AddVolumeMountSnafu)?; + // Operator-managed mounts first: their names and paths are constants, so they cannot collide + // with each other. User-supplied mounts are added last (below) and stay fallible. airflow_container .add_volume_mount(&*CONFIG_VOLUME_NAME, CONFIG_PATH) - .context(AddVolumeMountSnafu)?; - airflow_container + .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount(&*LOG_CONFIG_VOLUME_NAME, LOG_CONFIG_DIR) - .context(AddVolumeMountSnafu)?; - airflow_container + .expect("The mount paths are statically defined and there should be no duplicates.") .add_volume_mount(&*LOG_VOLUME_NAME, STACKABLE_LOG_DIR) - .context(AddVolumeMountSnafu)?; + .expect("The mount paths are statically defined and there should be no duplicates."); if let AirflowExecutor::KubernetesExecutors { .. } = executor { airflow_container .add_volume_mount(&*TEMPLATE_VOLUME_NAME, TEMPLATE_LOCATION) - .context(AddVolumeMountSnafu)?; + .expect("The mount paths are statically defined and there should be no duplicates."); } // for roles with an http endpoint @@ -252,9 +245,14 @@ pub fn build_server_rolegroup_statefulset( airflow_container .add_volume_mount(&*LISTENER_PVC_NAME, LISTENER_VOLUME_DIR) - .context(AddVolumeMountSnafu)?; + .expect("The mount paths are statically defined and there should be no duplicates."); } + // User-supplied mounts can collide with the operator-managed ones above, so this stays fallible. + airflow_container + .add_volume_mounts(validated_cluster.volume_mounts()) + .context(AddVolumeMountSnafu)?; + add_git_sync_resources( &mut pb, &mut airflow_container, @@ -306,13 +304,12 @@ pub fn build_server_rolegroup_statefulset( .build(); pb.add_container(metrics_container); - pb.add_volumes(validated_cluster.volumes().clone()) - .context(AddVolumeSnafu)?; + // Operator-managed volumes first (static names), user-supplied volumes last (fallible). pb.add_volumes(volumes::create_volumes( resource_names.role_group_config_map().as_ref(), &logging.product_container, )) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); if let AirflowExecutor::KubernetesExecutors { .. } = executor { pb.add_volume( @@ -320,9 +317,12 @@ pub fn build_server_rolegroup_statefulset( .with_config_map(validated_cluster.executor_template_configmap_name()) .build(), ) - .context(AddVolumeSnafu)?; + .expect("The volume names are statically defined and there should be no duplicates."); } + pb.add_volumes(validated_cluster.volumes().clone()) + .context(AddVolumeSnafu)?; + if let Some(vector_log_config) = &logging.vector_container { pb.add_container(build_logging_container( resolved_product_image, @@ -333,14 +333,10 @@ pub fn build_server_rolegroup_statefulset( let mut pod_template = pb.build_template(); pod_template.merge_from(validated_rg_config.pod_overrides.clone()); - let restarter_label = - Label::try_from(("restarter.stackable.tech/enabled", "true")).context(BuildLabelSnafu)?; - let metadata = build_rolegroup_metadata( validated_cluster, airflow_role, role_group_name, - restarter_label, resource_names.stateful_set_name().to_string(), ); diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index fb92a5a4..17448784 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -262,25 +262,56 @@ impl ValidatedCluster { /// The Secret holding the shared internal secret (`-internal-secret`). pub fn internal_secret_name(&self) -> SecretName { - SecretName::from_str(&format!("{}-internal-secret", self.name_any())) + const SUFFIX: &str = "-internal-secret"; + const _: () = assert!( + ClusterName::MAX_LENGTH + SUFFIX.len() <= SecretName::MAX_LENGTH, + "The string `-internal-secret` must not exceed the limit of Secret names." + ); + // A ClusterName is an RFC 1035 label, so appending an alphanumeric-terminated suffix keeps + // it a valid RFC 1123 subdomain. + let _ = ClusterName::IS_RFC_1123_SUBDOMAIN_NAME; + + SecretName::from_str(&format!("{}{SUFFIX}", self.name)) .expect("the internal secret name is a valid Secret name") } /// The Secret holding the shared JWT secret (`-jwt-secret`). pub fn jwt_secret_name(&self) -> SecretName { - SecretName::from_str(&format!("{}-jwt-secret", self.name_any())) + const SUFFIX: &str = "-jwt-secret"; + const _: () = assert!( + ClusterName::MAX_LENGTH + SUFFIX.len() <= SecretName::MAX_LENGTH, + "The string `-jwt-secret` must not exceed the limit of Secret names." + ); + let _ = ClusterName::IS_RFC_1123_SUBDOMAIN_NAME; + + SecretName::from_str(&format!("{}{SUFFIX}", self.name)) .expect("the JWT secret name is a valid Secret name") } /// The Secret holding the shared Fernet key (`-fernet-key`). pub fn fernet_key_name(&self) -> SecretName { - SecretName::from_str(&format!("{}-fernet-key", self.name_any())) + const SUFFIX: &str = "-fernet-key"; + const _: () = assert!( + ClusterName::MAX_LENGTH + SUFFIX.len() <= SecretName::MAX_LENGTH, + "The string `-fernet-key` must not exceed the limit of Secret names." + ); + let _ = ClusterName::IS_RFC_1123_SUBDOMAIN_NAME; + + SecretName::from_str(&format!("{}{SUFFIX}", self.name)) .expect("the Fernet key secret name is a valid Secret name") } /// The ConfigMap holding the Kubernetes-executor pod template (`-executor-pod-template`). pub fn executor_template_configmap_name(&self) -> ConfigMapName { - ConfigMapName::from_str(&format!("{}-executor-pod-template", self.name_any())) + const SUFFIX: &str = "-executor-pod-template"; + const _: () = assert!( + ClusterName::MAX_LENGTH + SUFFIX.len() <= ConfigMapName::MAX_LENGTH, + "The string `-executor-pod-template` must not exceed the limit of \ + ConfigMap names." + ); + let _ = ClusterName::IS_RFC_1123_SUBDOMAIN_NAME; + + ConfigMapName::from_str(&format!("{}{SUFFIX}", self.name)) .expect("the executor pod-template ConfigMap name is a valid ConfigMap name") } diff --git a/rust/operator-binary/src/crd/authentication.rs b/rust/operator-binary/src/crd/authentication.rs index b5d3e455..d77f147e 100644 --- a/rust/operator-binary/src/crd/authentication.rs +++ b/rust/operator-binary/src/crd/authentication.rs @@ -27,9 +27,6 @@ pub enum Error { AuthenticationClassRetrievalFailed { source: stackable_operator::client::Error, }, - // TODO: Adapt message if multiple authentication classes are supported simultaneously - #[snafu(display("Only one authentication class is currently supported at a time"))] - MultipleAuthenticationClassesProvided, #[snafu(display( "Failed to use authentication provider [{provider}] for authentication class [{auth_class_name}] - supported providers: {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}", ))] @@ -280,8 +277,8 @@ impl AirflowClientAuthenticationDetailsResolved { None => { info!( "No OIDC provider hint given in AuthClass {auth_class_name}, assuming {default_oidc_provider_name}", - default_oidc_provider_name = - serde_json::to_string(&DEFAULT_OIDC_PROVIDER).unwrap() + default_oidc_provider_name = serde_json::to_string(&DEFAULT_OIDC_PROVIDER) + .expect("an IdentityProviderHint serialises to a plain JSON string") ); DEFAULT_OIDC_PROVIDER } @@ -292,7 +289,8 @@ impl AirflowClientAuthenticationDetailsResolved { SUPPORTED_OIDC_PROVIDERS.contains(&oidc_provider), OidcProviderNotSupportedSnafu { auth_class_name, - oidc_provider: serde_json::to_string(&oidc_provider).unwrap(), + oidc_provider: serde_json::to_string(&oidc_provider) + .expect("an IdentityProviderHint serialises to a plain JSON string"), } );