feat: fault tolerant execution

Author: dervoeti

rust/operator-binary/src/command.rs

@@ -15,13 +15,15 @@ use crate::{
         STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR,
         STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD,
         SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD, TrinoRole, v1alpha1,
+        fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig
     },
 };
 
 pub fn container_prepare_args(
     trino: &v1alpha1::TrinoCluster,
     catalogs: &[CatalogConfig],
     merged_config: &v1alpha1::TrinoConfig,
+    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
 ) -> Vec<String> {
     let mut args = vec![];
 
@@ -78,12 +80,18 @@ pub fn container_prepare_args(
         args.extend_from_slice(&catalog.init_container_extra_start_commands);
     });
 
+    // Add the commands that are needed for fault tolerant execution (e.g., TLS certificates for S3)
+    if let Some(resolved_fte) = resolved_fte_config {
+        args.extend_from_slice(&resolved_fte.init_container_extra_start_commands);
+    }
+
     args
 }
 
 pub fn container_trino_args(
     authentication_config: &TrinoAuthenticationConfig,
     catalogs: &[CatalogConfig],
+    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
 ) -> Vec<String> {
     let mut args = vec![
         // copy config files to a writeable empty folder
@@ -110,6 +118,14 @@ pub fn container_trino_args(
             args.push(format!("export {env_name}=\"$(cat {file})\""));
         }
     });
+
+    // Add fault tolerant execution environment variables from files
+    if let Some(resolved_fte) = resolved_fte_config {
+        for (env_name, file) in &resolved_fte.load_env_from_files {
+            args.push(format!("export {env_name}=\"$(cat {file})\""));
+        }
+    }
+
     args.push("set -x".to_string());
 
     // Start command

rust/operator-binary/src/controller.rs

