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 f83312e7 [SPARK-60052] Suspend and requeue a running
`SparkApplication` on Kueue eviction
f83312e7 is described below
commit f83312e78641ee6749997a93955d6caf1f6f4c70
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 14:05:32 2026 -0700
[SPARK-60052] Suspend and requeue a running `SparkApplication` on Kueue
eviction
### What changes were proposed in this pull request?
This PR aims to suspend a running `SparkApplication` whose Kueue `Workload`
is evicted or deactivated, and to queue it again with a new attempt, like a
running `SparkCluster`.
- `AppSuspendStep` moves such an application to `Suspended` and releases
its driver and executors. Then it deletes the `Workload` after the requeue
backoff, or keeps a deactivated one until it is reactivated.
- The new attempt starts with `ApplicationStatus#resume`, so it does not
count as a restart.
- A `PodsReadyTimeout` is handled like any other eviction, and
`spec.suspend` takes precedence.
- `AppKueueEvictionStep` and the `KueueEvictionIgnored` event are removed.
### Why are the changes needed?
Currently, the driver and executors keep running after an eviction. So
preemption, cohort quota reclamation and `waitForPodsReady` do not work, and a
deactivated `Workload` over-commits the quota.
### 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 #952 from dongjoon-hyun/SPARK-60052.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/architecture.md | 9 +-
docs/config_properties.md | 2 +-
docs/configuration.md | 5 +-
docs/spark_custom_resources.md | 32 +-
.../org/apache/spark/k8s/operator/Constants.java | 16 +
.../operator/status/ApplicationStateSummary.java | 8 +-
.../k8s/operator/config/SparkOperatorConf.java | 21 +-
.../k8s/operator/kueue/KueueWorkloadUtils.java | 2 +-
.../operator/reconciler/SparkAppReconciler.java | 3 -
.../reconcilesteps/AppKueueEvictionStep.java | 84 ------
.../reconcilesteps/AppReconcileStep.java | 4 +-
.../reconciler/reconcilesteps/AppSuspendStep.java | 116 +++++++-
.../spark/k8s/operator/utils/EventUtils.java | 6 -
.../reconciler/SparkAppReconcilerTest.java | 20 +-
.../reconcilesteps/AppKueueEvictionStepTest.java | 150 ----------
.../reconcilesteps/AppSuspendStepTest.java | 327 +++++++++++++++++++++
tests/e2e/kueue/chainsaw-test.yaml | 221 ++++++++++----
tests/e2e/kueue/spark-example-preempted.yaml | 42 +++
18 files changed, 700 insertions(+), 368 deletions(-)
diff --git a/docs/architecture.md b/docs/architecture.md
index 7d4b70a1..bfccdc5a 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -138,8 +138,8 @@ stateDiagram-v2
RunningWithPartialCapacity --> Succeeded
RunningWithPartialCapacity --> Failed
- RunningHealthy --> Suspended : spec.suspend=true
- Suspended --> Submitted : spec.suspend=false, new attempt
+ RunningHealthy --> Suspended : spec.suspend=true or Kueue eviction
+ Suspended --> Submitted : spec.suspend=false or after Kueue eviction, new
attempt
state Failures {
SchedulingFailure
@@ -171,7 +171,10 @@ stateDiagram-v2
* Likewise, an application moves to `Suspended` from any of the states from
`DriverRequested` to
`RunningWithBelowThresholdExecutors` when
[`spec.suspend`](spark_custom_resources.md#suspend) is
set to `true`. Its driver and executors are released, and setting it back to
`false` starts a
- new attempt from `Submitted`, which does not count as a restart.
+ new attempt from `Submitted`, which does not count as a restart. An
application queued by
+ [Kueue](spark_custom_resources.md#kueue) is suspended as well when Kueue
evicts its `Workload`,
+ and starts such a new attempt once its driver and executors are released and
the requeue
+ backoff of its `Workload` elapsed, or once its `Workload` is reactivated if
it was deactivated.
* User may configure the app CR to time-out after given threshold of time if
it cannot reach healthy
state after given threshold. The timeout can be configured for different
lifecycle stages,
when driver starting, when driver becoming ready, and when requesting
executor pods.
diff --git a/docs/config_properties.md b/docs/config_properties.md
index 2a8285b9..7ededc17 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -53,7 +53,7 @@
| spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds | Integer
| 5 | false | Initial interval(in seconds) of retries on unhandled controller
errors. |
| spark.kubernetes.operator.reconciler.retry.intervalMultiplier | Double |
1.5 | false | Interval multiplier of retries on unhandled controller errors.
Setting this to 1 for linear retry. |
| spark.kubernetes.operator.reconciler.retry.maxIntervalSeconds | Integer |
-1 | false | Max interval(in seconds) of retries on unhandled controller
errors. Set to non-positive for unlimited. |
- | spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds |
Long | 1800 | true | Requeue interval (in seconds) at which a resource held by
spec.suspend is reconciled, so that its SuspendHeld event is republished. A
SparkApplication or SparkCluster suspended while running publishes no such
event, since its Suspended status says so, and uses this interval only to look
at its released resources again. The hold ends only when a user clears
spec.suspend, which arrives as a wat [...]
+ | spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds |
Long | 1800 | true | Requeue interval (in seconds) at which a resource held by
spec.suspend is reconciled, so that its SuspendHeld event is republished. A
SparkApplication or SparkCluster suspended while running publishes no such
event, since its Suspended status says so, and uses this interval only to look
at its released resources again. The hold ends only when a user clears
spec.suspend, which arrives as a wat [...]
| spark.kubernetes.operator.reconciler.terminationTimeoutSeconds | Integer |
30 | false | Grace period for operator shutdown before reconciliation threads
are killed. |
| spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled |
Boolean | true | true | When enabled, operator would trim state transition
history when a new attempt starts, keeping previous attempt summary only. It
also drops the history of a suspended SparkCluster when it is resumed. |
| spark.kubernetes.operator.watchedNamespaces | String | default | true |
Comma-separated list of namespaces that the operator would be watching for
Spark resources. If set to '*', operator would watch all namespaces. |
diff --git a/docs/configuration.md b/docs/configuration.md
index 115d0815..ece6014a 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -102,7 +102,7 @@ replacing its `message`. A repeat carrying the same
`message` may be paced by
The initial `Submitted` status of a new resource is not persisted to the API
server on its own, so
no event is published for it. A `Submitted` event is published only when a
`SparkApplication` or a
`SparkCluster` moves from `Suspended` back to `Submitted`, after
`spec.suspend` is set back to
-`false`, or after the eviction of a `SparkCluster` by Kueue.
+`false`, or after its eviction by Kueue.
In addition, the operator publishes the following `Warning` events.
@@ -115,9 +115,8 @@ 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 driver pod (`SparkApplication`), the
master and worker StatefulSets or the HorizontalPodAutoscaler or
PodDisruptionBudget of the workers (`SparkCluster`), or the Kueue `Workload` of
a resource 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`, 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 [...]
+| `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 it and queues it again, a
`SparkApplication` with a new attempt, even if its pods are ready.
Transport-level errors and conflicts are skipped. A `SparkCluster` retries them
every 5 seconds and a persistent failur [...]
| `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 neither
stopped nor been suspended. 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_re [...]
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 5ffcb9ac..97fd7e2c 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -829,18 +829,26 @@ spec:
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
- creates the resources, so after a later eviction the driver and executors
keep running, and
- until the attempt stops or is suspended, the `KueueEvictionIgnored`
- [event](configuration.md#kubernetes-events) is published instead by default.
An evicted
- `Workload` keeps its Kueue quota until the driver and executors are
released, so a workload
- which preempts the application waits until then, while a deactivated one no
longer counts
- against it. Set `spec.suspend` to `true`, as described in
[Suspend](#suspend), or delete the
- application to release its quota earlier. With `blockAdmission`, an
- application whose pods are not all ready holds back every other workload
until they are ready
- or its quota is released as described above.
+* A `SparkApplication` whose driver is requested, i.e. from `DriverRequested`
to
+ `RunningWithBelowThresholdExecutors`, is released like a running
`SparkCluster` above when Kueue
+ evicts its `Workload`, e.g. to preempt it or by a `PodsReadyTimeout`, or
deactivates it. The
+ application enters `Suspended` with a message giving the reason of the
eviction, and the
+ operator deletes its driver pod, which deletes its executor pods. Like for a
cluster, the
+ `Workload` is deleted once the pods are gone and its requeue backoff
elapsed, or kept until it is
+ reactivated if it was deactivated, and then the application starts a new
attempt from
+ `Submitted`, which is queued again with a new `Workload`. Like a resumed
one, the new attempt
+ runs the application again from scratch and does not count against the
restart limits, so it
+ starts even with `restartPolicy: Never`. An attempt whose driver has
completed or failed by then
+ ends as usual instead, and `spec.suspend` takes precedence, so an
application which is suspended
+ by it as well stays `Suspended` until it is resumed rather than being queued
again.
+* Unlike without Kueue, a `SparkApplication` whose driver and
`spark.executor.instances` executors
+ are not all ready within the `waitForPodsReady` timeout, e.g. since some
executors cannot be
+ scheduled, does not keep running with fewer executors, even if its
`applicationTolerations`
+ allow it, e.g. in `RunningWithPartialCapacity`. Its `Workload` requested the
quota for all of
+ them, so the `PodsReadyTimeout` releases the application as above, and it
runs again from
+ scratch with a new attempt, which waits for the quota of all of them again.
Like such a cluster,
+ an application whose executors never get ready is admitted and evicted again
and again, with a
+ new attempt each time, until it is fixed, suspended or deleted.
* To limit the execution time of a `SparkApplication`, the Spark native
`spark.driver.timeout` is
recommended instead of the `kueue.x-k8s.io/max-exec-time-seconds` label,
which is not copied to
the `Workload`. It requires
`spark.plugins=org.apache.spark.deploy.DriverTimeoutPlugin`, as in
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
index 7286a29e..c0ad3a05 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
@@ -246,6 +246,22 @@ public final class Constants {
"Application is resumed as spec.suspend is set to false, so a new
attempt starts from "
+ "scratch.";
+ /**
+ * Message indicating that the application is evicted by Kueue after its
driver was requested,
+ * which is followed by the reason and the message of the eviction. The
operator tells such an
+ * application apart from one suspended by spec.suspend by this prefix of
its persisted state
+ * message, so a changed wording no longer matches the applications
suspended before.
+ */
+ public static final String APP_EVICTED_MESSAGE =
+ "Application is evicted by Kueue, so its driver and executors are being
released before its "
+ + "Workload. It is queued again with a new attempt then, or once the
Workload is "
+ + "reactivated if it was deactivated.";
+
+ /** Message indicating that the application evicted by Kueue is submitted
again. */
+ public static final String APP_REQUEUED_MESSAGE =
+ "Application is submitted again after its eviction by Kueue, so a new
attempt starts from "
+ + "scratch.";
+
// Spark Cluster Messages
/** Message indicating a failure to request the Spark cluster from the
scheduler backend. */
public static final String CLUSTER_SCHEDULE_FAILURE_MESSAGE =
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
index 48186b6d..8f994109 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
@@ -112,10 +112,10 @@ public enum ApplicationStateSummary implements
BaseStateSummary {
TerminatedWithoutReleaseResources,
/**
- * The application is suspended by spec.suspend after its driver was
requested, so its driver
- * and executors are released until it is resumed with a new attempt from
Submitted. It is
- * declared last, after the terminated states, so that the ordinal ranges of
{@link
- * #isStarting()} and {@link #isStopping()} exclude it.
+ * The application is suspended by spec.suspend, or by the eviction of its
Kueue Workload, after
+ * its driver was requested, so its driver and executors are released until
it is resumed with a
+ * new attempt from Submitted. It is declared last, after the terminated
states, so that the
+ * ordinal ranges of {@link #isStarting()} and {@link #isStopping()} exclude
it.
*/
Suspended;
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index 2cb9ec3f..737c798b 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -268,13 +268,15 @@ public final class SparkOperatorConf {
* suspended while running publishes no such event, since its Suspended
status says so, and uses
* this interval only to look at its released resources again. The hold ends
only when a user
* clears {@code spec.suspend}, which arrives as a watch event and
reconciles right away, so
- * nothing waits for this interval. It is deliberately much coarser than
{@link
- * #RECONCILER_INTERVAL_SECONDS}, since a suspended resource has nothing to
observe while each
- * republish costs a read and a write on the API server. Keep it below the
{@code --event-ttl} of
- * the API server (one hour by default), or the event expires between
repeats and the hold stops
- * being visible, and keep {@link #KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS}
within the rest of
- * that TTL, that is at most the TTL minus this interval, or a repeat is
dropped and the next
- * one lands after the event expired.
+ * nothing waits for this interval. It also paces a SparkApplication or
SparkCluster which is
+ * suspended by a Kueue eviction and whose Workload is deactivated. That
hold ends when the
+ * Workload is reactivated, which arrives as a watch event as well. It is
deliberately much
+ * coarser than {@link #RECONCILER_INTERVAL_SECONDS}, since a suspended
resource has nothing to
+ * observe while each republish costs a read and a write on the API server.
Keep it below the
+ * {@code --event-ttl} of the API server (one hour by default), or the event
expires between
+ * repeats and the hold stops being visible, and keep {@link
+ * #KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS} within the rest of that TTL,
that is at most the TTL
+ * minus this interval, or a repeat is dropped and the next one lands after
the event expired.
*/
public static final ConfigOption<Long> SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS
=
ConfigOption.<Long>builder()
@@ -288,7 +290,10 @@ public final class SparkOperatorConf {
+ "says so, and uses this interval only to look at its
released resources "
+ "again. The hold ends only "
+ "when a user clears spec.suspend, which arrives as a watch
event and "
- + "reconciles right away, so nothing waits for this
interval. It is "
+ + "reconciles right away, so nothing waits for this
interval. It also paces a "
+ + "SparkApplication or SparkCluster which is suspended by a
Kueue eviction and "
+ + "whose Workload is deactivated. That hold ends when the
Workload is "
+ + "reactivated, which arrives as a watch event as well. It
is "
+ "deliberately much coarser than "
+ "spark.kubernetes.operator.reconciler.intervalSeconds,
since a suspended "
+ "resource has nothing to observe while each republish
costs a read and a "
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 482fdba1..b06cd5a1 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
@@ -923,7 +923,7 @@ public final class KueueWorkloadUtils {
* timeout of Kueue does not evict it, and with `blockAdmission`, other
workloads are admitted.
* Unlike the built-in Kueue integrations, the condition is not set back to
`False` when a pod is
* lost later, e.g. an executor which Spark replaces, since an eviction
would restart a running
- * cluster with its applications, and is not acted on for a running
application.
+ * cluster with its applications, or the attempt of a running application
from scratch.
*
* <p>The informer cache answers whether the condition is still missing, so
that a resource which
* reports it already, or whose pods are not ready yet, costs no request.
Like {@link
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
index 9e8cbf53..0b894045 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
@@ -55,7 +55,6 @@ import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverStartObserver
import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverTimeoutObserver;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppInitStep;
-import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppKueueEvictionStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppResourceObserveStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppRunningStep;
@@ -207,7 +206,6 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
switch (app.getStatus().getCurrentState().getCurrentStateSummary()) {
case Submitted, ScheduledToRestart -> steps.add(new AppInitStep());
case DriverRequested, DriverStarted -> {
- steps.add(new AppKueueEvictionStep());
steps.add(
new AppResourceObserveStep(
List.of(new AppDriverStartObserver(), new
AppDriverReadyObserver())));
@@ -220,7 +218,6 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
RunningHealthy,
RunningWithPartialCapacity,
RunningWithBelowThresholdExecutors -> {
- steps.add(new AppKueueEvictionStep());
steps.add(new AppRunningStep());
steps.add(new AppResourceObserveStep(List.of(new
AppDriverRunningObserver())));
steps.add(new AppResourceObserveStep(List.of(new
AppDriverTimeoutObserver())));
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
deleted file mode 100644
index 26439d41..00000000
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
-
-import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed;
-
-import java.util.Optional;
-
-import io.fabric8.kubernetes.api.model.Condition;
-
-import org.apache.spark.k8s.operator.context.SparkAppContext;
-import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
-import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
-import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
-import org.apache.spark.k8s.operator.utils.EventUtils;
-import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
-
-/**
- * Reports an eviction or a deactivation of the Kueue Workload of an
application whose driver is
- * requested, which the operator does not act on yet. Unlike before the driver
is requested, see
- * {@link AppInitStep}, the driver and executors keep running. The status does
not show it, so the
- * event is published on every reconciliation while it lasts, like the pending
event of a queued
- * application, and its message stays the same meanwhile, so that {@link
- *
org.apache.spark.k8s.operator.config.SparkOperatorConf#KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS}
- * paces the repeats.
- */
-public final class AppKueueEvictionStep extends AppReconcileStep {
- /**
- * Publishes the event if needed and always proceeds, so that the driver is
still observed.
- *
- * @param context The SparkAppContext for the application.
- * @param statusRecorder The SparkAppStatusRecorder for recording status
updates.
- * @return The ReconcileProgress to proceed with.
- */
- @Override
- public ReconcileProgress reconcile(
- SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
- Optional<Workload> workload = context.getCachedKueueWorkload();
- // Kueue marks a deactivated Workload as evicted too, but only once it
sees the deactivation
- Optional<Condition> eviction =
workload.flatMap(KueueWorkloadUtils::findEviction);
- boolean deactivated =
workload.map(KueueWorkloadUtils::isDeactivated).orElse(false);
- if (eviction.isEmpty() && !deactivated) {
- return proceed();
- }
- String name = workload.get().getMetadata().getName();
- String cause =
- eviction.map(c -> " (" + c.getReason() + ": " + c.getMessage() +
")").orElse("");
- EventUtils.warn(
- context.getEventRecorder(),
- EventUtils.REASON_KUEUE_EVICTION_IGNORED,
- deactivated
- ? "Kueue Workload "
- + name
- + " is deactivated"
- + cause
- + ", which the operator does not act on once the driver is
requested, so the "
- + "driver and executors keep running although Kueue no longer
counts them against "
- + "its quota."
- : "Kueue evicted Workload "
- + name
- + cause
- + ", which the operator does not act on once the driver is
requested, so the "
- + "driver and executors keep running and holding its quota
until they are "
- + "released. Set spec.suspend to true or delete the
application to release them.");
- return proceed();
- }
-}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
index caea2e58..7b4edf38 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
@@ -44,8 +44,8 @@ import
org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Basic reconcile step for application. */
@Log4j2
public abstract sealed class AppReconcileStep
- permits AppCleanUpStep, AppInitStep, AppKueueEvictionStep,
AppResourceObserveStep,
- AppRunningStep, AppSuspendStep, AppUnknownStateStep, AppValidateStep {
+ permits AppCleanUpStep, AppInitStep, AppResourceObserveStep,
AppRunningStep, AppSuspendStep,
+ AppUnknownStateStep, AppValidateStep {
/**
* Reconciles a specific step for a Spark application.
*
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
index 9737b743..b9ec67d8 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
@@ -19,6 +19,8 @@
package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
+import static org.apache.spark.k8s.operator.Constants.APP_EVICTED_MESSAGE;
+import static org.apache.spark.k8s.operator.Constants.APP_REQUEUED_MESSAGE;
import static org.apache.spark.k8s.operator.Constants.APP_RESUMED_MESSAGE;
import static org.apache.spark.k8s.operator.Constants.APP_SUSPENDED_MESSAGE;
import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_APPLICATION_NAME;
@@ -39,6 +41,7 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
+import io.fabric8.kubernetes.api.model.Condition;
import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -48,6 +51,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.context.SparkAppContext;
import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverRunningObserver;
import org.apache.spark.k8s.operator.spec.ApplicationTimeoutConfig;
@@ -58,8 +62,9 @@ import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/**
- * Suspends an application whose driver is requested by spec.suspend, and
resumes it. An
- * application suspended before its driver is requested is held by {@link
AppInitStep} instead.
+ * Suspends an application whose driver is requested by spec.suspend, or by
the eviction of its
+ * Kueue Workload, and resumes it. An application suspended before its driver
is requested is held
+ * by {@link AppInitStep} instead, which handles an eviction before then as
well.
*
* <p>An application enters Suspended before anything is released, and its
driver is released only
* in Suspended. Releasing it first would leave an application without its
driver in a running state
@@ -76,6 +81,12 @@ import
org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
* meanwhile ends as usual, rather than being suspended and run again from
scratch on resume. The
* observers read the informer cache, which may not show such a driver yet, so
the driver is
* observed once more on the API server before the application is suspended.
+ *
+ * <p>An application whose Kueue Workload is evicted, e.g. to preempt it, is
suspended the same way
+ * without spec.suspend, so that it is queued again with a new attempt once
everything is released.
+ * Like a resumed one, that attempt does not count against the restart limits,
since an eviction
+ * is a requeue rather than a failure. spec.suspend takes precedence, so that
an application which
+ * is suspended by both is held until it is resumed, rather than queued again.
*/
@Slf4j
public final class AppSuspendStep extends AppReconcileStep {
@@ -87,7 +98,7 @@ public final class AppSuspendStep extends AppReconcileStep {
/**
* Reconciles an application whose driver is requested, or which is
suspended, against
- * spec.suspend.
+ * spec.suspend and the eviction of its Kueue Workload.
*
* @param context The SparkAppContext for the application.
* @param statusRecorder The SparkAppStatusRecorder for recording status
updates.
@@ -99,7 +110,12 @@ public final class AppSuspendStep extends AppReconcileStep {
SparkApplication app = context.getResource();
boolean suspend = app.getSpec().isSuspend();
ApplicationStateSummary summary =
app.getStatus().getCurrentState().getCurrentStateSummary();
- if (DRIVER_REQUESTED_STATES.contains(summary) && suspend) {
+ if (DRIVER_REQUESTED_STATES.contains(summary)) {
+ Optional<String> message =
+ suspend ? Optional.of(APP_SUSPENDED_MESSAGE) :
findKueueEviction(context);
+ if (message.isEmpty()) {
+ return proceed();
+ }
Optional<ApplicationState> termination;
try {
termination = observeDriverTermination(context);
@@ -113,30 +129,104 @@ public final class AppSuspendStep extends
AppReconcileStep {
// A rejected update, e.g. of a state unknown to an older CRD, would be
rejected again right
// away, so it is retried with the default interval, not immediately.
return statusRecorder.appendNewStateAndPersist(
- context,
- new ApplicationState(ApplicationStateSummary.Suspended,
APP_SUSPENDED_MESSAGE))
+ context, new ApplicationState(ApplicationStateSummary.Suspended,
message.get()))
? completeAndImmediateRequeue()
: completeAndDefaultRequeue();
}
if (summary != ApplicationStateSummary.Suspended) {
return proceed();
}
- Optional<ReconcileProgress> waiting = releaseResources(context);
+ boolean evicted = isSuspendedByEviction(app);
+ if (evicted && suspend) {
+ // The application is held by spec.suspend from now on, so that its
status no longer says
+ // that it is queued again, and a later resume is reported as such.
+ return appendStateAndImmediateRequeue(
+ context,
+ statusRecorder,
+ new ApplicationState(ApplicationStateSummary.Suspended,
APP_SUSPENDED_MESSAGE));
+ }
+ Optional<Duration> keepWorkload = evicted ? keepKueueWorkload(context) :
Optional.empty();
+ Optional<ReconcileProgress> waiting = releaseResources(context,
keepWorkload.isEmpty());
if (waiting.isPresent()) {
return waiting.get();
}
+ if (keepWorkload.isPresent()) {
+ return completeAndRequeueAfter(keepWorkload.get());
+ }
if (suspend) {
return completeAndRequeueAfter(
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
}
+ String resumedMessage = evicted ? APP_REQUEUED_MESSAGE :
APP_RESUMED_MESSAGE;
return statusRecorder.persistStatus(
context,
app.getStatus()
- .resume(APP_RESUMED_MESSAGE,
TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue()))
+ .resume(resumedMessage,
TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue()))
? completeAndImmediateRequeue()
: completeAndDefaultRequeue();
}
+ /**
+ * Returns the message of the Suspended state of an application whose Kueue
Workload is evicted,
+ * so that its driver and executors and then its Workload are released like
on spec.suspend. Like
+ * the built-in Kueue integrations, which stop the job on any eviction, a
`PodsReadyTimeout`
+ * suspends the application as well, even if its driver and executors are
ready by now, or if its
+ * applicationTolerations let it run with fewer executors than requested.
Kueue marks a
+ * deactivated Workload as evicted, too.
+ *
+ * @param context The SparkAppContext for the application.
+ * @return The message which gives the reason of the eviction, or empty if
the Workload is not
+ * evicted.
+ */
+ private static Optional<String> findKueueEviction(SparkAppContext context) {
+ Optional<Condition> eviction =
+
context.getCachedKueueWorkload().flatMap(KueueWorkloadUtils::findEviction);
+ if (eviction.isEmpty()) {
+ return Optional.empty();
+ }
+ String cause = eviction.get().getReason() + ": " +
eviction.get().getMessage();
+ log.info("Kueue evicted the Workload of the application ({}), suspending
it.", cause);
+ return Optional.of(APP_EVICTED_MESSAGE + " " + cause);
+ }
+
+ /**
+ * Checks whether the application is suspended by the eviction of its Kueue
Workload rather than
+ * by spec.suspend. Unlike a suspended cluster, which may name its stuck
pods in a later state, a
+ * suspended application appends no other Suspended state, so its current
one says why.
+ *
+ * @param app The suspended SparkApplication.
+ * @return True if the application was suspended by an eviction and not held
by spec.suspend
+ * since.
+ */
+ private static boolean isSuspendedByEviction(SparkApplication app) {
+ String message = app.getStatus().getCurrentState().getMessage();
+ return message != null && message.startsWith(APP_EVICTED_MESSAGE);
+ }
+
+ /**
+ * Returns how long the Kueue Workload of an application which is suspended
without spec.suspend
+ * is kept after its driver and executors are released, or empty to release
it right away and
+ * queue the application again. A deactivated Workload is kept until it is
reactivated, which the
+ * Workload informer reconciles right away: Kueue no longer counts its
quota, and a new Workload
+ * of the application would come back active. An evicted Workload is kept
until the requeue
+ * backoff which Kueue records on it elapses, see {@link
+ * KueueWorkloadUtils#remainingRequeueBackoff}.
+ *
+ * @param context The SparkAppContext for the application.
+ * @return The requeue interval while the Workload is kept, or empty to
release it.
+ */
+ private static Optional<Duration> keepKueueWorkload(SparkAppContext context)
{
+ Optional<Workload> workload = context.getCachedKueueWorkload();
+ if (workload.isEmpty()) {
+ return Optional.empty();
+ }
+ if (KueueWorkloadUtils.isDeactivated(workload.get())) {
+ return
Optional.of(Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
+ }
+ Duration backoff =
KueueWorkloadUtils.remainingRequeueBackoff(workload.get());
+ return backoff.isZero() ? Optional.empty() : Optional.of(backoff);
+ }
+
/**
* Observes whether the driver of the attempt completed or failed, as the
API server knows it. A
* driver which is being deleted, e.g. one of an earlier attempt, is not
looked at, see {@link
@@ -181,10 +271,12 @@ public final class AppSuspendStep extends
AppReconcileStep {
* Suspended is a steady state which would not tell it apart from waiting
for the pods to go.
*
* @param context The SparkAppContext for the application.
+ * @param releaseWorkload Whether the Kueue Workload is released as well.
* @return Empty once everything is released, or the ReconcileProgress to
retry with while pods
* remain or a release failed.
*/
- private Optional<ReconcileProgress> releaseResources(SparkAppContext
context) {
+ private Optional<ReconcileProgress> releaseResources(
+ SparkAppContext context, boolean releaseWorkload) {
SparkApplication app = context.getResource();
ApplicationTimeoutConfig timeoutConfig =
app.getSpec().getApplicationTolerations().getApplicationTimeoutConfig();
@@ -223,8 +315,10 @@ public final class AppSuspendStep extends AppReconcileStep
{
? completeAndRequeueAfter(remaining)
: defaultRequeue);
}
- // A Workload admitted before the queue label was removed is released as
well.
- KueueWorkloadUtils.deleteWorkloadOf(client, app);
+ if (releaseWorkload) {
+ // A Workload admitted before the queue label was removed is released
as well.
+ KueueWorkloadUtils.deleteWorkloadOf(client, app);
+ }
} catch (KubernetesClientException e) {
log.warn("Failed to release the resources of the suspended application,
will retry.", e);
if (!ReconcilerUtils.isTransientError(e)) {
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
index 376fa3a3..bb0ecd39 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
@@ -98,12 +98,6 @@ public final class EventUtils {
*/
public static final String REASON_KUEUE_DISABLED = "KueueDisabled";
- /**
- * Reason for an event describing that Kueue evicted or deactivated the
Workload of a running
- * resource in a way which the operator does not act on, so the resource
keeps running.
- */
- public static final String REASON_KUEUE_EVICTION_IGNORED =
"KueueEvictionIgnored";
-
/** Maximum number of characters an event message may have, including the
ellipsis. */
static final int MAX_MESSAGE_LENGTH = 1024;
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
index fc6ad9b5..59989468 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
@@ -88,7 +88,6 @@ import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus;
import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep;
-import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppKueueEvictionStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppSuspendStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppValidateStep;
@@ -613,7 +612,8 @@ class SparkAppReconcilerTest {
} else {
boolean expected = DRIVER_REQUESTED_STATES.contains(summary);
// After the observers, so that a driver which completed meanwhile is
recorded rather than
- // suspended and run again when the application is resumed
+ // suspended, by spec.suspend or by a Kueue eviction, and run again
when the application is
+ // resumed. Before the driver is requested, AppInitStep handles both
itself.
assertEquals(
expected, steps.get(steps.size() - 1) instanceof AppSuspendStep,
summary.name());
assertEquals(
@@ -623,20 +623,4 @@ class SparkAppReconcilerTest {
}
}
}
-
- @Test
- void kueueEvictionIsReportedFromTheDriverRequestUntilTheAttemptStops() {
- for (ApplicationStateSummary summary : ApplicationStateSummary.values()) {
- app.setStatus(new ApplicationStatus().appendNewState(new
ApplicationState(summary, "")));
- List<AppReconcileStep> steps = reconciler.getReconcileSteps(app);
- boolean expected = DRIVER_REQUESTED_STATES.contains(summary);
- // Right after the validation and the clean up, before a state
transition ends the
- // reconciliation. Before the driver is requested, AppInitStep handles
the eviction itself.
- assertEquals(expected, steps.get(2) instanceof AppKueueEvictionStep,
summary.name());
- assertEquals(
- expected ? 1 : 0,
-
steps.stream().filter(AppKueueEvictionStep.class::isInstance).count(),
- summary.name());
- }
- }
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
deleted file mode 100644
index 810b656e..00000000
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoInteractions;
-import static org.mockito.Mockito.when;
-
-import java.util.List;
-import java.util.Optional;
-
-import io.fabric8.kubernetes.api.model.Condition;
-import io.fabric8.kubernetes.api.model.ConditionBuilder;
-import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
-import io.javaoperatorsdk.operator.api.event.EventRecord;
-import io.javaoperatorsdk.operator.api.event.EventType;
-import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
-import org.mockito.ArgumentCaptor;
-
-import org.apache.spark.k8s.operator.context.SparkAppContext;
-import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
-import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadSpec;
-import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus;
-import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
-import org.apache.spark.k8s.operator.utils.EventUtils;
-import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
-
-class AppKueueEvictionStepTest {
- private final SparkAppContext context = mock(SparkAppContext.class);
- private final SparkAppStatusRecorder recorder =
mock(SparkAppStatusRecorder.class);
- private final ResourceEventRecorder eventRecorder =
mock(ResourceEventRecorder.class);
-
- @BeforeEach
- void setUp() {
- when(context.getEventRecorder()).thenReturn(eventRecorder);
- }
-
- @Test
- void workloadWhichIsMissingOrAdmittedIsNotReported() {
- AppKueueEvictionStep step = new AppKueueEvictionStep();
- when(context.getCachedKueueWorkload()).thenReturn(Optional.empty());
- Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
-
-
when(context.getCachedKueueWorkload()).thenReturn(Optional.of(workload(true,
admitted())));
- Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
-
- verifyNoInteractions(eventRecorder, recorder);
- }
-
- @ParameterizedTest
- @ValueSource(strings = {"Preempted", "PodsReadyTimeout",
"ClusterQueueStopped"})
- void evictionIsReportedAndIgnored(String reason) {
- when(context.getCachedKueueWorkload())
- .thenReturn(Optional.of(workload(true, admitted(), evicted(reason))));
-
- // The driver is still observed, e.g. until it completes
- Assertions.assertEquals(
- ReconcileProgress.proceed(), new
AppKueueEvictionStep().reconcile(context, recorder));
-
- String message = captureMessages(1).get(0);
- Assertions.assertTrue(
- message.contains("(" + reason + ": " + reason + " by the test)"),
message);
- // Kueue keeps counting the quota of an evicted Workload until the driver
and executors are
- // released
- Assertions.assertTrue(message.contains("holding its quota"), message);
- Assertions.assertTrue(message.contains("Set spec.suspend to true"),
message);
- verifyNoInteractions(recorder);
- }
-
- @Test
- void deactivationIsReportedAndIgnored() {
- when(context.getCachedKueueWorkload())
- // Before Kueue marks the deactivated Workload as evicted
- .thenReturn(Optional.of(workload(false, admitted())))
- .thenReturn(Optional.of(workload(false, admitted(),
evicted("Deactivated"))))
- // Reactivated, while Kueue keeps the Evicted condition
- .thenReturn(Optional.of(workload(true, admitted(),
evicted("Deactivated"))));
- AppKueueEvictionStep step = new AppKueueEvictionStep();
- for (int i = 0; i < 3; i++) {
- Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
- }
-
- List<String> messages = captureMessages(3);
- // Kueue stops counting the quota of a deactivated Workload
- Assertions.assertTrue(messages.get(0).contains("is deactivated, "));
- Assertions.assertTrue(messages.get(0).contains("no longer counts"));
- Assertions.assertTrue(
- messages.get(1).contains("is deactivated (Deactivated: Deactivated by
the test)"));
- Assertions.assertTrue(messages.get(1).contains("no longer counts"));
- // And counts it again once it is reactivated
- Assertions.assertTrue(messages.get(2).contains("holding its quota"));
- verifyNoInteractions(recorder);
- }
-
- /** Returns the messages of the published events, which are all
KueueEvictionIgnored warnings. */
- private List<String> captureMessages(int count) {
- ArgumentCaptor<EventRecord> captor =
ArgumentCaptor.forClass(EventRecord.class);
- verify(eventRecorder, times(count)).record(captor.capture());
- for (EventRecord event : captor.getAllValues()) {
- Assertions.assertEquals(EventType.WARNING, event.type());
- Assertions.assertEquals(EventUtils.REASON_KUEUE_EVICTION_IGNORED,
event.reason());
- }
- return captor.getAllValues().stream().map(EventRecord::message).toList();
- }
-
- private static Workload workload(boolean active, Condition... conditions) {
- Workload workload = new Workload();
- workload.setMetadata(new
ObjectMetaBuilder().withName("sparkapplication-app1").build());
- workload.setSpec(WorkloadSpec.builder().active(active).build());
-
workload.setStatus(WorkloadStatus.builder().conditions(List.of(conditions)).build());
- return workload;
- }
-
- private static Condition admitted() {
- return new
ConditionBuilder().withType("Admitted").withStatus("True").build();
- }
-
- private static Condition evicted(String reason) {
- return new ConditionBuilder()
- .withType("Evicted")
- .withStatus("True")
- .withReason(reason)
- .withMessage(reason + " by the test")
- .build();
- }
-}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
index 5d8cde58..bb1a7c3a 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
@@ -34,8 +34,10 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import io.fabric8.kubernetes.api.model.ConditionBuilder;
import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
@@ -52,6 +54,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
@@ -61,7 +64,9 @@ import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.config.SparkOperatorConf;
import org.apache.spark.k8s.operator.context.SparkAppContext;
import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
+import org.apache.spark.k8s.operator.kueue.v1beta2.RequeueState;
import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
+import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
import org.apache.spark.k8s.operator.status.ApplicationState;
@@ -593,6 +598,286 @@ class AppSuspendStepTest {
new AppSuspendStep().reconcile(mockContext, recorder));
}
+ @ParameterizedTest
+ @CsvSource({"Preempted, true", "Deactivated, false", "Deactivated, true"})
+ void evictedRunningAppEntersSuspendedBeforeReleasingResources(String reason,
boolean active) {
+ // A deactivated Workload no longer counts against the quota, and a
reactivated one is queued
+ // again, so both release the driver and executors like a preemption
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus(reason), active);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ ApplicationState state = captureAppendedState();
+ Assertions.assertEquals(ApplicationStateSummary.Suspended,
state.getCurrentStateSummary());
+ Assertions.assertEquals(
+ Constants.APP_EVICTED_MESSAGE + " " + reason + ": " + reason + " by
the test",
+ state.getMessage());
+ // Like spec.suspend, nothing is released until Suspended is persisted,
and nothing is reported
+ // but the Suspended state
+ Assertions.assertNotNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(eventRecorder);
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = ApplicationStateSummary.class,
+ from = "DriverRequested",
+ to = "RunningWithBelowThresholdExecutors")
+ void podsReadyTimeoutSuspendsAppLikeAnyOtherEviction(ApplicationStateSummary
summary) {
+ // Including an application which runs with fewer executors than
spark.executor.instances, as
+ // its applicationTolerations allow, since its Workload requested the
quota for all of them
+ SparkApplication app = buildKueueApp(summary, false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus("PodsReadyTimeout"), true);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertEquals(
+ ApplicationStateSummary.Suspended,
captureAppendedState().getCurrentStateSummary());
+ }
+
+ @Test
+ void runningAppWithAdmittedWorkloadProceeds() {
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(
+ app,
+ WorkloadStatus.builder()
+ .conditions(
+ List.of(new
ConditionBuilder().withType("Admitted").withStatus("True").build()))
+ .build(),
+ true);
+
+ Assertions.assertEquals(
+ ReconcileProgress.proceed(), new
AppSuspendStep().reconcile(mockContext, recorder));
+
+ verifyNoInteractions(recorder, eventRecorder);
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"Succeeded", "Failed"})
+ void attemptWhoseDriverEndedMeanwhileEndsAsUsualOnEviction(String phase) {
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, false);
+ stubContext(app);
+ kubernetesClient
+ .resource(new
PodBuilder(driver(app)).withNewStatus().withPhase(phase).endStatus().build())
+ .create();
+ createWorkload(app, evictedStatus("Preempted"), true);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ // The attempt ends rather than being queued again and run from scratch
+ Assertions.assertEquals(
+ ApplicationStateSummary.valueOf(phase),
captureAppendedState().getCurrentStateSummary());
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @Test
+ void suspendTakesPrecedenceOverEviction() {
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, true);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus("Preempted"), true);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ ApplicationState state = captureAppendedState();
+ Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+
+ // Once everything is released, the application is held rather than queued
again
+ app.setStatus(app.getStatus().appendNewState(state));
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertEquals(
+ SUSPEND_HOLD_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getWorkload());
+ verify(recorder).appendNewStateAndPersist(any(), any());
+ verify(recorder, never()).persistStatus(any(), any());
+ }
+
+ @Test
+ void suspendingEvictedAppIsRecorded() {
+ SparkApplication app = buildEvictedApp(true);
+ stubContext(app);
+ createWorkload(app, evictedStatus("Preempted"), true);
+
+ // The application is no longer queued again once released, and resuming
it later says so
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ ApplicationState state = captureAppendedState();
+ Assertions.assertEquals(ApplicationStateSummary.Suspended,
state.getCurrentStateSummary());
+ Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+
+ // Once that state is persisted, the application is held rather than
suspended again, and its
+ // Workload is released like on spec.suspend
+ app.setStatus(app.getStatus().appendNewState(state));
+ Assertions.assertEquals(
+ SUSPEND_HOLD_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ verify(recorder).appendNewStateAndPersist(any(), any());
+ Assertions.assertNull(getWorkload());
+ }
+
+ @Test
+ void evictedThenSuspendedAppIsResumedRatherThanQueuedAgain() {
+ SparkApplication app = buildEvictedApp(true);
+ stubContext(app);
+ createWorkload(app, evictedStatus("Preempted"), true);
+ new AppSuspendStep().reconcile(mockContext, recorder);
+ app.setStatus(app.getStatus().appendNewState(captureAppendedState()));
+
+ // Held by spec.suspend since the eviction, so clearing it is a resume,
not a requeue
+ app.getSpec().setSuspend(false);
+ new AppSuspendStep().reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(Constants.APP_RESUMED_MESSAGE,
capturePersistedState().getMessage());
+ }
+
+ @Test
+ void evictedAppIsQueuedAgainOnceEverythingIsReleased() {
+ SparkApplication app = buildEvictedApp(false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus("Preempted"), true);
+
+ // The driver which is deleted is waited for, along with the Workload
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder);
+
+ // Then it starts a new attempt like a resumed application, see
ApplicationStatus#resume
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+ Assertions.assertEquals(Constants.APP_REQUEUED_MESSAGE,
capturePersistedState().getMessage());
+ // AppInitStep creates a new Workload of the same name and waits for its
admission
+ Assertions.assertNull(getWorkload());
+ }
+
+ @Test
+ void deactivatedWorkloadIsKeptUntilItIsReactivated() {
+ SparkApplication app = buildEvictedApp(false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus("Deactivated"), false);
+
+ // The driver and executors are released, since Kueue no longer counts
their quota
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getDriver());
+
+ // While the Workload is kept, since a new Workload of the application
would come back active
+ Assertions.assertEquals(
+ SUSPEND_HOLD_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder);
+
+ // Its reactivation, which the Workload informer reconciles, queues the
application again
+ Workload workload = getWorkload();
+ workload.getSpec().setActive(true);
+ kubernetesClient.resource(workload).update();
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertEquals(Constants.APP_REQUEUED_MESSAGE,
capturePersistedState().getMessage());
+ Assertions.assertNull(getWorkload());
+ }
+
+ @Test
+ void evictedWorkloadIsKeptUntilTheRequeueBackoffElapses() {
+ SparkApplication app = buildEvictedApp(false);
+ stubContext(app);
+ WorkloadStatus status = evictedStatus("PodsReadyTimeout");
+ status.setRequeueState(
+ RequeueState.builder()
+ .count(1)
+ .requeueAt(Instant.now().plusSeconds(60).toString())
+ .build());
+ createWorkload(app, status, true);
+
+ ReconcileProgress progress = new AppSuspendStep().reconcile(mockContext,
recorder);
+
+ // Its deletion would drop the backoff, so it is released once the backoff
elapsed
+ Duration requeueAfter = progress.getRequeueAfterDuration();
+ Assertions.assertTrue(progress.isRequeue());
+ Assertions.assertTrue(
+ requeueAfter.compareTo(Duration.ofSeconds(50)) > 0
+ && requeueAfter.compareTo(Duration.ofSeconds(60)) <= 0,
+ requeueAfter::toString);
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void failedReleaseIsRetriedWhileDeactivatedWorkloadIsKept() {
+ SparkApplication app = buildEvictedApp(false);
+ createWorkload(app, evictedStatus("Deactivated"), false);
+ KubernetesClient client = spy(kubernetesClient);
+ doThrow(new KubernetesClientException("Service Unavailable", 503,
null)).when(client).pods();
+ stubContext(app, client);
+
+ // Rather than after the hold interval of the deactivated Workload, which
would keep the driver
+ // and executors running for up to 30 minutes outside of the quota
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @Test
+ void resumedAppReleasesDeactivatedWorkload() {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ stubContext(app);
+ // The Workload outlived the release on spec.suspend, e.g. after a failed
delete
+ createWorkload(app, evictedStatus("Deactivated"), false);
+
+ // Only an eviction keeps the Workload, so the resume does not wait for
its reactivation
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertEquals(Constants.APP_RESUMED_MESSAGE,
capturePersistedState().getMessage());
+ Assertions.assertNull(getWorkload());
+ }
+
+ @Test
+ void evictedAppWhoseQueueLabelIsRemovedEntersSuspended() {
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, false);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app, evictedStatus("Preempted"), true);
+ // Like the release on spec.suspend, the Workload admitted before the
label was removed counts
+ app.getMetadata().setLabels(Map.of());
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertEquals(
+ ApplicationStateSummary.Suspended,
captureAppendedState().getCurrentStateSummary());
+ }
+
@ParameterizedTest
@EnumSource(
value = ApplicationStateSummary.class,
@@ -630,6 +915,8 @@ class AppSuspendStepTest {
when(mockContext.getResource()).thenReturn(app);
when(mockContext.getClient()).thenReturn(client);
when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+ // Like the Workload informer, the cache follows the Workload in the API
server
+ when(mockContext.getCachedKueueWorkload()).thenAnswer(i ->
Optional.ofNullable(getWorkload()));
}
private ApplicationState captureAppendedState() {
@@ -639,6 +926,14 @@ class AppSuspendStepTest {
return captor.getValue();
}
+ private ApplicationState capturePersistedState() {
+ ArgumentCaptor<ApplicationStatus> status =
ArgumentCaptor.forClass(ApplicationStatus.class);
+ verify(recorder).persistStatus(eq(mockContext), status.capture());
+ ApplicationState state = status.getValue().getCurrentState();
+ Assertions.assertEquals(ApplicationStateSummary.Submitted,
state.getCurrentStateSummary());
+ return state;
+ }
+
private EventRecord captureEvent() {
ArgumentCaptor<EventRecord> event =
ArgumentCaptor.forClass(EventRecord.class);
verify(eventRecorder).record(event.capture());
@@ -686,6 +981,29 @@ class AppSuspendStepTest {
kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(app)).create();
}
+ private void createWorkload(SparkApplication app, WorkloadStatus status,
boolean active) {
+ Workload desired = KueueWorkloadFactory.buildWorkload(app);
+ desired.getSpec().setActive(active);
+ kubernetesClient.resource(desired).create();
+ Workload workload = getWorkload();
+ workload.setStatus(status);
+ kubernetesClient.resource(workload).update();
+ }
+
+ private static WorkloadStatus evictedStatus(String reason) {
+ return WorkloadStatus.builder()
+ .conditions(
+ List.of(
+ new
ConditionBuilder().withType("Admitted").withStatus("True").build(),
+ new ConditionBuilder()
+ .withType("Evicted")
+ .withStatus("True")
+ .withReason(reason)
+ .withMessage(reason + " by the test")
+ .build()))
+ .build();
+ }
+
private static Pod driver(SparkApplication app) {
return pod(DRIVER, Utils.driverLabels(app));
}
@@ -726,4 +1044,13 @@ class AppSuspendStepTest {
app.getMetadata().setLabels(Map.of(Constants.LABEL_QUEUE_NAME,
"test-queue"));
return app;
}
+
+ /** Builds an application which entered Suspended on a Kueue eviction. */
+ private static SparkApplication buildEvictedApp(boolean suspend) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ app.getStatus()
+ .getCurrentState()
+ .setMessage(Constants.APP_EVICTED_MESSAGE + " Preempted: Preempted by
the test");
+ return app;
+ }
}
diff --git a/tests/e2e/kueue/chainsaw-test.yaml
b/tests/e2e/kueue/chainsaw-test.yaml
index 4435a009..ce84f314 100644
--- a/tests/e2e/kueue/chainsaw-test.yaml
+++ b/tests/e2e/kueue/chainsaw-test.yaml
@@ -804,6 +804,141 @@ spec:
kubectl delete clusterqueue "$CLUSTER_QUEUE" --ignore-not-found=true
--wait=false
kubectl delete resourceflavor "$FLAVOR" --ignore-not-found=true
--wait=false
kubectl delete workloadpriorityclass "$LOW_PRIORITY_CLASS"
"$PRIORITY_CLASS" --ignore-not-found=true --wait=false
+ - name: spark-application-is-preempted-by-kueue
+ # Each step uses its own queue objects, so that a step never waits for
Kueue to finish
+ # deleting the objects of another step.
+ bindings:
+ - name: FLAVOR
+ value: spark-e2e-app-preemption-flavor
+ - name: CLUSTER_QUEUE
+ value: spark-e2e-app-preemption-cluster-queue
+ - name: LOCAL_QUEUE
+ value: spark-e2e-app-preemption-queue
+ - name: LOW_PRIORITY_CLASS
+ value: spark-e2e-app-preemption-low-priority
+ - name: PRIORITY_CLASS
+ value: spark-e2e-app-preemption-high-priority
+ try:
+ - apply:
+ file: kueue-queues.yaml
+ # The quota fits only one application (1 CPU for the driver and 1 for the
executor), and a
+ # Workload of a higher priority may preempt one of a lower priority.
+ - script:
+ timeout: 30s
+ env:
+ - name: CLUSTER_QUEUE
+ value: ($CLUSTER_QUEUE)
+ content: |
+ kubectl patch clusterqueue "$CLUSTER_QUEUE" --type json -p \
+ '[{"op": "replace", "path":
"/spec/resourceGroups/0/flavors/0/resources/0/nominalQuota", "value": "2"},
+ {"op": "replace", "path":
"/spec/resourceGroups/0/flavors/0/resources/1/nominalQuota", "value": "8Gi"},
+ {"op": "add", "path": "/spec/preemption", "value":
{"withinClusterQueue": "LowerPriority"}}]'
+ - apply:
+ file: spark-example-preempted.yaml
+ # Its driver and executor are ready, so the eviction below is a
preemption, not a
+ # PodsReadyTimeout
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: Workload
+ metadata:
+ name: sparkapplication-spark-job-kueue-preempted-test
+ namespace: default
+ status:
+ (conditions[?type == 'PodsReady' && status == 'True'] |
length(@)): 1
+ - apply:
+ file: spark-example-prioritized.yaml
+ # Kueue evicts the Workload of the running application, and keeps its
quota until the operator
+ # releases the driver and executor first and then the Workload.
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: v1
+ kind: Event
+ metadata:
+ namespace: default
+ involvedObject:
+ kind: SparkApplication
+ name: spark-job-kueue-preempted-test
+ reason: Suspended
+ (contains(message || '', 'evicted by Kueue')): true
+ (contains(message || '', 'Preempted')): true
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: v1
+ kind: Event
+ metadata:
+ namespace: default
+ involvedObject:
+ kind: SparkApplication
+ name: spark-job-kueue-preempted-test
+ reason: Submitted
+ (contains(message || '', 'after its eviction')): true
+ # The application of the higher priority is admitted into the quota which
the preempted one
+ # gave back
+ - assert:
+ bindings:
+ - name: SPARK_APP_NAMESPACE
+ value: default
+ - name: SPARK_APPLICATION_NAME
+ value: spark-job-kueue-prioritized-test
+ timeout: 10m
+ file: "../assertions/spark-application/spark-state-transition.yaml"
+ # The preempted application is admitted again with a new Workload once the
other one releases
+ # the quota, and runs again with a new attempt, which does not count as a
restart, even with
+ # the default restartPolicy Never
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-kueue-preempted-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: RunningHealthy
+ currentAttemptSummary:
+ attemptInfo:
+ id: 1
+ restartCounter: 0
+ catch:
+ - describe:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ namespace: default
+ - describe:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: Workload
+ namespace: default
+ - describe:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: ClusterQueue
+ - events:
+ namespace: default
+ finally:
+ - script:
+ timeout: 120s
+ env:
+ - name: FLAVOR
+ value: ($FLAVOR)
+ - name: CLUSTER_QUEUE
+ value: ($CLUSTER_QUEUE)
+ - name: LOCAL_QUEUE
+ value: ($LOCAL_QUEUE)
+ - name: LOW_PRIORITY_CLASS
+ value: ($LOW_PRIORITY_CLASS)
+ - name: PRIORITY_CLASS
+ value: ($PRIORITY_CLASS)
+ content: |
+ kubectl delete sparkapplication spark-job-kueue-prioritized-test
spark-job-kueue-preempted-test -n default --ignore-not-found=true
+ # Kueue finalizers may delay the deletion, which later steps do not
depend on.
+ kubectl delete localqueue "$LOCAL_QUEUE" -n default
--ignore-not-found=true --wait=false
+ kubectl delete clusterqueue "$CLUSTER_QUEUE" --ignore-not-found=true
--wait=false
+ kubectl delete resourceflavor "$FLAVOR" --ignore-not-found=true
--wait=false
+ kubectl delete workloadpriorityclass "$LOW_PRIORITY_CLASS"
"$PRIORITY_CLASS" --ignore-not-found=true --wait=false
- name: spark-cluster-whose-pods-are-not-ready-is-evicted-by-kueue
# Each step uses its own queue objects, so that a step never waits for
Kueue to finish
# deleting the objects of another step.
@@ -1027,7 +1162,9 @@ spec:
namespace: default
status:
(conditions[?type == 'Evicted' && status == 'True' && reason ==
'PodsReadyTimeout'] | length(@)): 1
- # A running application is not released on it yet
+ # Like a running SparkCluster, the operator releases the driver and
executor first and then the
+ # Workload once its requeue backoff, which the CI shortens to 10 seconds,
elapses, and the
+ # application is queued again with a new attempt.
- assert:
timeout: 60s
resource:
@@ -1038,70 +1175,24 @@ spec:
involvedObject:
kind: SparkApplication
name: spark-job-kueue-executors-not-ready-test
- type: Warning
- reason: KueueEvictionIgnored
+ reason: Suspended
+ (contains(message || '', 'evicted by Kueue')): true
(contains(message || '', 'PodsReadyTimeout')): true
- assert:
- timeout: 30s
- resource:
- apiVersion: spark.apache.org/v1
- kind: SparkApplication
- metadata:
- name: spark-job-kueue-executors-not-ready-test
- namespace: default
- status:
- currentState:
- currentStateSummary: RunningHealthy
- # As the event advises, suspending the application releases its driver and
executors, and then
- # the quota which its evicted Workload keeps
- - script:
- timeout: 30s
- content: |
- kubectl patch sparkapplication
spark-job-kueue-executors-not-ready-test -n default --type merge -p
'{"spec":{"suspend":true}}'
- - assert:
- timeout: 60s
- resource:
- apiVersion: spark.apache.org/v1
- kind: SparkApplication
- metadata:
- name: spark-job-kueue-executors-not-ready-test
- namespace: default
- status:
- currentState:
- currentStateSummary: Suspended
- - error:
- timeout: 120s
+ timeout: 5m
resource:
apiVersion: v1
- kind: Pod
- metadata:
- namespace: default
- labels:
- spark.operator/spark-app-name:
spark-job-kueue-executors-not-ready-test
- - error:
- timeout: 120s
- resource:
- apiVersion: kueue.x-k8s.io/v1beta2
- kind: Workload
+ kind: Event
metadata:
- name: sparkapplication-spark-job-kueue-executors-not-ready-test
namespace: default
- - assert:
- timeout: 60s
- resource:
- apiVersion: kueue.x-k8s.io/v1beta2
- kind: ClusterQueue
- metadata:
- name: ($CLUSTER_QUEUE)
- status:
- admittedWorkloads: 0
- reservingWorkloads: 0
- # Resuming queues it again with a new attempt and a new Workload, which
Kueue admits into the
- # quota it gave back
- - script:
- timeout: 30s
- content: |
- kubectl patch sparkapplication
spark-job-kueue-executors-not-ready-test -n default --type merge -p
'{"spec":{"suspend":false}}'
+ involvedObject:
+ kind: SparkApplication
+ name: spark-job-kueue-executors-not-ready-test
+ reason: Submitted
+ (contains(message || '', 'after its eviction')): true
+ # The new attempt does not count as a restart, even with the default
restartPolicy Never, and
+ # runs again on the quota which Kueue admits its new Workload into. It is
evicted again once the
+ # waitForPodsReady timeout elapses, since its executor is never scheduled.
- assert:
timeout: 5m
resource:
@@ -1113,15 +1204,21 @@ spec:
status:
currentState:
currentStateSummary: RunningHealthy
+ currentAttemptSummary:
+ attemptInfo:
+ id: 1
+ restartCounter: 0
- assert:
timeout: 60s
resource:
apiVersion: kueue.x-k8s.io/v1beta2
- kind: ClusterQueue
+ kind: Workload
metadata:
- name: ($CLUSTER_QUEUE)
+ name: sparkapplication-spark-job-kueue-executors-not-ready-test
+ namespace: default
status:
- admittedWorkloads: 1
+ (conditions[?type == 'Admitted' && status == 'True'] | length(@)):
1
+ (conditions[?type == 'Evicted' && status == 'True'] | length(@)): 0
catch:
- describe:
apiVersion: spark.apache.org/v1
diff --git a/tests/e2e/kueue/spark-example-preempted.yaml
b/tests/e2e/kueue/spark-example-preempted.yaml
new file mode 100644
index 00000000..2f2eec03
--- /dev/null
+++ b/tests/e2e/kueue/spark-example-preempted.yaml
@@ -0,0 +1,42 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+apiVersion: kueue.x-k8s.io/v1beta2
+kind: WorkloadPriorityClass
+metadata:
+ name: ($LOW_PRIORITY_CLASS)
+value: 100
+---
+apiVersion: spark.apache.org/v1
+kind: SparkApplication
+metadata:
+ name: spark-job-kueue-preempted-test
+ namespace: default
+ labels:
+ kueue.x-k8s.io/queue-name: ($LOCAL_QUEUE)
+ kueue.x-k8s.io/priority-class: ($LOW_PRIORITY_CLASS)
+spec:
+ # A streaming job which runs until it is deleted, so that it is still
running when it is
+ # preempted, and when its next attempt is checked
+ mainClass: "org.apache.spark.examples.streaming.CustomReceiver"
+ driverArgs: ["localhost", "9999"]
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ sparkConf:
+ spark.executor.instances: "1"
+ spark.kubernetes.container.image: "apache/spark:{{SPARK_VERSION}}-scala"
+ spark.kubernetes.authenticate.driver.serviceAccountName: "spark"
+ runtimeVersions:
+ sparkVersion: "4.2.0"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
