You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/09/29 18:24:42 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-59876] Suspend and requeue a running `SparkCluster` on Kueue `PodsReadyTimeout`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new a5984d4c [SPARK-59876] Suspend and requeue a running `SparkCluster` on Kueue `PodsReadyTimeout`
a5984d4c is described below
commit a5984d4cc22e5cd3ceb6010dc5192bf125bf0dfd
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Sep 29 11:24:34 2026 -0700
[SPARK-59876] Suspend and requeue a running `SparkCluster` on Kueue `PodsReadyTimeout`
### What changes were proposed in this pull request?
This PR aims to handle a Kueue `PodsReadyTimeout` eviction of a running `SparkCluster` like any other eviction, instead of only reporting it by the `KueueEvictionIgnored` warning event.
- The cluster enters `Suspended`, releases its master and workers, and is queued again with a new `Workload` after Kueue's requeue backoff, even if its pods are ready by now, like the Kueue built-in integrations.
- `KueueEvictionIgnored` is published only for a `SparkApplication`.
- A cluster whose pods are stuck terminating releases its `Workload` when the requeue backoff elapses, instead of after the 30-minute hold interval.
- The Kueue E2E test runs with a shortened requeue backoff (10 seconds).
### Why are the changes needed?
SPARK-59803 ignored a `PodsReadyTimeout` of a running `SparkCluster` only because the operator did not record `PodsReady` yet, which SPARK-59846 added. An ignored eviction is never released, so a preempting workload waits for its quota forever, and with `blockAdmission`, a cluster whose pods never get ready holds back every other workload.
### Does this PR introduce _any_ user-facing change?
No for the released versions, because the Kueue integration is not released yet in [1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0) (2026-07-23).
### How was this patch tested?
Pass the CIs with the newly added test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #910 from dongjoon-hyun/SPARK-59876.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.github/workflows/build_and_test.yml | 9 +-
docs/architecture.md | 4 +-
docs/configuration.md | 4 +-
docs/spark_custom_resources.md | 17 ++-
.../k8s/operator/kueue/KueueWorkloadUtils.java | 27 +----
.../reconcilesteps/ClusterSuspendStep.java | 37 +++----
.../k8s/operator/kueue/KueueWorkloadUtilsTest.java | 43 +++-----
.../reconcilesteps/ClusterSuspendStepTest.java | 119 ++++++++++++++-------
tests/e2e/kueue/chainsaw-test.yaml | 34 +++++-
9 files changed, 165 insertions(+), 129 deletions(-)
diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml
index a5a76d9b..d6fb780c 100644
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -204,12 +204,13 @@ jobs:
if: matrix.mode == 'kueue'
run: |
kubectl apply --server-side -f https://github.com/kubernetes-sigs/kueue/releases/download/${{ env.KUEUE_VERSION }}/manifests.yaml
- # Kueue enables waitForPodsReady by default with a 30 minute timeout. Shorten it, so that a
- # Workload whose pods are not ready is evicted within the test. A Kueue version which sets
- # it already fails here, rather than with a duplicate key which Kueue rejects on start.
+ # Kueue enables waitForPodsReady by default with a 30 minute timeout and a 60 second
+ # requeue backoff. Shorten them, so that a Workload whose pods are not ready is evicted
+ # within the test and queued again sooner. A Kueue version which sets waitForPodsReady
+ # already fails here, rather than with a duplicate key which Kueue rejects on start.
set -o pipefail
kubectl get configmap kueue-manager-config -n kueue-system -o json \
- | jq '.data["controller_manager_config.yaml"] |= if ("\n" + .) | contains("\nwaitForPodsReady:") then error("waitForPodsReady is configured already") else . + "\nwaitForPodsReady:\n timeout: 2m\n" end' \
+ | jq '.data["controller_manager_config.yaml"] |= if ("\n" + .) | contains("\nwaitForPodsReady:") then error("waitForPodsReady is configured already") else . + "\nwaitForPodsReady:\n timeout: 2m\n requeuingStrategy:\n backoffBaseSeconds: 10\n" end' \
| kubectl replace -f -
kubectl rollout restart deploy/kueue-controller-manager -n kueue-system
# Kueue installs webhooks for pods, deployments and statefulsets with failurePolicy=Fail,
diff --git a/docs/architecture.md b/docs/architecture.md
index 1594b741..6b611286 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -213,8 +213,8 @@ stateDiagram-v2
* Spark clusters are expected to be always running after submitted, until they are suspended by
[`spec.suspend`](spark_custom_resources.md#suspend). A cluster queued by
[Kueue](spark_custom_resources.md#kueue) is suspended as well when Kueue evicts its `Workload`,
- and moves to `Submitted` once its master and workers are released, or once its `Workload` is
- reactivated if it was deactivated.
+ and moves to `Submitted` once its master and workers are released and the requeue backoff of its
+ `Workload` elapsed, or once its `Workload` is reactivated if it was deactivated.
* Apart from `spec.suspend` and a Kueue eviction, a cluster leaves `RunningHealthy`, `Suspended` or
`Failed` only when its custom resource is deleted. At that point, the K8s resources created for the cluster are
garbage collected through their `ownerReference` to the `SparkCluster` custom resource.
diff --git a/docs/configuration.md b/docs/configuration.md
index 3babb026..526d491a 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -107,9 +107,9 @@ In addition, the operator publishes the following `Warning` events.
| `ClusterRequestFailed` | Applying the Services, StatefulSets, NetworkPolicy, HorizontalPodAutoscaler or PodDisruptionBudget of a `SparkCluster` fails in a way which may yet succeed, such as a throttled request or an internal server error from an admission webhook, so it is retried with the default interval instead of failing the cluster. Transport-level errors are skipped. |
| `SuspendReleaseFailed` | Deleting the master and worker StatefulSets, the HorizontalPodAutoscaler or PodDisruptionBudget of the workers, or the Kueue `Workload` of a `SparkCluster` suspended while running fails, or its pods cannot be listed. A failure which may clear on its own, such as a timeout, is skipped. Every failure is retried with the default interval. |
| `KueueResourceFlavorReadFailed` | Reading the `ResourceFlavor`s which Kueue assigned to an admitted `Workload` fails, so the node selector and the tolerations of the flavors cannot be applied. Transport-level errors are skipped and retried every 5 seconds, while a persistent failure is retried with the default interval. |
-| `KueuePodsReadyUpdateFailed` | Recording the `PodsReady` condition on the Kueue `Workload` of a running `SparkApplication` or `SparkCluster` fails, e.g. without the permission for the `workloads/status` subresource, so Kueue may evict it by a `PodsReadyTimeout`. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure with the default interval, while a `SparkApplication` retries any failure with its next reconciliation, so [...]
+| `KueuePodsReadyUpdateFailed` | Recording the `PodsReady` condition on the Kueue `Workload` of a running `SparkApplication` or `SparkCluster` fails, e.g. without the permission for the `workloads/status` subresource, so Kueue may evict it by a `PodsReadyTimeout`, which releases a running `SparkCluster` and queues it again even if its master and workers are ready. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure with [...]
| `KueueDisabled` | A `SparkApplication` or `SparkCluster` labeled with `kueue.x-k8s.io/queue-name` is not queued, since `spark.kubernetes.operator.kueue.enabled` is disabled, so the driver (or master and worker) is requested without the Kueue admission. |
-| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a `SparkApplication` whose driver is requested and whose attempt has not stopped, or evicted the `Workload` of a running `SparkCluster` by a `PodsReadyTimeout`. The operator does not act on it, so the driver and executors (or master and workers) keep running. An evicted `Workload` keeps holding its quota, while a deactivated one no longer counts against it. It is republished on every reconciliation while it lasts, [...]
+| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a `SparkApplication` whose driver is requested and whose attempt has not stopped. The operator does not act on it, so the driver and executors keep running. An evicted `Workload` keeps holding its quota, while a deactivated one no longer counts against it. It is republished on every reconciliation while it lasts, subject to [`minIntervalSeconds`](#event-frequency). See [Kueue](spark_custom_resources.md#kueue). |
For a resource held by [`spec.suspend`](spark_custom_resources.md#suspend) or queued by
[Kueue](spark_custom_resources.md#kueue), the operator also publishes the following `Normal`
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index fe296f4b..c549bac8 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -739,7 +739,10 @@ spec:
evicted `Workload` until then, so the operator then deletes it, and the resource is queued again
with a new `Workload`. Unlike Kueue built-in integrations, which keep the evicted
`Workload`, the new `Workload` is ordered in its queue like a newly created one, and it starts
- over the retry counts of its admission checks.
+ over the retry counts of its admission checks and the requeue count of its `PodsReadyTimeout`s.
+ So their backoff does not grow with each eviction, e.g. it stays at
+ `requeuingStrategy.backoffBaseSeconds` (60 seconds by default) for a `PodsReadyTimeout`, and a
+ `requeuingStrategy.backoffLimitCount` above 0 never deactivates the `Workload`.
* A running `SparkCluster` whose `Workload` Kueue evicts, e.g. to preempt it for a workload of a
higher priority or since its `ClusterQueue` is stopped, is released like on `spec.suspend`
without setting it. The cluster enters `Suspended` with a message giving the reason of the
@@ -764,10 +767,14 @@ spec:
the pods are ready. Unlike Kueue built-in integrations, the condition stays `True` when a pod is
lost later, e.g. an executor which Spark replaces, so the `recoveryTimeout` does not apply.
* A `PodsReadyTimeout` of a running `SparkCluster`, whose master or workers are not ready in time,
- is not acted on yet, and the `KueueEvictionIgnored` [event](configuration.md#kubernetes-events)
- is published instead. Kueue keeps counting the quota of such a `Workload`, so suspend or delete
- the cluster to release it. The `PodsReady` condition is still recorded once the master and
- workers are ready, so that `blockAdmission` admits the next workload.
+ releases the cluster like any other eviction above, even if they are ready by now, as Kueue
+ built-in integrations do. So does a failure to record the `PodsReady` condition, e.g. without the
+ permission for the `workloads/status` subresource, which the `KueuePodsReadyUpdateFailed`
+ [event](configuration.md#kubernetes-events) reports. As the backoff does not grow, see above, a
+ cluster whose master or workers never get ready, e.g. since they cannot be scheduled, or whose
+ condition cannot be recorded, is admitted and evicted again and again until it is fixed,
+ suspended or deleted. With `blockAdmission`, it holds back every other workload from each
+ admission until its `Workload` is deleted after the backoff.
* Preemption is not honored yet after the driver of a `SparkApplication` is requested, and neither
is a `PodsReadyTimeout`, e.g. of an application which runs with fewer executors than
`spark.executor.instances`, or a deactivation. The operator checks the admission only before it
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
index eaaf9e61..50099eb5 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java
@@ -86,9 +86,6 @@ public final class KueueWorkloadUtils {
/** Type of the Kueue condition which marks an evicted Workload. */
private static final String CONDITION_EVICTED = "Evicted";
- /** Reason of the eviction of a Workload whose pods are not ready within the Kueue timeout. */
- private static final String EVICTED_BY_PODS_READY_TIMEOUT = "PodsReadyTimeout";
-
/**
* Type of the Kueue condition which reports that the pods of a Workload are ready. It has to
* match the type which {@link WorkloadStatus#isPodsReady} checks, since that guards the
@@ -699,23 +696,6 @@ public final class KueueWorkloadUtils {
.findFirst();
}
- /**
- * Checks whether the running resources of the given evicted Workload are kept rather than
- * released. That is the case for a `PodsReadyTimeout` of an active Workload only, which is not
- * acted on yet, while Kueue keeps counting the quota of the Workload. Any other eviction releases
- * the resources, e.g. to preempt them or since the ClusterQueue is stopped. That includes a
- * deactivation, since Kueue stops counting the quota of a deactivated Workload right away.
- *
- * @param workload The Workload to check.
- * @return True if the Workload is evicted and its resources keep running.
- */
- public static boolean isKeptOnEviction(final Workload workload) {
- Optional<Condition> eviction = findEviction(workload);
- return eviction.isPresent()
- && !isDeactivated(workload)
- && EVICTED_BY_PODS_READY_TIMEOUT.equals(eviction.get().getReason());
- }
-
/**
* Checks whether the given Workload is deactivated by `spec.active` set to false, e.g. by a user
* or by Kueue after too many retries. Kueue neither counts nor admits it until it is reactivated.
@@ -729,9 +709,10 @@ public final class KueueWorkloadUtils {
/**
* Returns how long Kueue still holds back the given Workload after an eviction, e.g. for the
- * backoff of an admission check which asked for a retry. Deleting the Workload drops the backoff
- * with it, so an evicted Workload is kept until then. The retry count which Kueue records with
- * the admission checks is lost anyway, since a new Workload starts over.
+ * backoff of a `PodsReadyTimeout` or of an admission check which asked for a retry. Deleting the
+ * Workload drops the backoff with it, so an evicted Workload is kept until then. The requeue
+ * count and the retry counts of the admission checks which Kueue records are lost anyway, since
+ * a new Workload starts over, so the backoff of the next eviction does not grow.
*
* @param workload The Workload to check.
* @return The time until `status.requeueState.requeueAt`, or zero if it is not in the future.
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
index 5ae56e90..91818074 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
@@ -124,7 +124,13 @@ public final class ClusterSuspendStep extends ClusterReconcileStep {
Optional<ReconcileProgress> waiting =
releaseResources(context, statusRecorder, keepWorkload.isEmpty());
if (waiting.isPresent()) {
- return waiting.get();
+ // Pods which are stuck terminating are looked at again only as often as a suspended
+ // cluster otherwise, while the Workload which they no longer hold is released once its
+ // backoff elapses, when nothing else reconciles the cluster
+ return keepWorkload
+ .filter(backoff -> backoff.compareTo(waiting.get().getRequeueAfterDuration()) < 0)
+ .map(ReconcileProgress::completeAndRequeueAfter)
+ .orElse(waiting.get());
}
if (keepWorkload.isPresent()) {
return completeAndRequeueAfter(keepWorkload.get());
@@ -153,12 +159,11 @@ public final class ClusterSuspendStep extends ClusterReconcileStep {
* Suspends a running cluster whose Kueue Workload is evicted, so that its master and workers and
* then its Workload are released like on spec.suspend. Kueue keeps the quota of an evicted
* Workload until then. As spec.suspend is not set, the cluster is queued again once everything is
- * released, see {@link #keepKueueWorkload} for when that waits. An eviction which {@link
- * KueueWorkloadUtils#isKeptOnEviction} is only reported, and the cluster keeps running. A cluster
- * which is not suspended records the `PodsReady` condition on the Workload once the master and
- * workers are ready, since RunningHealthy does not wait for them, see {@link
- * KueueWorkloadUtils#recordPodsReady}. That includes a kept eviction: with `blockAdmission`, a
- * Workload which holds quota without the condition holds back every other workload.
+ * released, see {@link #keepKueueWorkload} for when that waits. Like the built-in Kueue
+ * integrations, which stop the job on any eviction, a `PodsReadyTimeout` suspends the cluster as
+ * well, even if its master and workers are ready by now. A cluster which is not suspended records
+ * the `PodsReady` condition on the Workload once the master and workers are ready, since
+ * RunningHealthy does not wait for them, see {@link KueueWorkloadUtils#recordPodsReady}.
*
* @param context The SparkClusterContext for the cluster.
* @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
@@ -166,26 +171,12 @@ public final class ClusterSuspendStep extends ClusterReconcileStep {
*/
private ReconcileProgress suspendOnKueueEviction(
SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
- Optional<Workload> cached = context.getCachedKueueWorkload();
- Optional<Condition> eviction = cached.flatMap(KueueWorkloadUtils::findEviction);
+ Optional<Condition> eviction =
+ context.getCachedKueueWorkload().flatMap(KueueWorkloadUtils::findEviction);
if (eviction.isEmpty()) {
return KueueWorkloadUtils.recordPodsReady(context).orElse(proceed());
}
- Workload workload = cached.get();
String cause = eviction.get().getReason() + ": " + eviction.get().getMessage();
- if (KueueWorkloadUtils.isKeptOnEviction(workload)) {
- EventUtils.warn(
- context.getEventRecorder(),
- EventUtils.REASON_KUEUE_EVICTION_IGNORED,
- "Kueue evicted Workload "
- + workload.getMetadata().getName()
- + " ("
- + cause
- + "), which the operator does not act on, so the master and workers keep running "
- + "and holding its quota. Set spec.suspend to true or delete the cluster to release "
- + "them.");
- return KueueWorkloadUtils.recordPodsReady(context).orElse(proceed());
- }
log.info("Kueue evicted the Workload of the cluster ({}), suspending it.", cause);
return appendStateAndImmediateRequeue(
context,
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtilsTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtilsTest.java
index 6228ab26..662308ec 100644
--- a/spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtilsTest.java
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtilsTest.java
@@ -382,35 +382,6 @@ class KueueWorkloadUtilsTest {
Assertions.assertTrue(KueueWorkloadUtils.isDeactivated(workload));
}
- @Test
- void evictionKeepingTheResourcesIsDetected() {
- Workload admitted = workload("owner-uid-1", 1);
- admitted.setStatus(status("Admitted", "True"));
- Assertions.assertFalse(KueueWorkloadUtils.isKeptOnEviction(admitted));
-
- for (String reason :
- List.of(
- "Preempted",
- "NodeFailures",
- "ClusterQueueStopped",
- "LocalQueueStopped",
- "AdmissionCheck",
- "FlavorMigration",
- "Deactivated")) {
- Workload evicted = workload("owner-uid-1", 1);
- evicted.setStatus(evictedStatus(reason));
- Assertions.assertFalse(KueueWorkloadUtils.isKeptOnEviction(evicted), reason);
- }
-
- // The pods of a running cluster which are not ready in time are not released yet
- Workload podsReadyTimeout = workload("owner-uid-1", 1);
- podsReadyTimeout.setStatus(evictedStatus("PodsReadyTimeout"));
- Assertions.assertTrue(KueueWorkloadUtils.isKeptOnEviction(podsReadyTimeout));
- // Unless it is deactivated, e.g. after too many retries, since Kueue stops counting its quota
- podsReadyTimeout.getSpec().setActive(false);
- Assertions.assertFalse(KueueWorkloadUtils.isKeptOnEviction(podsReadyTimeout));
- }
-
@ParameterizedTest
@ValueSource(strings = {"Preempted", "PodsReadyTimeout"})
void evictedWorkloadIsRecreated(String reason) {
@@ -763,6 +734,20 @@ class KueueWorkloadUtilsTest {
Assertions.assertTrue(workload.getStatus().isAdmitted());
}
+ @Test
+ void podsReadyIsRecordedOnAnEvictedWorkload() {
+ KueueWorkloadUtils.requestAdmission(kubernetesClient, workloadWithDriver());
+ admit(evictedStatus("PodsReadyTimeout"));
+
+ // A running application is not stopped on an eviction, so that with blockAdmission, it holds
+ // back other workloads only until its pods are ready
+ Assertions.assertEquals(
+ Optional.empty(), KueueWorkloadUtils.recordPodsReady(readyPodsContext(1, 2)));
+
+ Assertions.assertEquals("True", findCondition(getWorkload(), "PodsReady").getStatus());
+ Assertions.assertTrue(KueueWorkloadUtils.findEviction(getWorkload()).isPresent());
+ }
+
@Test
void podsReadyIsRecordedOnce() {
KueueWorkloadUtils.requestAdmission(kubernetesClient, workloadWithDriver());
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStepTest.java
index da356fae..278dc098 100644
--- a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStepTest.java
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStepTest.java
@@ -504,14 +504,22 @@ class ClusterSuspendStepTest {
}
@ParameterizedTest
- @CsvSource({"Preempted, true", "Deactivated, false", "Deactivated, true"})
+ @CsvSource({
+ "Preempted, true",
+ "PodsReadyTimeout, true",
+ "Deactivated, false",
+ "Deactivated, true"
+ })
void evictedRunningClusterEntersSuspendedBeforeReleasingResources(String reason, boolean active) {
// A deactivated Workload no longer counts against the quota, and a reactivated one is queued
- // again, so both release the master and workers like a preemption
+ // again, so both release the master and workers like a preemption. So does a PodsReadyTimeout
+ // like the built-in Kueue integrations, even if the master and workers are ready by now
SparkCluster cluster = buildKueueCluster(ClusterStateSummary.RunningHealthy, false);
stubContext(cluster);
createRunningCluster();
createWorkload(cluster, evictedStatus(reason), active);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+ stubReadyPods(1L, 1L);
Assertions.assertEquals(
ReconcileProgress.completeAndImmediateRequeue(),
@@ -527,6 +535,10 @@ class ClusterSuspendStepTest {
Assertions.assertNotNull(get(masterStatefulSetSpec));
Assertions.assertNotNull(get(workerStatefulSetSpec));
Assertions.assertNotNull(getWorkload());
+ // Nothing is recorded on the Workload which is about to be released, and nothing is reported
+ // but the Suspended state
+ Assertions.assertFalse(getWorkload().getStatus().isPodsReady());
+ verifyNoInteractions(eventRecorder);
}
@ParameterizedTest
@@ -624,6 +636,73 @@ class ClusterSuspendStepTest {
verifyNoInteractions(recorder);
}
+ @Test
+ void evictedWorkloadIsReleasedOnceTheRequeueBackoffElapsesWhilePodsAreStuck() {
+ SparkCluster cluster = buildEvictedCluster(false);
+ KubernetesClient client = spy(kubernetesClient);
+ stubStuckPods(client);
+ stubContext(cluster, client);
+ WorkloadStatus status = evictedStatus("PodsReadyTimeout");
+ status.setRequeueState(
+ RequeueState.builder()
+ .count(1)
+ .requeueAt(Instant.now().plusSeconds(60).toString())
+ .build());
+ createWorkload(cluster, status);
+
+ ReconcileProgress progress = new ClusterSuspendStep().reconcile(mockContext, recorder);
+
+ // Rather than after the hold interval of the stuck pods, which would keep the quota for up to
+ // 30 minutes past the backoff
+ Assertions.assertTrue(progress.isRequeue());
+ Assertions.assertTrue(
+ progress.getRequeueAfterDuration().compareTo(Duration.ofSeconds(60)) <= 0,
+ progress.getRequeueAfterDuration().toString());
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @Test
+ void failedReleaseIsRetriedWhileDeactivatedWorkloadIsKept() {
+ SparkCluster cluster = buildEvictedCluster(false);
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ when(mockClient.apps())
+ .thenThrow(new KubernetesClientException("Service Unavailable", 503, null));
+ stubContext(cluster, mockClient);
+ createWorkload(cluster, evictedStatus("Deactivated"), false);
+
+ // Rather than after the hold interval of the deactivated Workload, which would keep the master
+ // and workers running for up to 30 minutes outside of the quota
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new ClusterSuspendStep().reconcile(mockContext, recorder));
+ }
+
+ @Test
+ void podReleaseDeadlineIsNotPushedBackByALongerBackoff() {
+ SparkCluster cluster = buildEvictedCluster(false);
+ KubernetesClient client = spy(kubernetesClient);
+ stubPodList(
+ client,
+ terminatingPod(
+ "cluster1-master-0", Constants.LABEL_SPARK_ROLE_MASTER_VALUE, Duration.ofSeconds(270)));
+ stubContext(cluster, client);
+ WorkloadStatus status = evictedStatus("PodsReadyTimeout");
+ status.setRequeueState(
+ RequeueState.builder()
+ .count(1)
+ .requeueAt(Instant.now().plusSeconds(600).toString())
+ .build());
+ createWorkload(cluster, status);
+
+ ReconcileProgress progress = new ClusterSuspendStep().reconcile(mockContext, recorder);
+
+ // The wait is still requeued to end when the pod has been terminating for the timeout, rather
+ // than once the backoff of the Workload elapses, since the pod may send no more events
+ Assertions.assertTrue(
+ progress.getRequeueAfterDuration().compareTo(Duration.ofSeconds(30)) <= 0,
+ progress.getRequeueAfterDuration().toString());
+ }
+
@Test
void suspendingEvictedClusterIsRecorded() {
SparkCluster cluster = buildEvictedCluster(true);
@@ -737,42 +816,6 @@ class ClusterSuspendStepTest {
verifyNoInteractions(recorder);
}
- @Test
- void podsReadyIsRecordedAfterAPodsReadyTimeout() {
- SparkCluster cluster = buildKueueCluster(ClusterStateSummary.RunningHealthy, false);
- stubContext(cluster);
- createWorkload(cluster, evictedStatus("PodsReadyTimeout"));
- when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
- stubReadyPods(1L, 1L);
-
- Assertions.assertEquals(
- ReconcileProgress.proceed(), new ClusterSuspendStep().reconcile(mockContext, recorder));
-
- // Kueue keeps counting the quota of the Workload, so that with blockAdmission, it would hold
- // back every other workload without the condition
- Assertions.assertTrue(getWorkload().getStatus().isPodsReady());
- verifyNoInteractions(recorder);
- }
-
- @Test
- void podsReadyTimeoutIsReportedAndIgnored() {
- SparkCluster cluster = buildKueueCluster(ClusterStateSummary.RunningHealthy, false);
- stubContext(cluster);
- createWorkload(cluster, evictedStatus("PodsReadyTimeout"));
- when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
-
- // Kueue keeps counting its quota, while the master and workers are not released yet
- Assertions.assertEquals(
- ReconcileProgress.proceed(), new ClusterSuspendStep().reconcile(mockContext, recorder));
-
- Assertions.assertNotNull(getWorkload());
- ArgumentCaptor<EventRecord> event = ArgumentCaptor.forClass(EventRecord.class);
- verify(eventRecorder).record(event.capture());
- Assertions.assertEquals(EventType.WARNING, event.getValue().type());
- Assertions.assertEquals(EventUtils.REASON_KUEUE_EVICTION_IGNORED, event.getValue().reason());
- verifyNoInteractions(recorder);
- }
-
@Test
void otherStatesProceed() {
for (ClusterStateSummary summary :
diff --git a/tests/e2e/kueue/chainsaw-test.yaml b/tests/e2e/kueue/chainsaw-test.yaml
index b18e2b1d..477e3566 100644
--- a/tests/e2e/kueue/chainsaw-test.yaml
+++ b/tests/e2e/kueue/chainsaw-test.yaml
@@ -903,7 +903,9 @@ spec:
namespace: default
status:
(conditions[?type == 'Evicted' && status == 'True' && reason == 'PodsReadyTimeout'] | length(@)): 1
- # A running cluster is not released on it yet
+ # Like on a preemption, the operator releases the master and worker first and then the Workload
+ # once its requeue backoff, which the CI shortens to 10 seconds, elapses, and the cluster is
+ # queued again.
- assert:
timeout: 60s
resource:
@@ -914,8 +916,21 @@ spec:
involvedObject:
kind: SparkCluster
name: spark-cluster-kueue-pods-not-ready-test
- type: Warning
- reason: KueueEvictionIgnored
+ reason: Suspended
+ (contains(message || '', 'evicted by Kueue')): true
+ (contains(message || '', 'PodsReadyTimeout')): true
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: v1
+ kind: Event
+ metadata:
+ namespace: default
+ involvedObject:
+ kind: SparkCluster
+ name: spark-cluster-kueue-pods-not-ready-test
+ reason: Submitted
+ (contains(message || '', 'after its eviction')): true
# The cluster whose master and worker are ready is not evicted
- assert:
timeout: 30s
@@ -927,6 +942,19 @@ spec:
namespace: default
status:
(conditions[?type == 'Evicted' && status == 'True'] | length(@)): 0
+ # Nor was it evicted before, which the check above misses once the evicted Workload is replaced
+ # by a new one of the same name
+ - error:
+ timeout: 10s
+ resource:
+ apiVersion: v1
+ kind: Event
+ metadata:
+ namespace: default
+ involvedObject:
+ kind: SparkCluster
+ name: spark-cluster-kueue-pods-ready-test
+ reason: Suspended
catch:
- describe:
apiVersion: spark.apache.org/v1
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org