chore(kuttl): add test for failing spark job

tests/templates/kuttl/graceful-shutdown/00-patch-ns.yaml.j2

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/graceful-shutdown/01-assert.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/graceful-shutdown/01-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}

tests/templates/kuttl/graceful-shutdown/10-assert.yaml

@@ -0,0 +1,19 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 900
+commands:
+  # Both executors have to be registered and working: an executor that is still registering is
+  # not in the driver's list yet and would never be asked to shut down.
+  - script: |
+      set -eu
+
+      registered=0
+      for pod in $(kubectl -n "$NAMESPACE" get pods -o name \
+        -l spark-role=executor,app.kubernetes.io/instance=graceful-shutdown | cut -d/ -f2); do
+        log=$(kubectl -n "$NAMESPACE" logs --tail=-1 "$pod" -c spark)
+        echo "$log" | grep -q 'Successfully registered with driver'
+        echo "$log" | grep -q 'Running task'
+        registered=$(( registered + 1 ))
+      done
+      test "$registered" -eq 2

tests/templates/kuttl/graceful-shutdown/10-deploy-long-running-app.yaml.j2

@@ -0,0 +1,55 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: long-running-job
+data:
+  # Keeps the executors busy for longer than the test runs, so that they are stopped mid-task.
+  long-running-job.py: |
+    import time
+
+    from pyspark.sql import SparkSession
+
+    spark = SparkSession.builder.appName("long-running-job").getOrCreate()
+
+    def slow(i):
+        time.sleep(600)
+        return i
+
+    print(spark.sparkContext.parallelize(range(4), 4).map(slow).collect())
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: graceful-shutdown
+spec:
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+  vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  sparkImage:
+{% if test_scenario['values']['spark'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['spark'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['spark'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  mode: cluster
+  mainApplicationFile: "local:///stackable/spark/jobs/long-running-job.py"
+  driver:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+      volumeMounts:
+        - name: script
+          mountPath: /stackable/spark/jobs
+  # Only the driver reads the script; the executors receive the closure over the wire.
+  executor:
+    replicas: 2
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+  volumes:
+    - name: script
+      configMap:
+        name: long-running-job

tests/templates/kuttl/graceful-shutdown/11-check-executor-shutdown.yaml

@@ -0,0 +1,61 @@
+---
+# An executor whose Pod is deleted (node drain, eviction, downscale) has to be told to stop while
+# it is running a task. If SIGTERM is not forwarded to the JVM, the container ignores it and is
+# SIGKILLed when the grace period expires, taking the running task with it (docker-images#1564).
+#
+# TestStep and not TestAssert: kuttl retries asserts until they time out, which would turn every
+# failure into a generic timeout.
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - timeout: 600
+    script: |
+      set -eu
+
+      selector="spark-role=executor,app.kubernetes.io/instance=graceful-shutdown"
+      budget_seconds=20
+
+      victim=$(kubectl -n "$NAMESPACE" get pods -o name -l "$selector" | head -1 | cut -d/ -f2)
+      if [ -z "$victim" ]; then
+        echo "FAIL: no executor Pod found to stop"
+        exit 1
+      fi
+
+      echo "stopping executor $victim"
+
+      # The log cannot be read once the Pod object is gone, so stream it away first.
+      log=$(mktemp)
+      kubectl -n "$NAMESPACE" logs -f "$victim" -c spark >"$log" 2>&1 &
+      logs_pid=$!
+      sleep 2
+
+      if ! grep -q 'Running task' "$log"; then
+        echo "FAIL: could not stream the log of $victim, so its shutdown cannot be observed"
+        tail -5 "$log"
+        exit 1
+      fi
+
+      kubectl -n "$NAMESPACE" delete pod "$victim" --wait=false
+
+      start=$(date +%s)
+      stopped=no
+      while [ $(( $(date +%s) - start )) -lt "$budget_seconds" ]; do
+        if grep -q 'shutdown-hook' "$log"; then
+          stopped=yes
+          break
+        fi
+        sleep 1
+      done
+      elapsed=$(( $(date +%s) - start ))
+      kill "$logs_pid" 2>/dev/null || true
+
+      if [ "$stopped" != yes ]; then
+        echo "FAIL: ${elapsed}s after its Pod was deleted the executor had run no shutdown hook,"
+        echo "so SIGTERM did not reach the JVM and the kubelet will SIGKILL it at the end of the"
+        echo "grace period, killing the task it is running."
+        tail -5 "$log"
+        exit 1
+      fi
+
+      rm -f "$log"
+      echo "OK: the executor ran its shutdown hooks ${elapsed}s after its Pod was deleted"

tests/templates/kuttl/graceful-shutdown/12-assert.yaml

@@ -0,0 +1,19 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 600
+commands:
+  # Same check as in step 10: the replacement for the stopped executor has to be registered and
+  # working before the driver is stopped, otherwise it is not in the driver's list yet.
+  - script: |
+      set -eu
+
+      registered=0
+      for pod in $(kubectl -n "$NAMESPACE" get pods -o name \
+        -l spark-role=executor,app.kubernetes.io/instance=graceful-shutdown | cut -d/ -f2); do
+        log=$(kubectl -n "$NAMESPACE" logs --tail=-1 "$pod" -c spark)
+        echo "$log" | grep -q 'Successfully registered with driver'
+        echo "$log" | grep -q 'Running task'
+        registered=$(( registered + 1 ))
+      done
+      test "$registered" -eq 2

tests/templates/kuttl/graceful-shutdown/13-check-driver-shutdown-propagation.yaml

@@ -0,0 +1,77 @@
+---
+# Deleting the driver Pod has to let the driver JVM stop the SparkContext, which is what asks the
+# executors to shut down. Without SIGTERM reaching the driver it is SIGKILLed at the end of its
+# grace period, the executors are never asked, and they keep working until the driver's RPC
+# connection dies.
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - timeout: 600
+    script: |
+      set -eu
+
+      app=graceful-shutdown
+      # The stop command reaches registered executors about a second after the driver is signalled.
+      budget_seconds=20
+
+      capture=$(mktemp -d)
+      driver=$(kubectl -n "$NAMESPACE" get pods -o name \
+        -l "spark-role=driver,app.kubernetes.io/instance=$app" | head -1 | cut -d/ -f2)
+      executors=$(kubectl -n "$NAMESPACE" get pods -o name \
+        -l "spark-role=executor,app.kubernetes.io/instance=$app" | cut -d/ -f2)
+
+      # Without these the loop below has nothing to wait for and would report success while having
+      # checked nothing at all.
+      if [ -z "$driver" ]; then
+        echo "FAIL: no driver Pod found for $app"
+        exit 1
+      fi
+      if [ -z "$executors" ]; then
+        echo "FAIL: no executor Pods found for $app, so no shutdown can be propagated to any"
+        exit 1
+      fi
+
+      echo "stopping driver $driver, expecting a shutdown at: $(echo "$executors" | tr '\n' ' ')"
+
+      log_pids=""
+      for pod in $driver $executors; do
+        kubectl -n "$NAMESPACE" logs -f "$pod" -c spark >"$capture/$pod.log" 2>&1 &
+        log_pids="$log_pids $!"
+      done
+      sleep 2
+
+      kubectl -n "$NAMESPACE" delete pod "$driver" --wait=false
+
+      start=$(date +%s)
+      missing=$executors
+      while [ $(( $(date +%s) - start )) -lt "$budget_seconds" ]; do
+        missing=""
+        for pod in $executors; do
+          grep -q 'Driver commanded a shutdown' "$capture/$pod.log" || missing="$missing $pod"
+        done
+        if [ -z "$missing" ]; then
+          break
+        fi
+        sleep 1
+      done
+      elapsed=$(( $(date +%s) - start ))
+      sleep 1
+      kill $log_pids 2>/dev/null || true
+
+      if [ -n "$missing" ]; then
+        echo "FAIL: ${elapsed}s after the driver was stopped, these executors had not been asked"
+        echo "to shut down:$missing"
+        if grep -q 'Asking each executor to shut down' "$capture/$driver.log"; then
+          echo "The driver did ask, so the request was lost on the way to the executors:"
+          for pod in $missing; do
+            tail -3 "$capture/$pod.log"
+          done
+        else
+          echo "The driver ran no shutdown hook, so SIGTERM did not reach the driver JVM:"
+          tail -5 "$capture/$driver.log"
+        fi
+        exit 1
+      fi
+
+      rm -rf "$capture"
+      echo "OK: every executor was asked to shut down ${elapsed}s after the driver was stopped"

tests/templates/kuttl/signal-propagation/00-patch-ns.yaml.j2

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/signal-propagation/01-assert.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if test_scenario['values']['vector'] == 'true' %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/signal-propagation/01-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,12 @@
+{% if test_scenario['values']['vector'] == 'true' %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  # This test only asserts that the Vector agent is shut down again, not that the log events
+  # arrive anywhere, so it does not need a reachable aggregator. If one is provided it is used,
+  # otherwise the agent runs and logs connection errors, which does not affect the assertions.
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') or 'localhost:6123' }}
+{% endif %}

tests/templates/kuttl/signal-propagation/10-deploy-failing-spark-app.yaml.j2

@@ -0,0 +1,55 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: failing-job
+data:
+  # A job that starts up correctly, runs a real Spark action, and only then fails.
+  # This makes sure we are testing the exit code of a genuinely failing application
+  # and not a startup or submission error.
+  failing-job.py: |
+    from pyspark.sql import SparkSession
+
+    spark = SparkSession.builder.appName("failing-job").getOrCreate()
+
+    # forces an actual computation
+    print(f"count = {spark.range(1000).count()}")
+
+    raise RuntimeError("deliberate failure to test exit code propagation")
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: failure-propagation
+spec:
+{% if test_scenario['values']['vector'] == 'true' %}
+  vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  sparkImage:
+{% if test_scenario['values']['spark'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['spark'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['spark'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  mode: cluster
+  mainApplicationFile: "local:///stackable/spark/jobs/failing-job.py"
+  driver:
+    config:
+      logging:
+        enableVectorAgent: {{ test_scenario['values']['vector'] }}
+      volumeMounts:
+        - name: script
+          mountPath: /stackable/spark/jobs
+  # The executors only run the tasks of the range/count job, they never read the script,
+  # so they do not need the volume mounted.
+  executor:
+    replicas: 1
+    config:
+      logging:
+        enableVectorAgent: {{ test_scenario['values']['vector'] }}
+  volumes:
+    - name: script
+      configMap:
+        name: failing-job