From f11b5821ca95c5325532e29f7ef9ea38015b1dd8 Mon Sep 17 00:00:00 2001
From: Felix Hennig <felix.hennig@stackable.de>
Date: Thu, 31 Mar 2022 13:44:12 +0200
Subject: [PATCH 1/4] Added s3 section

---
 deploy/crd/sparkapplication.crd.yaml          | 11 ++++++
 deploy/helm/spark-k8s-operator/crds/crds.yaml | 11 ++++++
 deploy/manifests/crds.yaml                    | 11 ++++++
 rust/crd/src/lib.rs                           | 39 ++++++++++++++++++-
 .../src/spark_k8s_controller.rs               |  5 +--
 5 files changed, 71 insertions(+), 6 deletions(-)

diff --git a/deploy/crd/sparkapplication.crd.yaml b/deploy/crd/sparkapplication.crd.yaml
index cbda8463..b79fe882 100644
--- a/deploy/crd/sparkapplication.crd.yaml
+++ b/deploy/crd/sparkapplication.crd.yaml
@@ -263,6 +263,17 @@ spec:
                 mode:
                   nullable: true
                   type: string
+                s3:
+                  nullable: true
+                  properties:
+                    credentialsSecret:
+                      type: string
+                    endpoint:
+                      nullable: true
+                      type: string
+                  required:
+                    - credentialsSecret
+                  type: object
                 sparkConf:
                   additionalProperties:
                     type: string
diff --git a/deploy/helm/spark-k8s-operator/crds/crds.yaml b/deploy/helm/spark-k8s-operator/crds/crds.yaml
index 5eb29f98..ef22569f 100644
--- a/deploy/helm/spark-k8s-operator/crds/crds.yaml
+++ b/deploy/helm/spark-k8s-operator/crds/crds.yaml
@@ -265,6 +265,17 @@ spec:
                 mode:
                   nullable: true
                   type: string
+                s3:
+                  nullable: true
+                  properties:
+                    credentialsSecret:
+                      type: string
+                    endpoint:
+                      nullable: true
+                      type: string
+                  required:
+                    - credentialsSecret
+                  type: object
                 sparkConf:
                   additionalProperties:
                     type: string
diff --git a/deploy/manifests/crds.yaml b/deploy/manifests/crds.yaml
index db8cf8f1..01331b42 100644
--- a/deploy/manifests/crds.yaml
+++ b/deploy/manifests/crds.yaml
@@ -267,6 +267,17 @@ spec:
                 mode:
                   nullable: true
                   type: string
+                s3:
+                  nullable: true
+                  properties:
+                    credentialsSecret:
+                      type: string
+                    endpoint:
+                      nullable: true
+                      type: string
+                  required:
+                    - credentialsSecret
+                  type: object
                 sparkConf:
                   additionalProperties:
                     type: string
diff --git a/rust/crd/src/lib.rs b/rust/crd/src/lib.rs
index d9a14374..8146bf00 100644
--- a/rust/crd/src/lib.rs
+++ b/rust/crd/src/lib.rs
@@ -3,7 +3,7 @@
 pub mod constants;
 
 use constants::*;
-use stackable_operator::k8s_openapi::api::core::v1::{EnvVar, Volume, VolumeMount};
+use stackable_operator::k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector, Volume, VolumeMount};
 
 use std::collections::HashMap;
 
