peter-toth commented on code in PR #893:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/893#discussion_r4114691231
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java:
##########
@@ -98,12 +107,30 @@ public ReconcileProgress reconcile(
statusRecorder,
new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_SUSPENDED_MESSAGE));
}
+ if (summary == ClusterStateSummary.RunningHealthy) {
+ return suspendOnKueueEviction(context, statusRecorder);
+ }
if (summary == ClusterStateSummary.Suspended) {
- Optional<ReconcileProgress> waiting = releaseResources(context,
statusRecorder);
+ boolean evicted = isSuspendedByEviction(cluster);
+ if (evicted && suspend) {
+ // The cluster 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 ClusterState(ClusterStateSummary.Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ }
+ Optional<Duration> keepWorkload = suspend ? Optional.empty() :
keepKueueWorkload(context);
Review Comment:
**Finding 2.** The ternary is wider than the helper it calls.
`keepKueueWorkload`'s javadoc opens with "Returns how long the Kueue
Workload of a cluster which is suspended **without spec.suspend** is kept", and
its whole body is about an eviction: the deactivated arm and the requeue
backoff. But `suspend ? empty : keepKueueWorkload(context)` also calls it on a
plain `spec.suspend` resume, where `evicted` is false.
That path is normally a no-op, because the `spec.suspend` release already
deleted the `Workload`, so the cache is empty and the helper returns empty. It
stops being a no-op when the `Workload` outlived that release. A
`deleteWorkloadOf` that kept failing leaves it there, and `releaseResources`
only retries the delete while `releaseWorkload` is true. So: the delete is
failing, the user clears `spec.suspend`, and someone has run `kueuectl stop
workload` on that leftover `Workload`. `keepKueueWorkload` then reports
deactivated, `releaseWorkload` goes false, and the cluster sits in `Suspended`
on a 30-minute requeue with the delete no longer even attempted, waiting for a
reactivation that has nothing to do with the resume the user asked for.
Narrow, and I am filing it Non-blocking for that reason. The fix is the
guard the javadoc already describes:
```suggestion
Optional<Duration> keepWorkload =
!suspend && evicted ? keepKueueWorkload(context) :
Optional.empty();
```
`ClusterSuspendStepTest` stays green with it, as do `checkstyleMain`,
`pmdMain` and `spotlessCheck`. Every eviction test passes `evicted == true`,
and the resume test has no `Workload` left, so nothing in the suite currently
distinguishes the two forms, which is also worth knowing.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java:
##########
@@ -122,6 +149,93 @@ public ReconcileProgress reconcile(
return proceed();
}
+ /**
+ * Suspends a running cluster whose Kueue Workload is evicted, so that its
master and workers and
+ * then its Workload are released like on spec.suspend. Kueue keeps the
quota of an evicted
+ * Workload until then. As spec.suspend is not set, the cluster is queued
again once everything is
+ * released, see {@link #keepKueueWorkload} for when that waits. An eviction
which {@link
+ * KueueWorkloadUtils#isKeptOnEviction} is only reported, and the cluster
keeps running.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status
updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
+ private ReconcileProgress suspendOnKueueEviction(
+ SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
+ Optional<Workload> cached = context.getCachedKueueWorkload();
+ Optional<Condition> eviction =
cached.flatMap(KueueWorkloadUtils::findEviction);
+ if (eviction.isEmpty()) {
+ return proceed();
+ }
+ Workload workload = cached.get();
+ String cause = eviction.get().getReason() + ": " +
eviction.get().getMessage();
+ if (KueueWorkloadUtils.isKeptOnEviction(workload)) {
+ EventUtils.warn(
+ context.getEventRecorder(),
+ EventUtils.REASON_KUEUE_EVICTION_IGNORED,
+ "Kueue evicted Workload "
+ + workload.getMetadata().getName()
+ + " ("
+ + cause
+ + "), which the operator does not act on, so the master and
workers keep running "
+ + "and holding its quota. Set spec.suspend to true or delete the
cluster to release "
+ + "them.");
+ return proceed();
+ }
+ log.info("Kueue evicted the Workload of the cluster ({}), suspending it.",
cause);
+ return appendStateAndImmediateRequeue(
+ context,
+ statusRecorder,
+ new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_EVICTED_MESSAGE + " " + cause));
+ }
+
+ /**
+ * 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.
+ *
+ * @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
Review Comment:
**Finding 1.** This clause is what stops the branch at `:115-122` from
re-firing, and nothing in the suite notices if it goes.
The code is right as written. The problem is that it reads like a cosmetic
refinement of the resume message, so a later edit would take it out without
resistance. It is not cosmetic: it is the loop terminator. Once the `evicted &&
suspend` branch appends `Suspended`/`CLUSTER_SUSPENDED_MESSAGE`, this clause is
the only thing that makes the next scan stop there instead of walking past it
to the older `Suspended`/`CLUSTER_EVICTED_MESSAGE` entry and reporting
`evicted` again.
Measured. I deleted the clause, leaving `if (state.getCurrentStateSummary()
!= ClusterStateSummary.Suspended)`, and ran `ClusterSuspendStepTest` and
`KueueWorkloadUtilsTest`: both **green**. Then I drove six reconciles of an
evicted cluster with `spec.suspend` set, simulating each persist:
| arm | per-reconcile progress | immediate requeues in 6 | history entries |
|---|---|---|---|
| head | `PT0S` once, then `PT30M` | 1 | 3 |
| clause deleted | `PT0S` every time | 6 | 8 |
So the regression is a hot loop that writes a status per pass, not a wrong
message.
Two assertions close it, and I verified each discriminates. The first is a
second reconcile on the existing `suspendingEvictedClusterIsRecorded`, which is
already set up for it:
```java
// The marker stops the branch from firing again
cluster.setStatus(cluster.getStatus().appendNewState(state));
Assertions.assertEquals(
SUSPEND_HOLD_PROGRESS, new
ClusterSuspendStep().reconcile(mockContext, recorder));
verify(recorder).appendNewStateAndPersist(any(), any());
```
The second pins the read side, which is the reason the marker is written at
all. I ran this one both ways: head gives `CLUSTER_RESUMED_MESSAGE`, the mutant
gives `CLUSTER_REQUEUED_MESSAGE`.
```java
@Test
void evictedThenSuspendedClusterIsResumedRatherThanQueuedAgain() {
SparkCluster cluster = buildEvictedCluster(true);
stubContext(cluster);
createWorkload(cluster, evictedStatus("Preempted"));
new ClusterSuspendStep().reconcile(mockContext, recorder);
cluster.setStatus(cluster.getStatus().appendNewState(captureAppendedState()));
// Held by spec.suspend since the eviction, so clearing it is a resume,
not a requeue
cluster.getSpec().setSuspend(false);
new ClusterSuspendStep().reconcile(mockContext, recorder);
Assertions.assertEquals(Constants.CLUSTER_RESUMED_MESSAGE,
capturePersistedState().getMessage());
}
```
While I am here: keying the decision on a message prefix is the part that
makes this fragile, and I checked the upgrade direction before raising it. A
cluster left `Suspended` across an operator upgrade that changed
`CLUSTER_EVICTED_MESSAGE` fails the `startsWith` and reads as not evicted, so
it resumes with the wrong message rather than looping. That is the safe
direction, which is why I am asking for the test rather than for a status field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
