dongjoon-hyun commented on code in PR #893:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/893#discussion_r4116896441
##########
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:
Thank you, @peter-toth. Addressed in cf1ad0c8 by adding both assertions: the
second reconcile in `suspendingEvictedClusterIsRecorded` and the new
`evictedThenSuspendedClusterIsResumedRatherThanQueuedAgain`. Both fail once `||
CLUSTER_SUSPENDED_MESSAGE.equals(message)` is removed.
##########
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:
Addressed in cf1ad0c8 with `evicted ? keepKueueWorkload(context) :
Optional.empty()`, which is equivalent to the suggestion because the `evicted
&& suspend` case returns above. The new
`resumedClusterReleasesDeactivatedWorkload` covers your scenario and fails
without this change, so the suite now distinguishes the two forms.
--
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]