@@ -76,6 +76,8 @@ pub struct SparkApplicationSpec {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub deps: Option<JobDependencies>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub s3: Option<S3>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub args: Option<Vec<String>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub volumes: Option<Vec<Volume>>,
@@ -96,6 +98,14 @@ pub struct JobDependencies {
     pub exclude_packages: Option<Vec<String>>,
 }
 
+#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct S3 {
+    pub credentials_secret: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub endpoint: Option<String>,
+}
+
 impl SparkApplication {
     pub fn enable_monitoring(&self) -> Option<bool> {
         let spec: &SparkApplicationSpec = &self.spec;
@@ -131,7 +141,12 @@ impl SparkApplication {
 
     pub fn env(&self) -> Vec<EnvVar> {
         let tmp = self.spec.env.as_ref();
-        tmp.iter().flat_map(|e| e.iter()).cloned().collect()
+        let mut e: Vec<EnvVar> = tmp.iter().flat_map(|e| e.iter()).cloned().collect();
+        if let Some(s3) = self.spec.s3.as_ref() {
+            e.push(env_var_from_secret("AWS_ACCCESS_KEY_ID", &s3.credentials_secret, "acessKeyId"));
+            e.push(env_var_from_secret("AWS_SECRET_ACCESS_KEY", &s3.credentials_secret, "secretAccessKey"));
+        }
+        e
     }
 
     pub fn volumes(&self) -> Vec<Volume> {
@@ -183,6 +198,11 @@ impl SparkApplication {
             //"--conf spark.hadoop.fs.s3a.fast.upload=true".to_string(),
         ];
 
+
+        if let Some(endpoint) = self.spec.s3.as_ref().and_then(|s3| s3.endpoint.as_ref()) {
+            submit_cmd.push(format!("--conf spark.hadoop.fs.s3a.endpoint={}", endpoint));
+        }
+
         // conf arguments that are not driver or executor specific
         if let Some(spark_conf) = self.spec.spark_conf.clone() {
             for (key, value) in spark_conf {
@@ -307,6 +327,21 @@ pub struct CommandStatus {
     pub finished_at: Option<Time>,
 }
 
+fn env_var_from_secret(var_name: &str, secret: &str, secret_key: &str) -> EnvVar {
+    EnvVar {
+        name: String::from(var_name),
+        value_from: Some(EnvVarSource {
+            secret_key_ref: Some(SecretKeySelector {
+                name: Some(String::from(secret)),
+                key: String::from(secret_key),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::SparkApplication;
diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs
index db399153..5e9e9262 100644
--- a/rust/operator-binary/src/spark_k8s_controller.rs
+++ b/rust/operator-binary/src/spark_k8s_controller.rs
@@ -3,10 +3,7 @@ use stackable_operator::builder::{
     ConfigMapBuilder, ContainerBuilder, ObjectMetaBuilder, PodBuilder, VolumeBuilder,
 };
 use stackable_operator::k8s_openapi::api::batch::v1::{Job, JobSpec};
-use stackable_operator::k8s_openapi::api::core::v1::{
-    ConfigMap, ConfigMapVolumeSource, Container, EmptyDirVolumeSource, EnvVar, Pod, PodSpec,
-    PodTemplateSpec, Volume, VolumeMount,
-};
+use stackable_operator::k8s_openapi::api::core::v1::{ConfigMap, ConfigMapVolumeSource, Container, EmptyDirVolumeSource, EnvVar, Pod, PodSpec, PodTemplateSpec, Volume, VolumeMount};
 use stackable_operator::logging::controller::ReconcilerError;
 use stackable_operator::{
     kube::runtime::controller::{Action, Context},

From 7649f5c2bef5803572c2926cf7f0b912825aeed4 Mon Sep 17 00:00:00 2001
From: Felix Hennig <felix.hennig@stackable.de>
Date: Thu, 31 Mar 2022 13:54:34 +0200
Subject: [PATCH 2/4] fmt; CHANGELOG

---
 CHANGELOG.md        |  4 +++-
 rust/crd/src/lib.rs | 17 +++++++++++++----
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 49b38cc9..128500b3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,7 +6,9 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
-- ServiceAccount, ClusterRole and RoleBinding for Spark driver ([#39])
 - Initial commit
+- ServiceAccount, ClusterRole and RoleBinding for Spark driver ([#39])
+- S3 credentials can be provided via a Secret ([#42])
 
 [#39]: https://github.com/stackabletech/spark-k8s-operator/pull/39
+[#42]: https://github.com/stackabletech/spark-k8s-operator/pull/42
diff --git a/rust/crd/src/lib.rs b/rust/crd/src/lib.rs
index fd591bb3..28c5fb14 100644
--- a/rust/crd/src/lib.rs
+++ b/rust/crd/src/lib.rs
@@ -3,7 +3,9 @@
 pub mod constants;
 
 use constants::*;
-use stackable_operator::k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector, Volume, VolumeMount};
+use stackable_operator::k8s_openapi::api::core::v1::{
+    EnvVar, EnvVarSource, SecretKeySelector, Volume, VolumeMount,
+};
 
 use std::collections::{BTreeMap, HashMap};
 
@@ -148,8 +150,16 @@ impl SparkApplication {
         let tmp = self.spec.env.as_ref();
         let mut e: Vec<EnvVar> = tmp.iter().flat_map(|e| e.iter()).cloned().collect();
         if let Some(s3) = self.spec.s3.as_ref() {
-            e.push(env_var_from_secret("AWS_ACCCESS_KEY_ID", &s3.credentials_secret, "acessKeyId"));
-            e.push(env_var_from_secret("AWS_SECRET_ACCESS_KEY", &s3.credentials_secret, "secretAccessKey"));
+            e.push(env_var_from_secret(
+                "AWS_ACCCESS_KEY_ID",
+                &s3.credentials_secret,
+                "acessKeyId",
+            ));
+            e.push(env_var_from_secret(
+                "AWS_SECRET_ACCESS_KEY",
+                &s3.credentials_secret,
+                "secretAccessKey",
+            ));
         }
         e
     }
@@ -216,7 +226,6 @@ impl SparkApplication {
             //"--conf spark.hadoop.fs.s3a.fast.upload=true".to_string(),
         ];
 
-
         if let Some(endpoint) = self.spec.s3.as_ref().and_then(|s3| s3.endpoint.as_ref()) {
             submit_cmd.push(format!("--conf spark.hadoop.fs.s3a.endpoint={}", endpoint));
         }

From afd52e0a6601c137734053959a6744b8523e421b Mon Sep 17 00:00:00 2001
From: Andrew Kenworthy <andrew.kenworthy@stackable.de>
Date: Thu, 31 Mar 2022 16:14:43 +0200
Subject: [PATCH 3/4] fixed aws env-variables

---
 rust/crd/src/constants.rs | 5 +++++
 rust/crd/src/lib.rs       | 8 ++++----
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/rust/crd/src/constants.rs b/rust/crd/src/constants.rs
index 8fdb14a6..47f03d31 100644
--- a/rust/crd/src/constants.rs
+++ b/rust/crd/src/constants.rs
@@ -16,3 +16,8 @@ pub const CONTAINER_NAME_DRIVER: &str = "spark-driver";
 
 pub const CONTAINER_IMAGE_NAME_EXECUTOR: &str = "dummy-overwritten-by-command-line";
 pub const CONTAINER_NAME_EXECUTOR: &str = "spark-executor";
+
+pub const ENV_AWS_ACCESS_KEY_ID: &str = "AWS_ACCESS_KEY_ID";
+pub const ENV_AWS_SECRET_ACCESS_KEY: &str = "AWS_SECRET_ACCESS_KEY";
+pub const ACCESS_KEY_ID: &str = "accessKeyId";
+pub const SECRET_ACCESS_KEY: &str = "secretAccessKey";
diff --git a/rust/crd/src/lib.rs b/rust/crd/src/lib.rs
index 28c5fb14..9510700b 100644
--- a/rust/crd/src/lib.rs
+++ b/rust/crd/src/lib.rs
@@ -151,14 +151,14 @@ impl SparkApplication {
         let mut e: Vec<EnvVar> = tmp.iter().flat_map(|e| e.iter()).cloned().collect();
         if let Some(s3) = self.spec.s3.as_ref() {
             e.push(env_var_from_secret(
-                "AWS_ACCCESS_KEY_ID",
+                ENV_AWS_ACCESS_KEY_ID,
                 &s3.credentials_secret,
-                "acessKeyId",
+                ACCESS_KEY_ID,
             ));
             e.push(env_var_from_secret(
-                "AWS_SECRET_ACCESS_KEY",
+                ENV_AWS_SECRET_ACCESS_KEY,
                 &s3.credentials_secret,
-                "secretAccessKey",
+                SECRET_ACCESS_KEY,
             ));
         }
         e

From 1f44e0623826a2eaeb75fb166cfcbaf744e92a21 Mon Sep 17 00:00:00 2001
From: Andrew Kenworthy <andrew.kenworthy@stackable.de>
Date: Thu, 31 Mar 2022 18:09:01 +0200
Subject: [PATCH 4/4] removed dead code

---
 rust/crd/src/lib.rs | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/rust/crd/src/lib.rs b/rust/crd/src/lib.rs
index 9510700b..96f9710d 100644
--- a/rust/crd/src/lib.rs
+++ b/rust/crd/src/lib.rs
@@ -220,10 +220,6 @@ impl SparkApplication {
             format!("--conf spark.kubernetes.driver.container.image={}", self.spec.spark_image.as_ref().context(NoSparkImageSnafu)?),
             format!("--conf spark.kubernetes.executor.container.image={}", self.spec.spark_image.as_ref().context(NoSparkImageSnafu)?),
             format!("--conf spark.kubernetes.authenticate.driver.serviceAccountName={}", serviceaccount_name),
-            //"--conf spark.kubernetes.file.upload.path=s3a://stackable-spark-k8s-jars/jobs".to_string(),
-            //"--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem".to_string(),
-            //"--conf spark.driver.extraClassPath=/stackable/.ivy2/cache".to_string(),
-            //"--conf spark.hadoop.fs.s3a.fast.upload=true".to_string(),
         ];
 
         if let Some(endpoint) = self.spec.s3.as_ref().and_then(|s3| s3.endpoint.as_ref()) {