implement user provided command line args for the connect server

Author: razvan

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

@@ -1872,6 +1872,12 @@ spec:
             spec:
               description: A Spark cluster connect server component. This resource is managed by the Stackable operator for Apache Spark. Find more information on how to use it in the [operator documentation](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/connect-server).
               properties:
+                args:
+                  default: []
+                  description: User provided command line arguments appended to the server entry point.
+                  items:
+                    type: string
+                  type: array
                 clusterConfig:
                   default:
                     listenerClass: cluster-internal

rust/operator-binary/src/connect/controller.rs

@@ -312,7 +312,7 @@ pub async fn reconcile(
             name: scs.name_unchecked(),
         })?;
 
-    let args = server::command_args(&resolved_product_image.product_version);
+    let args = server::command_args(&scs.spec.args, &resolved_product_image.product_version);
     let deployment = server::build_deployment(
         scs,
         &server_config,

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

@@ -102,6 +102,10 @@ pub mod versioned {
         #[serde(default)]
         pub cluster_operation: ClusterOperation,
 
+        /// User provided command line arguments appended to the server entry point.
+        #[serde(default)]
+        pub args: Vec<String>,
+
         /// Name of the Vector aggregator discovery ConfigMap.
         /// It must contain the key `ADDRESS` with the address of the Vector aggregator.
         #[serde(skip_serializing_if = "Option::is_none")]
@@ -151,6 +155,7 @@ pub mod versioned {
     pub struct ServerConfig {
         #[fragment_attrs(serde(default))]
         pub resources: Resources<crate::connect::crd::ConnectStorageConfig, NoRuntimeLimits>,
+
         #[fragment_attrs(serde(default))]
         pub logging: Logging<SparkConnectContainer>,
 

rust/operator-binary/src/connect/executor.rs

@@ -252,7 +252,10 @@ pub fn executor_properties(
         "spark.kubernetes.executor.limit.cores".to_string(),
         max.clone().map(|v| v.0),
     );
-    result.insert("spark.executor.cores".to_string(), min.clone().map(|v| v.0));
+    result.insert(
+        "spark.kubernetes.executor.request.cores".to_string(),
+        min.clone().map(|v| v.0),
+    );
     result.insert(
         "spark.executor.memory".to_string(),
         limit.clone().map(|v| v.0),

rust/operator-binary/src/connect/server.rs

@@ -400,8 +400,8 @@ pub fn build_service(
 }
 
 #[allow(clippy::result_large_err)]
-pub fn command_args(spark_version: &str) -> Vec<String> {
-    let command = [
+pub fn command_args(user_args: &[String], spark_version: &str) -> Vec<String> {
+    let mut command = vec![
         // ---------- start containerdebug
         format!(
             "containerdebug --output={VOLUME_MOUNT_PATH_LOG}/containerdebug-state.json --loop &"
@@ -415,6 +415,9 @@ pub fn command_args(spark_version: &str) -> Vec<String> {
         format!("--properties-file {VOLUME_MOUNT_PATH_CONFIG}/{SPARK_DEFAULTS_FILE_NAME}"),
     ];
 
+    // User provided command line arguments
+    command.extend_from_slice(user_args);
+
     vec![command.join(" ")]
 }
 

tests/templates/kuttl/spark-connect/10-assert.yaml

@@ -12,12 +12,12 @@ status:
 ---
 apiVersion: kuttl.dev/v1beta1
 kind: TestAssert
-timeout: 30
+timeout: 90
 commands:
   # Test that two spark-connect executors are running.
-  # Sleep 5 to prevent the following spark connect app from failing
+  # Sleep to prevent the following spark connect app from failing
   # while the spark-connect server is busy setting up the executors.
   - script: |
-      sleep 5
+      sleep 10
       EXECUTOR_COUNT=$(kubectl get pods -n $NAMESPACE --selector 'spark-app-name=spark-connect-server' --field-selector='status.phase=Running' -o NAME|wc -l)
-      test 2 -eq "$EXECUTOR_COUNT"
+      test 3 -eq "$EXECUTOR_COUNT"

tests/templates/kuttl/spark-connect/10-deploy-spark-connect.yaml.j2

@@ -53,10 +53,13 @@ spec:
           spark:
             custom:
               configMap: spark-connect-log-config
+    configOverrides:
+      spark-defaults.conf:
+        spark.executor.instances: "3"
   executor:
     configOverrides:
       spark-defaults.conf:
-        spark.executor.memoryOverhead: 1m
+        spark.executor.memoryOverhead: "1m"
     config:
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}