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 ac338832 [SPARK-60057] Add `suspendReason` to `Suspended` states
ac338832 is described below
commit ac33883212ecb47d0e848c56b9defe60f9fdd2ac
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 15:15:24 2026 -0700
[SPARK-60057] Add `suspendReason` to `Suspended` states
### What changes were proposed in this pull request?
This PR adds `status.currentState.suspendReason` (`SpecSuspend` or
`KueueEviction`), which is set on `Suspended` states, and uses it in
`AppSuspendStep` and `ClusterSuspendStep` instead of the prefix of the state
message.
- `SuspendReason` and the nullable field in `BaseState`. The existing 3-arg
constructor is kept.
- `isSuspendedByEviction` only checks the reason of the current state. The
cluster's walk through the history is removed. The stuck-pods state keeps the
reason of the state before it.
- A `Suspended` state without a reason is treated as `SpecSuspend`.
- Regenerate the `v1` CRDs (`v1beta1` is frozen), and document the field in
`docs/spark_custom_resources.md`.
### Why are the changes needed?
The message prefix became a control protocol, as pointed out in the review
of #952. If the wording changes, a resource suspended by an eviction by an
older version is taken for a manual suspend, so `keepKueueWorkload()` is
skipped and the remaining requeue backoff is ignored.
No message-prefix fallback is needed. `Suspended` is not in 1.0.0, and the
migration guide already requires replacing the CRDs for 1.1.0, which ships the
new field.
### Does this PR introduce _any_ user-facing change?
No for the released versions. `suspendReason` is a new field within the
unreleased 1.1.0 changes, and the state messages are unchanged.
### How was this patch tested?
Pass the CIs. `gradle build` passes locally, with new and updated tests in
`ApplicationStatusTest`, `ClusterStatusTest`, `AppSuspendStepTest`,
`ClusterSuspendStepTest` and `ClusterInitStepTest`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
Closes #954 from dongjoon-hyun/SPARK-60057.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../sparkapplications.spark.apache.org-v1.yaml | 20 +++++++++++
.../crds/sparkclusters.spark.apache.org-v1.yaml | 20 +++++++++++
docs/spark_custom_resources.md | 6 ++++
.../org/apache/spark/k8s/operator/Constants.java | 4 +--
.../k8s/operator/status/ApplicationState.java | 10 ++++++
.../spark/k8s/operator/status/BaseState.java | 14 ++++++++
.../spark/k8s/operator/status/ClusterState.java | 10 ++++++
.../status/{BaseState.java => SuspendReason.java} | 27 +++++----------
.../k8s/operator/status/ApplicationStatusTest.java | 25 ++++++++++++++
.../k8s/operator/status/ClusterStatusTest.java | 21 ++++++++++++
.../reconciler/reconcilesteps/AppSuspendStep.java | 16 +++++----
.../reconciler/reconcilesteps/ClusterInitStep.java | 5 ++-
.../reconcilesteps/ClusterSuspendStep.java | 40 +++++++++++-----------
.../reconcilesteps/AppSuspendStepTest.java | 27 +++++++++++++++
.../reconcilesteps/ClusterInitStepTest.java | 2 ++
.../reconcilesteps/ClusterSuspendStepTest.java | 35 ++++++++++++++++++-
16 files changed, 233 insertions(+), 49 deletions(-)
diff --git
a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
index 3c799123..87077ca4 100644
---
a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
+++
b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
@@ -17208,6 +17208,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
@@ -17884,6 +17889,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
previousAttemptSummary:
properties:
@@ -18572,6 +18582,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
@@ -19249,6 +19264,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
diff --git
a/build-tools/helm/spark-kubernetes-operator/crds/sparkclusters.spark.apache.org-v1.yaml
b/build-tools/helm/spark-kubernetes-operator/crds/sparkclusters.spark.apache.org-v1.yaml
index e2ccacca..9bef5365 100644
---
a/build-tools/helm/spark-kubernetes-operator/crds/sparkclusters.spark.apache.org-v1.yaml
+++
b/build-tools/helm/spark-kubernetes-operator/crds/sparkclusters.spark.apache.org-v1.yaml
@@ -14760,6 +14760,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
@@ -14778,6 +14783,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
previousAttemptSummary:
properties:
@@ -14804,6 +14814,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
@@ -14823,6 +14838,11 @@ spec:
type: string
message:
type: string
+ suspendReason:
+ enum:
+ - KueueEviction
+ - SpecSuspend
+ type: string
type: object
type: object
type: object
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index 97fd7e2c..36270370 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -606,6 +606,12 @@ spec:
a lost node, no longer holds the Kueue `Workload`, so that it does not hold
the quota forever. The
cluster still stays `Suspended` until such a pod is gone, since it keeps the
name of the master or
worker to create, and the message of its `Suspended` state names such pods.
+* The `Suspended` state says why the resource is suspended in
`status.currentState.suspendReason`:
+ `SpecSuspend` if it is held by `spec.suspend`, or `KueueEviction` if the
eviction of its Kueue
+ `Workload` released it, see [Kueue](#kueue). The operator relies on this
field, not on the
+ message, which is for users to read and may be reworded. `spec.suspend`
takes precedence, so
+ setting it on a resource suspended by an eviction records another
`Suspended` state with
+ `SpecSuspend`. A `Suspended` state without it is treated as `SpecSuspend`.
* An operator version without the `Suspended` state cannot read a
`SparkApplication` or a
`SparkCluster` whose status has it, even after the resource is resumed. A
resumed
`SparkApplication` keeps it in its state transition history, or with
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 c0ad3a05..fe36c015 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
@@ -248,9 +248,7 @@ public final class Constants {
/**
* 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.
+ * which is followed by the reason and the message of the eviction.
*/
public static final String APP_EVICTED_MESSAGE =
"Application is evicted by Kueue, so its driver and executors are being
released before its "
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationState.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationState.java
index ec76cef4..60698bf8 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationState.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationState.java
@@ -62,4 +62,14 @@ public class ApplicationState extends
BaseState<ApplicationStateSummary> impleme
public ApplicationState(ApplicationStateSummary currentStateSummary, String
message) {
super(currentStateSummary, Instant.now().toString(), message);
}
+
+ /**
+ * Constructs a new ApplicationState with the given state summary, message
and suspend reason.
+ *
+ * @since 1.1.0
+ */
+ public ApplicationState(
+ ApplicationStateSummary currentStateSummary, String message,
SuspendReason suspendReason) {
+ super(currentStateSummary, Instant.now().toString(), message,
suspendReason);
+ }
}
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
index 0ae4c03b..d5997a66 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
@@ -40,4 +40,18 @@ public class BaseState<T> implements Serializable {
protected T currentStateSummary;
protected String lastTransitionTime;
protected String message;
+
+ /** Why the resource is suspended, only set on a Suspended state. */
+ protected SuspendReason suspendReason;
+
+ /**
+ * Constructs a new BaseState without a suspend reason.
+ *
+ * @param currentStateSummary The current state summary.
+ * @param lastTransitionTime The time of the last transition.
+ * @param message The message of the state.
+ */
+ public BaseState(T currentStateSummary, String lastTransitionTime, String
message) {
+ this(currentStateSummary, lastTransitionTime, message, null);
+ }
}
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterState.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterState.java
index f2e4d77c..cc7f5480 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterState.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ClusterState.java
@@ -56,4 +56,14 @@ public class ClusterState extends
BaseState<ClusterStateSummary> implements Seri
public ClusterState(ClusterStateSummary currentStateSummary, String message)
{
super(currentStateSummary, Instant.now().toString(), message);
}
+
+ /**
+ * Constructs a new ClusterState with the given state summary, message and
suspend reason.
+ *
+ * @since 1.1.0
+ */
+ public ClusterState(
+ ClusterStateSummary currentStateSummary, String message, SuspendReason
suspendReason) {
+ super(currentStateSummary, Instant.now().toString(), message,
suspendReason);
+ }
}
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/SuspendReason.java
similarity index 60%
copy from
spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
copy to
spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/SuspendReason.java
index 0ae4c03b..e30e106b 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/BaseState.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/SuspendReason.java
@@ -19,25 +19,16 @@
package org.apache.spark.k8s.operator.status;
-import java.io.Serializable;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonInclude;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-
/**
- * Base class for state information.
+ * Reason of a Suspended state, which tells the suspend by spec.suspend apart
from the one by the
+ * eviction of the Kueue Workload.
*
- * @param <T> The type of the current state summary.
- * @since 0.1.0
+ * @since 1.1.0
*/
-@Data
-@AllArgsConstructor
-@JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class BaseState<T> implements Serializable {
- protected T currentStateSummary;
- protected String lastTransitionTime;
- protected String message;
+public enum SuspendReason {
+ /** The resource is held by spec.suspend. */
+ SpecSuspend,
+
+ /** The resource is released after Kueue evicted its Workload, and is queued
again then. */
+ KueueEviction
}
diff --git
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
index dcc20385..1edc8532 100644
---
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
+++
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
@@ -23,7 +23,9 @@ import static
org.apache.spark.k8s.operator.status.ApplicationStateSummary.Submi
import static
org.apache.spark.k8s.operator.status.ApplicationStateSummary.Succeeded;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -927,4 +929,27 @@ class ApplicationStatusTest {
assertTrue(deserialized.getStateTransitionHistory().isEmpty());
assertEquals(Submitted,
deserialized.getCurrentState().getCurrentStateSummary());
}
+
+ @Test
+ void testSuspendReasonJsonRoundTrip() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+ ApplicationState suspended =
+ new ApplicationState(
+ ApplicationStateSummary.Suspended, "test",
SuspendReason.KueueEviction);
+
+ String json = mapper.writeValueAsString(suspended);
+ assertTrue(json.contains("\"suspendReason\":\"KueueEviction\""));
+ assertEquals(suspended, mapper.readValue(json, ApplicationState.class));
+
+ // The reason is set only on a Suspended state, and a status without one
still reads
+ ApplicationState submitted = new ApplicationState();
+ assertNull(submitted.getSuspendReason());
+
assertFalse(mapper.writeValueAsString(submitted).contains("suspendReason"));
+ assertNull(
+ mapper
+ .readValue(
+ "{\"currentStateSummary\":\"Suspended\",\"message\":\"test\"}",
+ ApplicationState.class)
+ .getSuspendReason());
+ }
}
diff --git
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ClusterStatusTest.java
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ClusterStatusTest.java
index 96270f93..d50edf1c 100644
---
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ClusterStatusTest.java
+++
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ClusterStatusTest.java
@@ -22,6 +22,7 @@ package org.apache.spark.k8s.operator.status;
import static
org.apache.spark.k8s.operator.status.ClusterStateSummary.Submitted;
import static org.junit.jupiter.api.Assertions.*;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
class ClusterStatusTest {
@@ -60,4 +61,24 @@ class ClusterStatusTest {
assertEquals(4, kept.getStateTransitionHistory().size());
assertEquals(resumed, kept.getStateTransitionHistory().get(3L));
}
+
+ @Test
+ void testSuspendReasonJsonRoundTrip() throws Exception {
+ ObjectMapper mapper = new ObjectMapper();
+ ClusterState suspended =
+ new ClusterState(ClusterStateSummary.Suspended, "test",
SuspendReason.SpecSuspend);
+
+ String json = mapper.writeValueAsString(suspended);
+ assertTrue(json.contains("\"suspendReason\":\"SpecSuspend\""));
+ assertEquals(suspended, mapper.readValue(json, ClusterState.class));
+
+ // The reason is set only on a Suspended state, and a status without one
still reads
+ assertFalse(mapper.writeValueAsString(new
ClusterState()).contains("suspendReason"));
+ assertNull(
+ mapper
+ .readValue(
+ "{\"currentStateSummary\":\"Suspended\",\"message\":\"test\"}",
+ ClusterState.class)
+ .getSuspendReason());
+ }
}
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 b9ec67d8..9257a7a3 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
@@ -57,6 +57,7 @@ import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverRunningObserv
import org.apache.spark.k8s.operator.spec.ApplicationTimeoutConfig;
import org.apache.spark.k8s.operator.status.ApplicationState;
import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
@@ -128,8 +129,10 @@ 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.
+ SuspendReason reason = suspend ? SuspendReason.SpecSuspend :
SuspendReason.KueueEviction;
return statusRecorder.appendNewStateAndPersist(
- context, new ApplicationState(ApplicationStateSummary.Suspended,
message.get()))
+ context,
+ new ApplicationState(ApplicationStateSummary.Suspended,
message.get(), reason))
? completeAndImmediateRequeue()
: completeAndDefaultRequeue();
}
@@ -143,7 +146,8 @@ public final class AppSuspendStep extends AppReconcileStep {
return appendStateAndImmediateRequeue(
context,
statusRecorder,
- new ApplicationState(ApplicationStateSummary.Suspended,
APP_SUSPENDED_MESSAGE));
+ new ApplicationState(
+ ApplicationStateSummary.Suspended, APP_SUSPENDED_MESSAGE,
SuspendReason.SpecSuspend));
}
Optional<Duration> keepWorkload = evicted ? keepKueueWorkload(context) :
Optional.empty();
Optional<ReconcileProgress> waiting = releaseResources(context,
keepWorkload.isEmpty());
@@ -191,16 +195,16 @@ public final class AppSuspendStep extends
AppReconcileStep {
/**
* 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.
+ * by spec.suspend, by the reason of its current Suspended state. A
Suspended state without a
+ * reason, e.g. one whose reason the API server pruned for a CRD which does
not have it yet, is
+ * not taken for an eviction.
*
* @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);
+ return app.getStatus().getCurrentState().getSuspendReason() ==
SuspendReason.KueueEviction;
}
/**
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
index d5eb1859..c9f7347f 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
@@ -46,6 +46,7 @@ import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
import org.apache.spark.k8s.operator.status.ClusterState;
import org.apache.spark.k8s.operator.status.ClusterStatus;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
@@ -87,7 +88,9 @@ public final class ClusterInitStep extends
ClusterReconcileStep {
// where ClusterSuspendStep releases whatever it holds only after its
pods are gone.
if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
return appendStateAndImmediateRequeue(
- context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ context,
+ statusRecorder,
+ new ClusterState(Suspended, CLUSTER_SUSPENDED_MESSAGE,
SuspendReason.SpecSuspend));
}
return SuspendUtils.holdForSuspend(context, "master and workers");
}
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 91818074..15871973 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
@@ -41,7 +41,6 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.List;
-import java.util.NavigableMap;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -58,6 +57,7 @@ import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
import org.apache.spark.k8s.operator.status.ClusterState;
import org.apache.spark.k8s.operator.status.ClusterStateSummary;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
@@ -105,7 +105,8 @@ public final class ClusterSuspendStep extends
ClusterReconcileStep {
return appendStateAndImmediateRequeue(
context,
statusRecorder,
- new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ new ClusterState(
+ ClusterStateSummary.Suspended, CLUSTER_SUSPENDED_MESSAGE,
SuspendReason.SpecSuspend));
}
if (summary == ClusterStateSummary.RunningHealthy) {
return suspendOnKueueEviction(context, statusRecorder);
@@ -118,7 +119,10 @@ public final class ClusterSuspendStep extends
ClusterReconcileStep {
return appendStateAndImmediateRequeue(
context,
statusRecorder,
- new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ new ClusterState(
+ ClusterStateSummary.Suspended,
+ CLUSTER_SUSPENDED_MESSAGE,
+ SuspendReason.SpecSuspend));
}
Optional<Duration> keepWorkload = evicted ? keepKueueWorkload(context) :
Optional.empty();
Optional<ReconcileProgress> waiting =
@@ -181,31 +185,24 @@ public final class ClusterSuspendStep extends
ClusterReconcileStep {
return appendStateAndImmediateRequeue(
context,
statusRecorder,
- new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_EVICTED_MESSAGE + " " + cause));
+ new ClusterState(
+ ClusterStateSummary.Suspended,
+ CLUSTER_EVICTED_MESSAGE + " " + cause,
+ SuspendReason.KueueEviction));
}
/**
* Checks whether the cluster is suspended by the eviction of its Kueue
Workload rather than by
- * spec.suspend. The Suspended state which names its stuck pods, if any,
follows the one which
- * says why it is suspended.
+ * spec.suspend, by the reason of its current Suspended state, which the
Suspended state naming
+ * its stuck pods keeps from the one before. A Suspended state without a
reason, e.g. one whose
+ * reason the API server pruned for a CRD which does not have it yet, is not
taken for an
+ * eviction.
*
* @param cluster The suspended SparkCluster.
* @return True if the cluster was suspended by an eviction and not held by
spec.suspend since.
*/
private static boolean isSuspendedByEviction(SparkCluster cluster) {
- NavigableMap<Long, ClusterState> history =
- (NavigableMap<Long, ClusterState>)
cluster.getStatus().getStateTransitionHistory();
- for (ClusterState state : history.descendingMap().values()) {
- String message = state.getMessage();
- if (state.getCurrentStateSummary() != ClusterStateSummary.Suspended
- || CLUSTER_SUSPENDED_MESSAGE.equals(message)) {
- return false;
- }
- if (message != null && message.startsWith(CLUSTER_EVICTED_MESSAGE)) {
- return true;
- }
- }
- return false;
+ return cluster.getStatus().getCurrentState().getSuspendReason() ==
SuspendReason.KueueEviction;
}
/**
@@ -347,7 +344,10 @@ public final class ClusterSuspendStep extends
ClusterReconcileStep {
appendStateAndRequeueAfter(
context,
statusRecorder,
- new ClusterState(ClusterStateSummary.Suspended, message),
+ new ClusterState(
+ ClusterStateSummary.Suspended,
+ message,
+ cluster.getStatus().getCurrentState().getSuspendReason()),
holdInterval));
}
} catch (KubernetesClientException e) {
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 bb1a7c3a..7752a9f7 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
@@ -21,6 +21,7 @@ package
org.apache.spark.k8s.operator.reconciler.reconcilesteps;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -72,6 +73,7 @@ import
org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
import org.apache.spark.k8s.operator.status.ApplicationState;
import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
import org.apache.spark.k8s.operator.status.ApplicationStatus;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
import org.apache.spark.k8s.operator.utils.TestUtils;
@@ -147,6 +149,7 @@ class AppSuspendStepTest {
ApplicationState state = captureAppendedState();
Assertions.assertEquals(ApplicationStateSummary.Suspended,
state.getCurrentStateSummary());
Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
state.getSuspendReason());
// Nothing is released until Suspended is persisted, so that an
application which is resumed
// in the meantime is never left running without its driver, which would
fail it.
Assertions.assertNotNull(getDriver());
@@ -617,6 +620,7 @@ class AppSuspendStepTest {
Assertions.assertEquals(
Constants.APP_EVICTED_MESSAGE + " " + reason + ": " + reason + " by
the test",
state.getMessage());
+ Assertions.assertEquals(SuspendReason.KueueEviction,
state.getSuspendReason());
// Like spec.suspend, nothing is released until Suspended is persisted,
and nothing is reported
// but the Suspended state
Assertions.assertNotNull(getDriver());
@@ -698,6 +702,7 @@ class AppSuspendStepTest {
ApplicationState state = captureAppendedState();
Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
state.getSuspendReason());
// Once everything is released, the application is held rather than queued
again
app.setStatus(app.getStatus().appendNewState(state));
@@ -724,6 +729,7 @@ class AppSuspendStepTest {
ApplicationState state = captureAppendedState();
Assertions.assertEquals(ApplicationStateSummary.Suspended,
state.getCurrentStateSummary());
Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
state.getSuspendReason());
// Once that state is persisted, the application is held rather than
suspended again, and its
// Workload is released like on spec.suspend
@@ -734,6 +740,26 @@ class AppSuspendStepTest {
Assertions.assertNull(getWorkload());
}
+ @Test
+ void evictionIsToldByTheSuspendReasonRatherThanByTheMessage() {
+ // A reworded message of an application suspended by an eviction before
still says so
+ SparkApplication evicted = buildEvictedApp(false);
+ evicted.getStatus().getCurrentState().setMessage("Reworded in another
version");
+ stubContext(evicted);
+ createWorkload(evicted, evictedStatus("Preempted"), true);
+ new AppSuspendStep().reconcile(mockContext, recorder);
+ Assertions.assertEquals(Constants.APP_REQUEUED_MESSAGE,
capturePersistedState().getMessage());
+
+ // So does a Suspended state without a reason, even if its message looks
like an eviction
+ SparkApplication unknown = buildEvictedApp(false);
+ unknown.getStatus().getCurrentState().setSuspendReason(null);
+ stubContext(unknown);
+ createWorkload(unknown, evictedStatus("Preempted"), true);
+ clearInvocations(recorder);
+ new AppSuspendStep().reconcile(mockContext, recorder);
+ Assertions.assertEquals(Constants.APP_RESUMED_MESSAGE,
capturePersistedState().getMessage());
+ }
+
@Test
void evictedThenSuspendedAppIsResumedRatherThanQueuedAgain() {
SparkApplication app = buildEvictedApp(true);
@@ -1051,6 +1077,7 @@ class AppSuspendStepTest {
app.getStatus()
.getCurrentState()
.setMessage(Constants.APP_EVICTED_MESSAGE + " Preempted: Preempted by
the test");
+
app.getStatus().getCurrentState().setSuspendReason(SuspendReason.KueueEviction);
return app;
}
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
index 1f136cb6..c32511c8 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
@@ -95,6 +95,7 @@ import
org.apache.spark.k8s.operator.spec.WorkerInstanceConfig;
import org.apache.spark.k8s.operator.status.ClusterState;
import org.apache.spark.k8s.operator.status.ClusterStateSummary;
import org.apache.spark.k8s.operator.status.ClusterStatus;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
import org.apache.spark.k8s.operator.utils.TestUtils;
@@ -342,6 +343,7 @@ class ClusterInitStepTest {
ClusterStateSummary.Suspended,
stateCaptor.getValue().getCurrentStateSummary());
Assertions.assertEquals(
Constants.CLUSTER_SUSPENDED_MESSAGE,
stateCaptor.getValue().getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
stateCaptor.getValue().getSuspendReason());
verify(mockContext, never()).getEventRecorder();
}
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 5e318ff7..c0d37d58 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
@@ -22,6 +22,7 @@ package
org.apache.spark.k8s.operator.reconciler.reconcilesteps;
import static java.net.HttpURLConnection.HTTP_CONFLICT;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -89,6 +90,7 @@ import
org.apache.spark.k8s.operator.spec.WorkerInstanceConfig;
import org.apache.spark.k8s.operator.status.ClusterState;
import org.apache.spark.k8s.operator.status.ClusterStateSummary;
import org.apache.spark.k8s.operator.status.ClusterStatus;
+import org.apache.spark.k8s.operator.status.SuspendReason;
import org.apache.spark.k8s.operator.utils.EventUtils;
import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
import org.apache.spark.k8s.operator.utils.TestUtils;
@@ -148,6 +150,7 @@ class ClusterSuspendStepTest {
ClusterState state = captureAppendedState();
Assertions.assertEquals(ClusterStateSummary.Suspended,
state.getCurrentStateSummary());
Assertions.assertEquals(Constants.CLUSTER_SUSPENDED_MESSAGE,
state.getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
state.getSuspendReason());
// Nothing is released until Suspended is persisted, so that a cluster
which is resumed in the
// meantime is never left in RunningHealthy without its master and workers.
Assertions.assertNotNull(get(masterStatefulSetSpec));
@@ -211,6 +214,8 @@ class ClusterSuspendStepTest {
@ValueSource(booleans = {true, false})
void stuckPodsReleaseKueueWorkloadAndAreReportedOnce(boolean suspend) {
SparkCluster cluster = buildKueueCluster(ClusterStateSummary.Suspended,
suspend);
+ SuspendReason reason = suspend ? SuspendReason.SpecSuspend :
SuspendReason.KueueEviction;
+ cluster.getStatus().getCurrentState().setSuspendReason(reason);
KubernetesClient client = spy(kubernetesClient);
// Pods stuck in terminating, e.g. on a lost node, do not hold the quota
forever
stubStuckPods(client);
@@ -231,6 +236,8 @@ class ClusterSuspendStepTest {
Constants.CLUSTER_SUSPENDED_WITH_STUCK_PODS_MESSAGE,
"cluster1-master-0, cluster1-worker-0"),
state.getMessage());
+ // The state which names the stuck pods keeps the reason of the one before
+ Assertions.assertEquals(reason, state.getSuspendReason());
// The same state is not appended again on every requeue
cluster.setStatus(cluster.getStatus().appendNewState(state));
@@ -526,6 +533,7 @@ class ClusterSuspendStepTest {
Assertions.assertEquals(
Constants.CLUSTER_EVICTED_MESSAGE + " " + reason + ": " + reason + "
by the test",
state.getMessage());
+ Assertions.assertEquals(SuspendReason.KueueEviction,
state.getSuspendReason());
// Like spec.suspend, nothing is released until Suspended is persisted,
and the Workload is
// released only after the master and worker pods are gone
Assertions.assertNotNull(get(masterStatefulSetSpec));
@@ -567,7 +575,8 @@ class ClusterSuspendStepTest {
ClusterStateSummary.Suspended,
String.format(
Constants.CLUSTER_SUSPENDED_WITH_STUCK_PODS_MESSAGE,
- "cluster1-master-0"))));
+ "cluster1-master-0"),
+ SuspendReason.KueueEviction)));
stubContext(cluster);
Assertions.assertEquals(
@@ -713,6 +722,7 @@ class ClusterSuspendStepTest {
ClusterState state = captureAppendedState();
Assertions.assertEquals(ClusterStateSummary.Suspended,
state.getCurrentStateSummary());
Assertions.assertEquals(Constants.CLUSTER_SUSPENDED_MESSAGE,
state.getMessage());
+ Assertions.assertEquals(SuspendReason.SpecSuspend,
state.getSuspendReason());
// Once that state is persisted, the cluster is held rather than suspended
again
cluster.setStatus(cluster.getStatus().appendNewState(state));
@@ -721,6 +731,28 @@ class ClusterSuspendStepTest {
verify(recorder).appendNewStateAndPersist(any(), any());
}
+ @Test
+ void evictionIsToldByTheSuspendReasonRatherThanByTheMessage() {
+ // A reworded message of a cluster suspended by an eviction before still
says so
+ SparkCluster evicted = buildEvictedCluster(false);
+ evicted.getStatus().getCurrentState().setMessage("Reworded in another
version");
+ stubContext(evicted);
+ createWorkload(evicted, evictedStatus("Preempted"), true);
+ new ClusterSuspendStep().reconcile(mockContext, recorder);
+ Assertions.assertEquals(
+ Constants.CLUSTER_REQUEUED_MESSAGE,
capturePersistedState().getMessage());
+
+ // So does a Suspended state without a reason, even if its message looks
like an eviction
+ SparkCluster unknown = buildEvictedCluster(false);
+ unknown.getStatus().getCurrentState().setSuspendReason(null);
+ stubContext(unknown);
+ createWorkload(unknown, evictedStatus("Preempted"), true);
+ clearInvocations(recorder);
+ new ClusterSuspendStep().reconcile(mockContext, recorder);
+ Assertions.assertEquals(
+ Constants.CLUSTER_RESUMED_MESSAGE,
capturePersistedState().getMessage());
+ }
+
@Test
void evictedThenSuspendedClusterIsResumedRatherThanQueuedAgain() {
SparkCluster cluster = buildEvictedCluster(true);
@@ -938,6 +970,7 @@ class ClusterSuspendStepTest {
.getStatus()
.getCurrentState()
.setMessage(Constants.CLUSTER_EVICTED_MESSAGE + " Preempted: Preempted
by the test");
+
cluster.getStatus().getCurrentState().setSuspendReason(SuspendReason.KueueEviction);
return cluster;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
