dongjoon-hyun commented on code in PR #952:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/952#discussion_r4211680670
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java:
##########
@@ -113,30 +129,104 @@ public ReconcileProgress reconcile(
// 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()));
Review Comment:
Thank you for the review and approval, @peter-toth. You're right. I added
the sentence to the description of `suspendHoldRequeueIntervalSeconds`, which
now covers the hold of an evicted `SparkApplication` or `SparkCluster` whose
`Workload` is deactivated, and regenerated `config_properties.md`.
--
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]