@@ -78,14 +78,15 @@ use crate::{
     command, config,
     crd::{
         ACCESS_CONTROL_PROPERTIES, APP_NAME, CONFIG_DIR_NAME, CONFIG_PROPERTIES, Container,
-        DISCOVERY_URI, ENV_INTERNAL_SECRET, HTTP_PORT, HTTP_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME,
+        DISCOVERY_URI, ENV_INTERNAL_SECRET, EXCHANGE_MANAGER_PROPERTIES, HTTP_PORT, HTTP_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME,
         JVM_CONFIG, JVM_SECURITY_PROPERTIES, LOG_PROPERTIES, MAX_TRINO_LOG_FILES_SIZE,
         METRICS_PORT, METRICS_PORT_NAME, NODE_PROPERTIES, RW_CONFIG_DIR_NAME,
         STACKABLE_CLIENT_TLS_DIR, STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR,
         STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, TrinoRole,
         authentication::resolve_authentication_classes,
         catalog,
         discovery::{TrinoDiscovery, TrinoDiscoveryProtocol, TrinoPodRef},
+        fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig,
         rolegroup_headless_service_name, v1alpha1,
     },
     listener::{
@@ -298,6 +299,11 @@ pub enum Error {
         source: crate::operations::graceful_shutdown::Error,
     },
 
+    #[snafu(display("failed to configure fault tolerant execution"))]
+    FaultTolerantExecution {
+        source: crate::crd::fault_tolerant_execution::Error,
+    },
+
     #[snafu(display("failed to get required Labels"))]
     GetRequiredLabels {
         source:
@@ -424,6 +430,20 @@ pub async fn reconcile_trino(
         catalogs.push(catalog_config);
     }
 
+    // Resolve fault tolerant execution configuration with S3 connections if needed
+    let resolved_fte_config = match trino.spec.cluster_config.fault_tolerant_execution.as_ref() {
+        Some(fte_config) => Some(
+            ResolvedFaultTolerantExecutionConfig::from_config(
+                fte_config,
+                Some(client),
+                &trino.namespace_r().context(ReadRoleSnafu)?,
+            )
+            .await
+            .context(FaultTolerantExecutionSnafu)?
+        ),
+        None => None,
+    };
+
     let validated_config = validated_product_config(
         trino,
         // The Trino version is a single number like 396.
@@ -526,6 +546,7 @@ pub async fn reconcile_trino(
                 &trino_authentication_config,
                 &trino_opa_config,
                 &client.kubernetes_cluster_info,
+                &resolved_fte_config,
             )?;
             let rg_catalog_configmap = build_rolegroup_catalog_config_map(
                 trino,
@@ -543,6 +564,7 @@ pub async fn reconcile_trino(
                 &trino_authentication_config,
                 &catalogs,
                 &rbac_sa.name_any(),
+                &resolved_fte_config,
             )?;
 
             cluster_resources
@@ -651,6 +673,7 @@ fn build_rolegroup_config_map(
     trino_authentication_config: &TrinoAuthenticationConfig,
     trino_opa_config: &Option<TrinoOpaConfig>,
     cluster_info: &KubernetesClusterInfo,
+    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
 ) -> Result<ConfigMap> {
     let mut cm_conf_data = BTreeMap::new();
 
@@ -712,6 +735,14 @@ fn build_rolegroup_config_map(
                 dynamic_resolved_config
                     .extend(graceful_shutdown_config_properties(trino, trino_role));
 
+                // Add fault tolerant execution properties from resolved configuration
+                if let Some(resolved_fte) = resolved_fte_config {
+                    dynamic_resolved_config.extend(
+                        resolved_fte.config_properties.iter()
+                            .map(|(k, v)| (k.clone(), Some(v.clone())))
+                    );
+                }
+
                 // Add static properties and overrides
                 dynamic_resolved_config.extend(transformed_config);
 
@@ -776,6 +807,20 @@ fn build_rolegroup_config_map(
 
     cm_conf_data.insert(JVM_CONFIG.to_string(), jvm_config.to_string());
 
+    // Add exchange manager properties from resolved fault tolerant execution configuration
+    if let Some(resolved_fte) = resolved_fte_config {
+        if !resolved_fte.exchange_manager_properties.is_empty() {
+            let exchange_props_with_options: BTreeMap<String, Option<String>> =
+                resolved_fte.exchange_manager_properties.iter().map(|(k, v)| (k.clone(), Some(v.clone()))).collect();
+            cm_conf_data.insert(
+                EXCHANGE_MANAGER_PROPERTIES.to_string(),
+                to_java_properties_string(
+                    exchange_props_with_options.iter()
+                ).with_context(|_| FailedToWriteJavaPropertiesSnafu)?,
+            );
+        }
+    }
+
     let jvm_sec_props: BTreeMap<String, Option<String>> = config
         .get(&PropertyNameKind::File(JVM_SECURITY_PROPERTIES.to_string()))
         .cloned()
@@ -884,6 +929,7 @@ fn build_rolegroup_statefulset(
     trino_authentication_config: &TrinoAuthenticationConfig,
     catalogs: &[CatalogConfig],
     sa_name: &str,
+    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
 ) -> Result<StatefulSet> {
     let role = trino
         .role(trino_role)
@@ -974,6 +1020,7 @@ fn build_rolegroup_statefulset(
         &mut cb_trino,
         catalogs,
         &requested_secret_lifetime,
+        resolved_fte_config,
     )?;
 
     let mut prepare_args = vec![];
@@ -992,6 +1039,7 @@ fn build_rolegroup_statefulset(
         trino,
         catalogs,
         merged_config,
+        resolved_fte_config,
     ));
 
     prepare_args
@@ -1056,7 +1104,7 @@ fn build_rolegroup_statefulset(
             "-c".to_string(),
         ])
         .args(vec![
-            command::container_trino_args(trino_authentication_config, catalogs).join("\n"),
+            command::container_trino_args(trino_authentication_config, catalogs, resolved_fte_config).join("\n"),
         ])
         .add_env_vars(env)
         .add_volume_mount("config", CONFIG_DIR_NAME)
@@ -1532,6 +1580,7 @@ fn tls_volume_mounts(
     cb_trino: &mut ContainerBuilder,
     catalogs: &[CatalogConfig],
     requested_secret_lifetime: &Duration,
+    resolved_fte_config: &Option<ResolvedFaultTolerantExecutionConfig>,
 ) -> Result<()> {
     if let Some(server_tls) = trino.get_server_tls() {
         cb_prepare
@@ -1611,6 +1660,19 @@ fn tls_volume_mounts(
             .context(AddVolumeSnafu)?;
     }
 
+    // fault tolerant execution S3 credentials and other resources
+    if let Some(resolved_fte) = resolved_fte_config {
+        cb_prepare
+            .add_volume_mounts(resolved_fte.volume_mounts.clone())
+            .context(AddVolumeMountSnafu)?;
+        cb_trino
+            .add_volume_mounts(resolved_fte.volume_mounts.clone())
+            .context(AddVolumeMountSnafu)?;
+        pod_builder
+            .add_volumes(resolved_fte.volumes.clone())
+            .context(AddVolumeSnafu)?;
+    }
+
     Ok(())
 }
 
@@ -1780,6 +1842,7 @@ mod tests {
             &trino_authentication_config,
             &trino_opa_config,
             &cluster_info,
+            &None,
         )
         .unwrap()
     }