Added s3 section (#42)

## Description

- Added s3 section to the CRD.
- credentials and endpoint are written to env vars/config options

fixes #37 

A test case has been added here: https://github.com/stackabletech/integration-tests/pull/205 and should be reviewed at the same time.



Co-authored-by: Felix Hennig <[email protected]>
Co-authored-by: Andrew Kenworthy <[email protected]>

Author: 3 people

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

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

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

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

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";

rust/crd/src/lib.rs

@@ -3,7 +3,9 @@
 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::{BTreeMap, HashMap};
 
@@ -77,6 +79,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>>,
@@ -97,6 +101,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;
@@ -136,7 +148,20 @@ 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(
+                ENV_AWS_ACCESS_KEY_ID,
+                &s3.credentials_secret,
+                ACCESS_KEY_ID,
+            ));
+            e.push(env_var_from_secret(
+                ENV_AWS_SECRET_ACCESS_KEY,
+                &s3.credentials_secret,
+                SECRET_ACCESS_KEY,
+            ));
+        }
+        e
     }
 
     pub fn volumes(&self) -> Vec<Volume> {
@@ -195,12 +220,12 @@ 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()) {
+            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 {
@@ -325,6 +350,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;