feat: Handle S3 region

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 - Run a `containerdebug` process in the background of each Spark container to collect debugging information ([#508]).
 - Aggregate emitted Kubernetes events on the CustomResources ([#515]).
 - Support configuring JVM arguments ([#532]).
+- Support for S3 region ([#528]).
 
 ### Changed
 
@@ -20,6 +21,7 @@ All notable changes to this project will be documented in this file.
 [#508]: https://github.com/stackabletech/spark-k8s-operator/pull/508
 [#514]: https://github.com/stackabletech/spark-k8s-operator/pull/514
 [#515]: https://github.com/stackabletech/spark-k8s-operator/pull/515
+[#528]: https://github.com/stackabletech/spark-k8s-operator/pull/528
 [#532]: https://github.com/stackabletech/spark-k8s-operator/pull/532
 
 ## [24.11.1] - 2025-01-10

Cargo.toml

@@ -11,7 +11,7 @@ repository = "https://github.com/stackabletech/spark-k8s-operator"
 
 [workspace.dependencies]
 stackable-versioned = { git = "https://github.com/stackabletech/operator-rs.git", features = ["k8s"], tag = "stackable-versioned-0.5.0" }
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.85.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.87.2" }
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.7.0" }
 
 anyhow = "1.0"
@@ -25,7 +25,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
 snafu = "0.8"
-strum = { version = "0.26", features = ["derive"] }
+strum = { version = "0.27", features = ["derive"] }
 tokio = { version = "1.39", features = ["full"] }
 tracing = "0.1"
 tracing-futures = { version = "0.2", features = ["futures-03"] }

deploy/helm/spark-k8s-operator/crds/crds.yaml

@@ -794,6 +794,20 @@ spec:
                                           minimum: 0.0
                                           nullable: true
                                           type: integer
+                                        region:
+                                          default:
+                                            name: us-east-1
+                                          description: |-
+                                            Bucket region used for signing headers (sigv4).
+
+                                            This defaults to `us-east-1` which is compatible with other implementations such as Minio.
+
+                                            WARNING: Some products use the Hadoop S3 implementation which falls back to us-east-2.
+                                          properties:
+                                            name:
+                                              default: us-east-1
+                                              type: string
+                                          type: object
                                         tls:
                                           description: Use a TLS connection. If not specified no TLS will be used.
                                           nullable: true
@@ -930,6 +944,20 @@ spec:
                           minimum: 0.0
                           nullable: true
                           type: integer
+                        region:
+                          default:
+                            name: us-east-1
+                          description: |-
+                            Bucket region used for signing headers (sigv4).
+
+                            This defaults to `us-east-1` which is compatible with other implementations such as Minio.
+
+                            WARNING: Some products use the Hadoop S3 implementation which falls back to us-east-2.
+                          properties:
+                            name:
+                              default: us-east-1
+                              type: string
+                          type: object
                         tls:
                           description: Use a TLS connection. If not specified no TLS will be used.
                           nullable: true
@@ -1249,6 +1277,20 @@ spec:
                                           minimum: 0.0
                                           nullable: true
                                           type: integer
+                                        region:
+                                          default:
+                                            name: us-east-1
+                                          description: |-
+                                            Bucket region used for signing headers (sigv4).
+
+                                            This defaults to `us-east-1` which is compatible with other implementations such as Minio.
+
+                                            WARNING: Some products use the Hadoop S3 implementation which falls back to us-east-2.
+                                          properties:
+                                            name:
+                                              default: us-east-1
+                                              type: string
+                                          type: object
                                         tls:
                                           description: Use a TLS connection. If not specified no TLS will be used.
                                           nullable: true

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

@@ -484,6 +484,7 @@ mod test {
                     access_style: Default::default(),
                     credentials: None,
                     tls: TlsClientDetails { tls: None },
+                    region: Default::default(),
                 },
             },
             prefix: "prefix".to_string(),

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

@@ -182,7 +182,7 @@ impl S3LogDir {
     pub fn history_server_spark_config(&self) -> Result<BTreeMap<String, String>, Error> {
         let connection = &self.bucket.connection;
 
-        Ok(BTreeMap::from([
+        let config = BTreeMap::from([
             ("spark.history.fs.logDirectory".to_string(), self.url()),
             (
                 "spark.hadoop.fs.s3a.endpoint".to_string(),
@@ -192,7 +192,13 @@ impl S3LogDir {
                 "spark.hadoop.fs.s3a.path.style.access".to_string(),
                 (connection.access_style == S3AccessStyle::Path).to_string(),
             ),
-        ]))
+            (
+                "spark.hadoop.fs.s3a.endpoint.region".to_string(),
+                connection.region.name.clone(),
+            ),
+        ]);
+
+        Ok(config)
     }
 
     pub fn application_spark_config(&self) -> Result<BTreeMap<String, String>, Error> {
@@ -211,6 +217,10 @@ impl S3LogDir {
             format!("spark.hadoop.fs.s3a.bucket.{bucket_name}.path.style.access"),
             (connection.access_style == S3AccessStyle::Path).to_string(),
         );
+        result.insert(
+            format!("spark.hadoop.fs.s3a.bucket.{bucket_name}.region"),
+            connection.region.name.clone(),
+        );
         if let Some(secret_dir) = self.credentials_mount_path() {
             // We don't use the credentials at all here but assume they are available
             result.insert(

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

@@ -572,6 +572,10 @@ impl v1alpha1::SparkApplication {
                 "--conf spark.hadoop.fs.s3a.path.style.access={}",
                 s3conn.access_style == S3AccessStyle::Path
             ));
+            submit_cmd.push(format!(
+                "--conf spark.hadoop.fs.s3a.endpoint.region=\"{region_name}\"",
+                region_name = s3conn.region.name
+            ));
             if let Some(credentials) = &s3conn.credentials {
                 let secret_class_name = credentials.secret_class.clone();
                 let secret_dir = format!("{S3_SECRET_DIR_NAME}/{secret_class_name}");