Use "log directory" instead of "S3 log directory" when the resolved log directory structure is used

Author: siegfriedweber

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Make spark-env.sh configurable via `configOverrides` ([#473]).
+- Allow specifying a custom log directory ([#479]).
 
 ### Changed
 
@@ -33,6 +34,7 @@ All notable changes to this project will be documented in this file.
 [#460]: https://github.com/stackabletech/spark-k8s-operator/pull/460
 [#472]: https://github.com/stackabletech/spark-k8s-operator/pull/472
 [#473]: https://github.com/stackabletech/spark-k8s-operator/pull/473
+[#479]: https://github.com/stackabletech/spark-k8s-operator/pull/479
 
 ## [24.7.0] - 2024-07-24
 

rust/crd/src/history.rs

@@ -1,4 +1,4 @@
-use crate::s3logdir::ResolvedLogDir;
+use crate::logdir::ResolvedLogDir;
 use crate::{affinity::history_affinity, constants::*};
 
 use product_config::{types::PropertyNameKind, ProductConfigManager};
@@ -77,7 +77,6 @@ pub struct SparkHistoryServerSpec {
     pub vector_aggregator_config_map_name: Option<String>,
 
     /// The log file directory definition used by the Spark history server.
-    /// Currently only S3 buckets are supported.
     pub log_file_directory: LogFileDirectorySpec,
 
     /// A map of key/value strings that will be passed directly to Spark when deploying the history server.
@@ -456,7 +455,7 @@ impl Configuration for HistoryConfigFragment {
 
 #[cfg(test)]
 mod test {
-    use crate::s3logdir::S3LogDir;
+    use crate::logdir::S3LogDir;
 
     use super::*;
     use indoc::indoc;

rust/crd/src/lib.rs

@@ -3,15 +3,15 @@
 pub mod affinity;
 pub mod constants;
 pub mod history;
+pub mod logdir;
 pub mod roles;
-pub mod s3logdir;
 pub mod tlscerts;
 
 pub use crate::roles::*;
 use constants::*;
 use history::LogFileDirectorySpec;
+use logdir::ResolvedLogDir;
 use product_config::{types::PropertyNameKind, ProductConfigManager};
-use s3logdir::ResolvedLogDir;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
@@ -101,8 +101,8 @@ pub enum Error {
     #[snafu(display("failed to configure S3 connection/bucket"))]
     ConfigureS3 { source: S3Error },
 
-    #[snafu(display("failed to configure S3 log directory"))]
-    ConfigureS3LogDir { source: s3logdir::Error },
+    #[snafu(display("failed to configure log directory"))]
+    ConfigureLogDir { source: logdir::Error },
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Serialize, JsonSchema)]
@@ -313,10 +313,7 @@ impl SparkApplication {
         }
 
         if let Some(log_dir) = logdir.as_ref() {
-            if let Some(volume) = log_dir
-                .credentials_volume()
-                .context(ConfigureS3LogDirSnafu)?
-            {
+            if let Some(volume) = log_dir.credentials_volume().context(ConfigureLogDirSnafu)? {
                 result.push(volume);
             }
         }
@@ -648,7 +645,7 @@ impl SparkApplication {
             submit_conf.extend(
                 log_dir
                     .application_spark_config()
-                    .context(ConfigureS3LogDirSnafu)?,
+                    .context(ConfigureLogDirSnafu)?,
             );
         }
 

rust/crd/src/s3logdir.rs renamed to rust/crd/src/logdir.rs

@@ -8,7 +8,7 @@ use crate::{
         STACKABLE_MOUNT_PATH_TLS, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE,
         SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD,
     },
-    s3logdir::ResolvedLogDir,
+    logdir::ResolvedLogDir,
 };
 
 pub fn tls_secret_name(s3conn: &ResolvedS3Connection) -> Option<&str> {

rust/crd/src/tlscerts.rs

@@ -37,7 +37,7 @@ use stackable_operator::{
     time::Duration,
 };
 use stackable_spark_k8s_crd::constants::{METRICS_PORT, SPARK_ENV_SH_FILE_NAME};
-use stackable_spark_k8s_crd::s3logdir::ResolvedLogDir;
+use stackable_spark_k8s_crd::logdir::ResolvedLogDir;
 use stackable_spark_k8s_crd::{
     constants::{
         ACCESS_KEY_ID, APP_NAME, HISTORY_CONTROLLER_NAME, HISTORY_ROLE_NAME,
@@ -124,9 +124,9 @@ pub enum Error {
     #[snafu(display("number of cleaner replicas exceeds 1"))]
     TooManyCleanerReplicas,
 
-    #[snafu(display("failed to resolve the s3 log dir configuration"))]
-    S3LogDir {
-        source: stackable_spark_k8s_crd::s3logdir::Error,
+    #[snafu(display("failed to resolve the log dir configuration"))]
+    LogDir {
+        source: stackable_spark_k8s_crd::logdir::Error,
     },
 
     #[snafu(display("failed to create cluster resources"))]
@@ -184,9 +184,9 @@ pub enum Error {
             stackable_operator::kvp::KeyValuePairError<stackable_operator::kvp::LabelValueError>,
     },
 
-    #[snafu(display("failed to get create the S3 log dir"))]
-    CreateS3LogDirVolumes {
-        source: stackable_spark_k8s_crd::s3logdir::Error,
+    #[snafu(display("failed to create the log dir volumes specification"))]
+    CreateLogDirVolumesSpec {
+        source: stackable_spark_k8s_crd::logdir::Error,
     },
 
     #[snafu(display("failed to add needed volume"))]
@@ -230,7 +230,7 @@ pub async fn reconcile(shs: Arc<SparkHistoryServer>, ctx: Arc<Ctx>) -> Result<Ac
         client,
     )
     .await
-    .context(S3LogDirSnafu)?;
+    .context(LogDirSnafu)?;
 
     let vector_aggregator_address = resolve_vector_aggregator_address(
         client,
@@ -475,7 +475,7 @@ fn build_stateful_set(
                 .build(),
         )
         .context(AddVolumeSnafu)?
-        .add_volumes(log_dir.volumes().context(CreateS3LogDirVolumesSnafu)?)
+        .add_volumes(log_dir.volumes().context(CreateLogDirVolumesSpecSnafu)?)
         .context(AddVolumeSnafu)?
         .security_context(PodSecurityContext {
             run_as_user: Some(SPARK_UID),
@@ -684,9 +684,7 @@ fn spark_defaults(
     log_dir: &ResolvedLogDir,
     rolegroupref: &RoleGroupRef<SparkHistoryServer>,
 ) -> Result<String, Error> {
-    let mut log_dir_settings = log_dir
-        .history_server_spark_config()
-        .context(S3LogDirSnafu)?;
+    let mut log_dir_settings = log_dir.history_server_spark_config().context(LogDirSnafu)?;
 
     // add cleaner spark settings if requested
     log_dir_settings.extend(cleaner_config(shs, rolegroupref)?);

rust/operator-binary/src/history/history_controller.rs

@@ -17,7 +17,7 @@ use stackable_operator::{
     time::Duration,
 };
 use stackable_spark_k8s_crd::{
-    constants::*, s3logdir::ResolvedLogDir, tlscerts, to_spark_env_sh_string, RoleConfig,
+    constants::*, logdir::ResolvedLogDir, tlscerts, to_spark_env_sh_string, RoleConfig,
     SparkApplication, SparkApplicationRole, SparkApplicationStatus, SparkContainer, SubmitConfig,
 };
 
@@ -122,9 +122,9 @@ pub enum Error {
         source: stackable_operator::builder::pod::container::Error,
     },
 
-    #[snafu(display("failed to resolve the s3 log dir configuration"))]
-    S3LogDir {
-        source: stackable_spark_k8s_crd::s3logdir::Error,
+    #[snafu(display("failed to resolve the log dir configuration"))]
+    LogDir {
+        source: stackable_spark_k8s_crd::logdir::Error,
     },
 
     #[snafu(display("failed to resolve the Vector aggregator address"))]
@@ -253,7 +253,7 @@ pub async fn reconcile(spark_application: Arc<SparkApplication>, ctx: Arc<Ctx>)
                 client,
             )
             .await
-            .context(S3LogDirSnafu)?,
+            .context(LogDirSnafu)?,
         )
     } else {
         None

rust/operator-binary/src/spark_k8s_controller.rs

@@ -16,7 +16,7 @@ spec:
   vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
   logFileDirectory:
-    customLogDirectory: hdfs:///
+    customLogDirectory: hdfs:////
   nodes:
     config:
       logging: