cleanup, liveliness probe, do not use the iceberg test for now

Author: razvan

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

@@ -1880,11 +1880,11 @@ spec:
                   type: array
                 clusterConfig:
                   default:
-                    listenerClass: cluster-internal
+                    listenerClass: external-unstable
                   description: Global Spark connect server configuration that applies to all roles and role groups.
                   properties:
                     listenerClass:
-                      default: cluster-internal
+                      default: external-unstable
                       description: |-
                         This field controls which type of Service the Operator creates for this ConnectServer:
 

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

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

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

@@ -198,9 +198,9 @@ pub mod versioned {
 #[derive(Clone, Debug, Default, Display, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub enum CurrentlySupportedListenerClasses {
-    #[default]
     #[serde(rename = "cluster-internal")]
     ClusterInternal,
+    #[default]
     #[serde(rename = "external-unstable")]
     ExternalUnstable,
     #[serde(rename = "external-stable")]

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

@@ -223,10 +223,6 @@ pub fn executor_properties(
             "spark.executor.defaultJavaOptions".to_string(),
             Some(executor_jvm_args(scs, config)?),
         ),
-        (
-            "spark.executor.extraClassPath".to_string(),
-            Some("/stackable/spark/extra-jars/*".to_string()),
-        ),
         (
             "spark.kubernetes.executor.podTemplateFile".to_string(),
             Some(format!("{VOLUME_MOUNT_PATH_CONFIG}/{POD_TEMPLATE_FILE}")),

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

@@ -2,8 +2,8 @@ use std::collections::{BTreeMap, HashMap};
 
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
-    builder,
     builder::{
+        self,
         configmap::ConfigMapBuilder,
         meta::ObjectMetaBuilder,
         pod::{
@@ -17,11 +17,11 @@ use stackable_operator::{
         api::{
             apps::v1::{Deployment, DeploymentSpec},
             core::v1::{
-                ConfigMap, EnvVar, PodSecurityContext, Service, ServiceAccount, ServicePort,
-                ServiceSpec,
+                ConfigMap, EnvVar, HTTPGetAction, PodSecurityContext, Probe, Service,
+                ServiceAccount, ServicePort, ServiceSpec,
             },
         },
-        apimachinery::pkg::apis::meta::v1::LabelSelector,
+        apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
     },
     kube::{ResourceExt, runtime::reflector::ObjectRef},
     kvp::{Label, Labels},
@@ -253,7 +253,9 @@ pub fn build_deployment(
         .add_volume_mount(VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_PATH_CONFIG)
         .context(AddVolumeMountSnafu)?
         .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG)
-        .context(AddVolumeMountSnafu)?;
+        .context(AddVolumeMountSnafu)?
+        .readiness_probe(probe())
+        .liveness_probe(probe());
 
     // Add custom log4j config map volumes if configured
     if let Some(cm_name) = config.log_config_map() {
@@ -339,10 +341,18 @@ pub fn build_service(
     app_version_label: &str,
     service_cluster_ip: Option<String>,
 ) -> Result<Service, Error> {
-    let (service_name, service_type) = match service_cluster_ip.clone() {
+    let (service_name, service_type, publish_not_ready_addresses) = match service_cluster_ip.clone()
+    {
         Some(_) => (
+            // These are the properties of the headless driver service used for the internal
+            // communication with the executors as recommended by the Spark docs.
+            //
+            // The flag `publish_not_ready_addresses` *must* be `true` to allow for readiness
+            // probes. Without it, the driver runs into a deadlock beacuse the Pod cannot become
+            // "ready" until the Service is "ready" and vice versa.
             object_name(&scs.name_any(), SparkConnectRole::Server),
             "ClusterIP".to_string(),
+            Some(true),
         ),
         None => (
             format!(
@@ -351,6 +361,7 @@ pub fn build_service(
                 SparkConnectRole::Server
             ),
             scs.spec.cluster_config.listener_class.k8s_service_type(),
+            Some(false),
         ),
     };
 
@@ -393,14 +404,15 @@ pub fn build_service(
                 },
             ]),
             selector: Some(selector),
+            publish_not_ready_addresses,
             ..ServiceSpec::default()
         }),
         status: None,
     })
 }
 
 #[allow(clippy::result_large_err)]
-pub fn command_args(user_args: &[String], spark_version: &str) -> Vec<String> {
+pub fn command_args(user_args: &[String]) -> Vec<String> {
     let mut command = vec![
         // ---------- start containerdebug
         format!(
@@ -411,7 +423,6 @@ pub fn command_args(user_args: &[String], spark_version: &str) -> Vec<String> {
         "--deploy-mode client".to_string(), // 'cluster' mode not supported
         "--master k8s://https://${KUBERNETES_SERVICE_HOST}:${KUBERNETES_SERVICE_PORT_HTTPS}"
             .to_string(),
-        format!("--jars /stackable/spark/connect/spark-connect_2.12-{spark_version}.jar"),
         format!("--properties-file {VOLUME_MOUNT_PATH_CONFIG}/{SPARK_DEFAULTS_FILE_NAME}"),
     ];
 
@@ -461,6 +472,7 @@ pub fn server_properties(
     pi: &ResolvedProductImage,
 ) -> Result<BTreeMap<String, Option<String>>, Error> {
     let spark_image = pi.image.clone();
+    let spark_version = pi.product_version.clone();
     let service_account_name = service_account.name_unchecked();
     let namespace = driver_service
         .namespace()
@@ -498,7 +510,7 @@ pub fn server_properties(
         ),
         (
             "spark.driver.extraClassPath".to_string(),
-            Some("/stackable/spark/extra-jars/*".to_string()),
+            Some(format!("/stackable/spark/extra-jars/*:/stackable/spark/connect/spark-connect_2.12-{spark_version}.jar")),
         ),
     ]
     .into();
@@ -538,3 +550,16 @@ fn server_jvm_args(
         name: scs.name_any(),
     })
 }
+
+fn probe() -> Probe {
+    Probe {
+        http_get: Some(HTTPGetAction {
+            port: IntOrString::Int(CONNECT_UI_PORT),
+            scheme: Some("HTTP".to_string()),
+            path: Some("/metrics".to_string()),
+            ..Default::default()
+        }),
+        failure_threshold: Some(10),
+        ..Probe::default()
+    }
+}

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

@@ -14,10 +14,10 @@ apiVersion: kuttl.dev/v1beta1
 kind: TestAssert
 timeout: 90
 commands:
-  # Test that two spark-connect executors are running.
+  # Test that spark connect executors are running.
   # Sleep to prevent the following spark connect app from failing
   # while the spark-connect server is busy setting up the executors.
   - script: |
       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 3 -eq "$EXECUTOR_COUNT"
+      test 1 -eq "$EXECUTOR_COUNT"

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

@@ -35,6 +35,8 @@ spec:
 {% if lookup('env', 'VECTOR_AGGREGATOR') %}
   vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
+  args:
+    - --packages org.apache.iceberg:iceberg-spark-runtime-{{ ".".join(test_scenario['values']['spark-connect'].split('.')[:2]) }}_2.12:1.8.1
   server:
     podOverrides:
       spec:
@@ -55,10 +57,16 @@ spec:
               configMap: spark-connect-log-config
     configOverrides:
       spark-defaults.conf:
-        spark.executor.instances: "3"
+        spark.jars.ivy: /tmp/ivy2
+        spark.sql.extensions: org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
+        spark.sql.catalog.local: org.apache.iceberg.spark.SparkCatalog
+        spark.sql.catalog.local.type: hadoop
+        spark.sql.catalog.local.warehouse: /tmp/warehouse
+        spark.sql.defaultCatalog: local
   executor:
     configOverrides:
       spark-defaults.conf:
+        spark.executor.instances: "1"
         spark.executor.memoryOverhead: "1m"
     config:
       logging:

tests/templates/kuttl/spark-connect/20-run-connect-client.yaml.j2

@@ -1,3 +1,44 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: simple-connect-app
+  labels:
+    stackable.tech/vendor: Stackable
+data:
+  simple-connect-app.py: |
+    import sys
+
+    from pyspark.sql import SparkSession
+    from pyspark.sql.types import *
+
+    remote = sys.argv[1]
+    spark = (SparkSession.builder
+      .remote(remote)
+      .appName("simple-connect-app")
+      .getOrCreate())
+
+    schema = StructType([
+      StructField("id", LongType(), True),
+      StructField("data", StringType(), True)
+    ])
+
+
+    # create table
+    df = spark.createDataFrame([], schema)
+    df.writeTo("local.db.table").create()
+
+    # append to table
+    data = [
+        (1,"one"),
+        (2,"two"),
+        (3,"three"),
+        (4,"four")
+      ]
+
+    df = spark.createDataFrame(data, schema)
+    df.writeTo("local.db.table").append()
+
 ---
 apiVersion: batch/v1
 kind: Job
@@ -10,6 +51,10 @@ spec:
     spec:
       restartPolicy: OnFailure
       activeDeadlineSeconds: 100
+      volumes:
+        - name: script
+          configMap:
+            name: simple-connect-app
       containers:
         - name: simple-connect-app
 {% if test_scenario['values']['spark-connect-client'].find(",") > 0 %}
@@ -18,6 +63,12 @@ spec:
           image: oci.stackable.tech/sdp/spark-connect-client:{{ test_scenario['values']['spark-connect-client'] }}-stackable0.0.0-dev
 {% endif %}
           imagePullPolicy: IfNotPresent
+          #
+          # TODO: cannot use the PySpark job from the ConfigMap because it breaks
+          # with a "iceberg.SparkWrite$WriterFactory" ClassNotfoundException.
+          # Use the app bundled within spark-connect-client instead
+          # "/app/simple-connect-app.py",
+          #
           command:
             [
               "/usr/bin/python",
@@ -31,3 +82,6 @@ spec:
             requests:
               cpu: 200m
               memory: 128Mi
+          volumeMounts:
+            - name: script
+              mountPath: /app