Repository navigation
[SPARK-59542] Support spec.suspend to hold driver and master/worker creation in AppInitStep and ClusterInitStep - #828
dongjoon-hyun wants to merge 5 commits into
Conversation
…reation in `AppInitStep` and `ClusterInitStep`
There was a problem hiding this comment.
Thanks for the PR, @dongjoon-hyun!
AppInitStep and ClusterInitStep now return completeAndDefaultRequeue() before requesting the driver pod or the master / worker StatefulSets. The gate sits in the right place. Validation, cleanup and deletion all run ahead of it, and the restart backoff is skipped rather than reset. Two things block for me. Nothing in a suspended reconcile writes .status, so the two new *-suspended.yaml assertions have no status to match and the e2e group fails. And the doc's "no effect on a running application" does not hold for an app with a restart policy.
Blocking
- 1. Suspended resource never gets a
.status: every step of the suspended pipeline is a no-op on the status recorder, andtoUpdateControlreturnsnoUpdate(), so the API server keeps the resource with no.statusat all.tests/e2e/suspendassertsstatus.currentStateandstatus.stateTransitionHistoryfor both the app and the cluster, so the new group fails. [inline:tests/e2e/suspend/spark-application-suspended.yaml:25] - 2. Doc contradicts the
ScheduledToRestarthold: "Setting it totrueon a running application or cluster has no effect" is wrong for an app configured to restart. The current attempt keeps running, but the next one is held indefinitely, which is whatsuspendedAppScheduledToRestartDoesNotRequestDriverasserts. [inline:docs/spark_custom_resources.md:549]
Non-blocking
- 3. No app-side test for the non-initializing case:
ClusterInitStepTest.nonInitializingClusterProceedspins that aRunningHealthycluster ignoressuspend.AppInitStepTesthas no twin, and finding 2 is about exactly that boundary. [inline:AppInitStepTest.java:444]
Minor
- 4.
log.infoon every requeue: a held resource logs one INFO line per reconcile interval forever. [inline:ClusterInitStep.java:63] - 5. Assertion files duplicate
tests/e2e/assertions/: the two*-state-transition.yamlfiles are the shared assertions with a different resource name. [inline:tests/e2e/suspend/spark-application-state-transition.yaml:26]
| namespace: default | ||
| spec: | ||
| suspend: true | ||
| status: |
There was a problem hiding this comment.
Finding 1. Nothing in a suspended reconcile ever writes .status to the API server, so this assertion (and the spark-cluster-suspended.yaml twin) has no status to match.
The trace, for a SparkApplication created with suspend: true:
AppValidateSteppersists only when the status is invalid, and it never is.CustomResource's constructor callsinitStatus(), sogetStatus()returns aSubmittedApplicationStatuseven for a CR the server stored without one.isValidApplicationStatuspasses and the step returnsproceed().AppCleanUpStepreturnsproceed()forSubmittedwithout persisting.- the new branch returns
completeAndDefaultRequeue()before any persist. ReconcilerUtils.toUpdateControlreturnsUpdateControl.noUpdate(), and the CRD declaressubresources: status: {}, so JOSDK writes nothing either.
SparkCluster is the same, except ClusterValidateStep is an unconditional proceed(), so there is not even a reset path.
I ran the three steps of each pipeline against a mock recorder:
SparkApplication app = new SparkApplication();
app.setMetadata(new ObjectMetaBuilder().withName("a").withNamespace("default").build());
app.getSpec().setSuspend(true);
SparkAppContext ctx = mock(SparkAppContext.class);
when(ctx.getResource()).thenReturn(app);
SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
new AppValidateStep().reconcile(ctx, recorder);
new AppCleanUpStep().reconcile(ctx, recorder);
new AppInitStep().reconcile(ctx, recorder);
verifyNoInteractions(recorder); // passes; the ClusterValidate/Terminated/Init trio passes tooSo a held resource has no status on the server and an empty Current State printer column. Chainsaw then fails here on status.currentState, and on (*.currentStateSummary) over a missing stateTransitionHistory. I have no cluster here, so that last hop is read from the assertion files rather than observed.
Two ways out:
- Persist the initial status once when entering the hold, so the resource really is
Submittedon the server. Careful:StatusRecorder.updateStatusFromCacheseedsstatusCachewith the current status on a cache miss, andpatchAndStatusWithVersionLockedshort-circuits onnewStatusNode.equals(previousStatusNode). A plainpersistStatus(context, app.getStatus())in the new branch is therefore dropped silently on the first reconcile. - Or drop the
status:blocks fromspark-application-suspended.yamlandspark-cluster-suspended.yamland assert onlyspec.suspend: trueplus the existingerror:checks on the driver pod and the StatefulSets.
The first also fixes the operator-facing half. With Kueue holding a queue of workloads, kubectl get sparkapp currently shows a blank state for every one of them.
| ``` | ||
|
|
||
| * `suspend` only takes effect before the driver (or master / worker) resources are requested. | ||
| Setting it to `true` on a running application or cluster has no effect in the current version. |
There was a problem hiding this comment.
Finding 2. Not true for a SparkApplication configured with a restart policy.
AppCleanUpStep has no suspend check, so a running attempt that fails still goes through terminateOrRestart and lands in ScheduledToRestart (AppCleanUpStep.java:186-193). The next reconcile hits the new gate and holds there until someone flips the flag back. suspendedAppScheduledToRestartDoesNotRequestDriver asserts exactly that hold. So suspend: true on a running app does not stop the current attempt, but it does stop every attempt after it.
For SparkCluster the bullet is accurate: SparkClusterReconciler.getReconcileSteps only adds ClusterInitStep for Submitted, and a RunningHealthy cluster never goes back there.
| Setting it to `true` on a running application or cluster has no effect in the current version. | |
| Setting it to `true` on a running application does not stop the current attempt. If the | |
| application is configured to restart, the next attempt is held until `suspend` is set back to | |
| `false`. Setting it to `true` on a running cluster has no effect in the current version. |
The PR description's "Suspending an already running application or cluster is out of scope" reads the same way and is worth the same correction.
| Assertions.assertEquals( | ||
| ApplicationStateSummary.DriverRequested, | ||
| application.getStatus().getCurrentState().getCurrentStateSummary()); | ||
| } |
There was a problem hiding this comment.
Finding 3. Worth an app-side twin of ClusterInitStepTest.nonInitializingClusterProceeds: a RunningHealthy app with suspend: true must still return proceed(). That is the boundary finding 2 is about, and right now it is pinned only for SparkCluster.
I ran this against the PR head and it passes:
@Test
void suspendedNonInitializingAppProceeds() {
AppInitStep appInitStep = new AppInitStep();
SparkAppContext mockContext = mock(SparkAppContext.class);
SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
SparkApplication application = new SparkApplication();
application.setMetadata(applicationMetadata);
application.getSpec().setSuspend(true);
application.setStatus(
application
.getStatus()
.appendNewState(
new ApplicationState(ApplicationStateSummary.RunningHealthy, "running")));
when(mockContext.getResource()).thenReturn(application);
Assertions.assertEquals(
ReconcileProgress.proceed(), appInitStep.reconcile(mockContext, recorder));
verifyNoInteractions(recorder);
}| } | ||
| SparkCluster cluster = context.getResource(); | ||
| if (cluster.getSpec().isSuspend()) { | ||
| log.info("Cluster is suspended, master and worker resources would not be requested."); |
There was a problem hiding this comment.
Finding 4. A held resource logs this on every reconcile, so once per spark.kubernetes.operator.reconciler.intervalSeconds (120s by default) for as long as it stays suspended. With Kueue holding a few hundred workloads that is steady INFO traffic for a no-op. The other steady-state no-op paths report at debug (AppCleanUpStep.java:132, AppReconcileStep.java:86). Same line at AppInitStep.java:71.
| spec: | ||
| suspend: false | ||
| status: | ||
| stateTransitionHistory: |
There was a problem hiding this comment.
Finding 5. This is tests/e2e/assertions/spark-application/spark-state-transition.yaml with a different resource name, and spark-cluster-state-transition.yaml is the cluster one. AGENTS.md puts shared assertions in tests/e2e/assertions/. Binding the name there (name: ($SPARK_APPLICATION_NAME), as the namespace already is) would let this group reuse both files the way state-transition/chainsaw-test.yaml does, leaving only the two *-suspended.yaml files here. The extra spec.suspend: false check can move into an inline resource: assert.
|
Thank you for the review, @peter-toth. All five findings are addressed in 5351a9a.
|
|
Could you review this PR when you have some time, @viirya ? |
There was a problem hiding this comment.
Re-checked through 5351a9a. Findings 1, 2, 3, 4 and 5 are resolved and nothing regressed.
I re-ran the six new unit tests against ac5b0ee with both init steps reverted. Four fail there. The two nonInitializing*Proceeds guards pass on base, which is what a guard should do.
tests/e2e/state-transition already bound SPARK_APPLICATION_NAME and SPARK_CLUSTER_NAME at scenario level, so the parameterized shared assertions needed no change there. watched-namespaces and watched-namespaces-file were the only two consumers that did.
Blocking
- 6. Docs promise a
Submittedstate the user cannot see (late catch): the new section says the operator "keeps the resource in its initializing state (Submitted, ...)". A resource created withsuspend: truehas no.statuson the API server at all.kubectl get sparkappshows an emptyCurrent State, which is this repo's own signature for "not managed". One sentence in the docs fixes it. inline:docs/spark_custom_resources.md:533
Non-blocking
- 7. The two
*-suspended.yamlassertions cannot fail (new): each file's entire body isspec.suspend: true. That is the value the step applied two lines earlier, on a field the operator never writes. Theerror:checks carry the step on their own. inline:tests/e2e/suspend/spark-application-suspended.yaml:24 - 8. The hold is keyed on state, not on whether the driver was already requested (late catch): if a reconcile creates the driver pod and then fails to persist
DriverRequested, the app is back atSubmittedwith a live driver. Flippingsuspendon at that point holds it there forever, and no observer runs inSubmitted. Probably best carried into the running-attempt suspend you have planned. inline:.../AppInitStep.java:70
Minor
- 9. Kueue bullet describes an integration that is not wired yet (late catch):
KueueWorkloadFactoryhas no caller insrc/main, so nothing creates aWorkloadand nothing flipssuspendback. inline:docs/spark_custom_resources.md:554
| Both `SparkApplication` and `SparkCluster` support `.spec.suspend`. When it is set to `true`, the | ||
| operator keeps the resource in its initializing state (`Submitted`, or `ScheduledToRestart` for an | ||
| application that is scheduled to restart) and does not request the driver pod or the master / worker | ||
| StatefulSets. Setting it back to `false` resumes the regular lifecycle. |
There was a problem hiding this comment.
Finding 6. Submitted is not something the user can observe here. With the dedicated Suspended state deferred to a follow-up (#issuecomment-5684180823), the docs are the only place that can say so.
For a SparkApplication created with suspend: true, no step in the pipeline writes a status. AppValidateStep persists only when the status is invalid and it never is. AppCleanUpStep returns proceed() for Submitted. The new branch returns before any persist. toUpdateControl is noUpdate(). So .status is absent on the API server and the Current State printer column is blank.
That is indistinguishable from "the operator is not watching this namespace". tests/e2e/watched-namespaces/chainsaw-test.yaml:54-58 asserts .status == null as the signature of exactly that case.
Worth stating precisely, because the paragraph is only wrong for one of the two holds. An application suspended later, in ScheduledToRestart, does have a status on the server. AppCleanUpStep wrote it before the app got there.
| StatefulSets. Setting it back to `false` resumes the regular lifecycle. | |
| StatefulSets. Setting it back to `false` resumes the regular lifecycle. | |
| `Submitted` here is the operator's in-memory view. A resource created with `suspend: true` has no | |
| `.status` on the API server at all, so `kubectl get` shows an empty `Current State` until it | |
| resumes. An application suspended later, in `ScheduledToRestart`, keeps the status its previous | |
| attempt already wrote. |
There was a problem hiding this comment.
This distinction is still missing from the documentation at the current head. A valid resource created with suspend: true does not have its initial status persisted, so Submitted is an in-memory state and the Current State column remains blank. An application held in ScheduledToRestart, however, retains the status already written by cleanup.
Deferring a dedicated Suspended state seems reasonable for this PR, but please document the observable behavior explicitly. For example:
For a valid resource created with
suspend: true, the initialSubmittedstatus is not persisted to the API server, sokubectl getshows an emptyCurrent Stateuntil initialization resumes. An application held inScheduledToRestartretains its previously persisted status.
The PR description's statement that a held resource has no .status should likewise be limited to the initial-submission case.
| name: spark-job-suspend-test | ||
| namespace: default | ||
| spec: | ||
| suspend: true |
There was a problem hiding this comment.
Finding 7. This file, and spark-cluster-suspended.yaml, now assert only spec.suspend: true. That is the value spark-example-suspend.yaml applied two steps earlier, on a field the operator never writes. The assertion cannot fail.
Dropping the status: blocks was the right call. What is left is a 24-line file that adds nothing on top of the error: checks below it.
Two options. Delete both files and let the error: checks carry the step. Or make them assert the hold, using the idiom already in the tree at tests/e2e/watched-namespaces/chainsaw-test.yaml:54-58:
- script:
timeout: 30s
content:
kubectl get sparkapplication spark-job-suspend-test -n default -o json | jq ".status"
check:
(contains($stdout, 'null')): trueThat pins today's behavior positively rather than by the absence of a pod. It will need updating when the Suspended state lands, which seems right. That follow-up is exactly the kind of change this test should notice.
| return proceed(); | ||
| } | ||
| SparkApplication app = context.getResource(); | ||
| if (app.getSpec().isSuspend()) { |
There was a problem hiding this comment.
Finding 8. The gate asks "is the app initializing and suspended?", not "has anything been requested yet?". Those two differ in one window, and appInitStepShouldBeIdempotentWhenStatusUpdateFails (AppInitStepTest.java:206) exists because the codebase already knows about it.
The sequence that produces it:
- Reconcile N creates the pre-resources, the driver pod and the driver resources.
attemptStatusUpdatefails to persistDriverRequestedand returnscompleteAndImmediateRequeue().StatusRecorder.persistStatusnever updatedstatusCache, soupdateStatusFromCacheputs the resource back toSubmittedon reconcile N+1.- A driver pod is live and the app reads as
Submitted.
An operator restart between the pod create and the status patch lands in the same place. The server has no .status, so initStatus() hands back a fresh Submitted.
If suspend goes to true in that window, this branch fires and the app never leaves Submitted. SparkAppReconciler.getReconcileSteps adds only AppInitStep for Submitted, so no observer and no timeout ever looks at that pod. It runs to completion unnoticed and the resource is held forever. That contradicts docs/spark_custom_resources.md:548, "suspend only takes effect before the driver ... resources are requested".
I reproduced it as a unit test at head. Reconcile 1 with persistStatus returning false, reset the status the way updateStatusFromCache would, set suspend, reconcile 2:
ReconcileProgress p2 = appInitStep.reconcile(ctx, recorder);
Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), p2);
Assertions.assertEquals(
ApplicationStateSummary.Submitted,
application.getStatus().getCurrentState().getCurrentStateSummary());
// the driver pod from reconcile 1 is still there
Assertions.assertNotNull(
kubernetesClient.pods().inNamespace("default").withName("driver-pod").get());It passes, so the hold wins over the live driver.
The narrow fix here is if (app.getSpec().isSuspend() && context.getDriverPod().isEmpty()), which makes the code match the documented contract. One caveat I could not rule out. getDriverPod() reads the informer cache, and in ScheduledToRestart a just-deleted pod from the previous attempt could still be in it, which would make the operator start a driver for a suspended app.
If that risk is not worth taking in this PR, the running-attempt suspend you have planned is the natural home for this window. A suspend that can stop a live attempt stops this one too. Worth carrying into that work rather than leaving it only in this thread.
There was a problem hiding this comment.
I agree with this finding, but I think it should be addressed before merging rather than deferred to support for suspending running attempts.
Resource creation and status persistence are separate operations. If the driver is created but persisting DriverRequested fails, the next reconciliation restores Submitted from the status cache. An operator restart between those operations produces the same discrepancy.
If suspend becomes true at that point, this branch prevents initialization recovery. SparkAppReconciler selects only validation, cleanup, and initialization for Submitted, so the existing driver is no longer observed for completion or timeout while the flag remains set. The equivalent window also exists in ClusterInitStep between resource creation and persisting RunningHealthy.
Could we distinguish an initialization that has already started from one that has not requested resources yet, and allow the former to recover? Please add a regression test covering successful resource creation, failed status persistence, and then enabling suspend.
I would avoid fixing this solely with getDriverPod().isEmpty(): that lookup uses the informer cache, and the driver labels do not distinguish attempts, so a previous attempt's pod could incorrectly bypass the hold.
There was a problem hiding this comment.
The assumption that the driver pod name carries the attempt ID does not hold for all supported configurations.
SparkAppSubmissionWorker.buildDriverConf() preserves a user-specified spark.app.id through setIfMissing(), and SparkAppDriverConf.resourceNamePrefix() returns that ID. The existing checkAppIdWhenUserSpecifiedInSparkConf test explicitly covers this behavior. Spark also allows spark.kubernetes.driver.pod.name to override the pod name directly.
Consequently, consecutive attempts can have the same expected pod name. If the previous attempt's pod remains in the informer cache after cleanup, this comparison returns true for a suspended ScheduledToRestart application. Once backoff has elapsed, initialization proceeds; if the old pod has already disappeared from the API server, it can create a new driver despite suspend: true.
Could we identify the current attempt independently of a potentially reused pod name, or explicitly validate any naming restrictions required by this approach? Please extend previousAttemptDriverPodDoesNotBypassSuspend to cover a fixed spark.app.id and an explicitly configured driver pod name. Its current use of two different names does not exercise this case.
There was a problem hiding this comment.
This handles a previous attempt's pod once the informer has observed its deletion, but the stale pre-deletion snapshot can still pass this filter.
The remaining sequence is:
- Consecutive attempts reuse the driver pod name.
- Cleanup successfully deletes the old pod and persists
ScheduledToRestart. - With
suspend: trueand zero or elapsed backoff, the next reconcile still sees the old informer snapshot: the expected name and a nulldeletionTimestamp. - This helper returns the old pod, so
AppInitStepbypasses the hold. - Initialization queries the API server, finds that the pod is gone, and creates a new driver despite suspend.
I agree that this is informer lag, but the missing-driver grace period does not make this path safe. That flow defers a decision and eventually performs live verification; this path immediately authorizes resource creation. The grace-period observer also does not run for ScheduledToRestart.
Could we perform a live lookup by the desired pod name before allowing the suspend bypass, require the pod to exist and not be terminating, and requeue if verification fails?
The regression test should keep the pre-deletion pod snapshot in the informer while the API server reports the pod absent, then verify that a suspended application creates no driver. The current test sets deletionTimestamp on the cached object, so it covers the already-updated cache rather than this remaining window.
There was a problem hiding this comment.
Confirmed resolved in 18636ac. The live GET prevents the stale pre-deletion snapshot from authorizing initialization, and lookup failures keep the hold. staleInformerSnapshotDoesNotBypassSuspend now covers the specific sequence discussed here and verifies that no driver is created and the application stays in ScheduledToRestart. Thanks for adding this coverage.
| `false`. Setting it to `true` on a running cluster has no effect in the current version. | ||
| * Deleting a suspended resource works as usual. | ||
| * This is the building block for external job queueing systems such as | ||
| [Kueue](https://kueue.sigs.k8s.io/), which admit a workload by flipping `suspend` to `false`. |
There was a problem hiding this comment.
Finding 9. Reads as if Kueue does this today. KueueWorkloadFactory is the only code outside the two init steps that touches spec.isSuspend(), and it has no caller in src/main. Nothing creates a Workload, and nothing flips suspend back.
| [Kueue](https://kueue.sigs.k8s.io/), which admit a workload by flipping `suspend` to `false`. | |
| [Kueue](https://kueue.sigs.k8s.io/), which admit a workload by flipping `suspend` to `false`. | |
| The operator does not integrate with such a system yet. |
|
I will take a look. |
There was a problem hiding this comment.
Thanks for the update. I reviewed this against 5351a9a, including the earlier review threads and the interactions with status recovery, cleanup, and restart handling.
The two init gates are small and consistent with the existing reconciliation structure. The first-round findings have been addressed, and the CI run for this commit is green, including both suspend e2e jobs.
I still have two correctness concerns:
- If resource creation succeeds but the status update fails, enabling suspend prevents the existing initialization recovery path from completing. A driver can remain live while the application stays in
Submitted, where no driver observers run. - Time spent suspended is included in the attempt duration used by
restartCounterResetMillis. A long hold can therefore reset retry counters even when the resumed attempt fails immediately.
I would address these interactions and add regression coverage before merging. The suspend documentation also needs to distinguish the in-memory Submitted state from the status actually visible on the API server.
Two smaller suggestions:
- The
spec.suspend: trueassertions confirm the applied value, but do not establish that the operator processed the hold. The absence-of-resource checks and successful resume provide the behavioral coverage. Replacing these assertions with.status == nullwould not establish reconciliation either. - Please clarify that this is a prerequisite for future Kueue integration; the operator does not yet wire
KueueWorkloadFactoryinto workload creation or admission handling.
| if (app.getSpec().isSuspend()) { | ||
| log.debug("Application is suspended, driver resources would not be requested."); | ||
| return completeAndDefaultRequeue(); |
There was a problem hiding this comment.
Holding the application here also preserves the initializing state's timestamp. This affects the existing restart-counter logic: ApplicationStatus.calculateCurrentAttemptDuration() measures from the latest Submitted or ScheduledToRestart, and terminateOrRestart() applies the time-based counter reset before checking retry limits.
For example, with maxRestartAttempts: 1 and restartCounterResetMillis: 3600000, hold the permitted retry in ScheduledToRestart for over an hour, then resume it and let it fail immediately. The suspended hour satisfies the reset threshold, so the counter resets and another retry is allowed instead of exhausting the configured limit.
This makes time spent waiting for admission count toward the documented reset for a long-running attempt. Could we exclude suspended time from that calculation and cover this sequence with both trimmed and untrimmed histories?
Simply changing the initializing timestamp on resume would also change restart-backoff and history semantics, so those should be preserved when addressing this.
|
Thank you for the second round, @peter-toth and @viirya. All findings are addressed in 34dbda1. 6. Docs / 7. 8. Hold keyed on state, not on whether the driver was requested (peter-toth, viirya): the hold now applies only when the driver pod of the current attempt does not exist. 9. Kueue bullet (peter-toth, viirya): applied the suggestion. The PR description now says this is a prerequisite and that Suspended time counted toward |
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks for addressing the previous comments. I re-reviewed 34dbda1.
The documentation now distinguishes the in-memory Submitted state from persisted status and clarifies the scope of the Kueue integration. The attempt-duration change also addresses the suspended-time issue while preserving the restart-backoff timestamp.
The initialization recovery change is an improvement, but I do not think Finding 8 is fully resolved yet. The new helper assumes that pod names uniquely identify attempts, which does not hold when users configure spark.app.id or spark.kubernetes.driver.pod.name. It also checks the name only after selecting an arbitrary driver pod from the informer cache.
Could we tighten the current-attempt lookup and extend the regression coverage for these cases before merging?
| private boolean isDriverRequested(SparkAppContext context) { | ||
| Optional<Pod> driverPod = context.getDriverPod(); | ||
| return driverPod.isPresent() |
There was a problem hiding this comment.
context.getDriverPod() filters by application/driver labels and then calls findAny(). The name comparison here happens after that selection.
If both an older attempt's pod and the current attempt's pod are present in the informer cache, this can select the older one and return false without checking the current pod. In the status-persistence failure scenario this helper is intended to recover, that leaves the application held even though its current driver exists.
Could we apply the current-attempt predicate before selecting a pod, rather than checking only the single pod returned by getDriverPod()? A regression test with both pods available, with the older pod encountered first, would cover this. The existing tests mock only one returned pod, so they cannot detect this selection issue.
|
Thank you for the careful follow-up, @viirya. Both points are addressed in 67d1cfc. Selection before the name check: Reused pod names ( I considered labeling the driver pod with the attempt id, which would make the identification exact regardless of naming. That changes every driver pod the operator creates and is independent of The |
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks for the update. I re-reviewed 67d1cfc.
The selection-order issue is resolved: the current-attempt predicate is now applied before selecting a pod, and the new test covers the older pod appearing first. The documentation and restart-duration fixes also remain addressed.
I still think Finding 8 needs one more change. Filtering on the cached deletionTimestamp handles a terminating pod only after the informer has observed the deletion. It does not cover the stale pre-deletion snapshot described in the previous comment.
That distinction matters here because a false positive allows initialization to create a new driver despite suspend: true. Could we verify the pod against the API server before using its existence to bypass suspend, and add a regression test for that stale-snapshot case? This can stay local to the suspend recovery path without changing labels on every driver pod.
|
Thank you, @viirya. You are right that the cached Live verification before the bypass: Regression coverage:
The javadoc and the PR description describe the verification step. The Java 21 CI failure on the previous run was |
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks for addressing the review feedback. I re-reviewed 18636ac, including the earlier discussion threads.
The live lookup now closes the stale-cache case: a cached pod only permits the suspend bypass after the API server confirms that the pod exists and is not terminating. Verification errors keep the application suspended and requeue.
The new regression test covers the pre-deletion informer snapshot with the pod already absent from the API server, using the real SparkAppContext. The selection-order, restart-duration, and documentation concerns are also addressed.
All 35 CI jobs for this commit passed, including the suspend e2e tests on Kubernetes 1.34 and 1.37.
I have no remaining blocking concerns. LGTM.
|
Thank you, @viirya and @peter-toth ! This is the first step and I'm going to add future steps. It will become more mature than the AS-IS status. |
|
Merged to main |
3 participants
What changes were proposed in this pull request?
This PR aims to honor
spec.suspendinAppInitStepandClusterInitStep.spec.suspend: trueSparkApplicationSubmitted,ScheduledToRestartcompleteAndDefaultRequeue()SparkClusterSubmittedcompleteAndDefaultRequeue()suspend: truetherefore has no.statuson the API server until it resumes, while an application held inScheduledToRestartkeeps the status its previous attempt already wrote. A dedicatedSuspendedstate is planned in a follow-up.suspendis set meanwhile, so a live driver is never left unobserved. The current attempt's driver is the driver-labeled pod that has the name of the desired driver pod spec and is not terminating. The clean-up step deletes the previous attempt's driver before the application is scheduled to restart, so a previous attempt's pod is either gone or terminating even when the pod name is reused across attempts (e.g. user-specifiedspark.app.id). Because the informer cache may still hold the pre-deletion snapshot of that pod, the candidate is verified against the API server and the hold is bypassed only if the pod exists there and is not terminating; an API error keeps the hold for this reconciliation.spec.suspendback tofalsetriggers a reconcile and enters the existing init path.spec.suspend.spec.suspend: trueon a running application does not stop the current attempt. If the application is configured to restart, the next attempt is held inScheduledToRestart. A running cluster is not affected.restartCounterResetMillisis now measured from the first state afterSubmitted/ScheduledToRestart(normallyDriverRequested), so time spent suspended or in restart backoff no longer counts as a successful run.tests/e2e/suspendand aSuspendsection todocs/spark_custom_resources.md.tests/e2e/assertions/so thattests/e2e/suspendcan reuse them.Suspending a running attempt itself is out of scope and will be handled in a follow-up.
Why are the changes needed?
spec.suspendwas added in SPARK-59475 but the operator does not read it yet. This is a prerequisite for the Kueue integration: job queueing systems such as Kueue create a workload withsuspend: trueand flip it tofalseonce quota is available. The operator does not wireKueueWorkloadFactoryinto workload creation or admission handling yet.Does this PR introduce any user-facing change?
Yes, in one place.
spec.suspenditself is not released yet, but the attempt duration used byrestartCounterResetMillis(released in 1.0.0) is now measured fromDriverRequestedinstead ofSubmitted/ScheduledToRestart. Time spent in restart backoff (and suspended) no longer counts as a successful run, so an attempt whose running time is withinrestartBackoffMillisofrestartCounterResetMillisno longer resets the restart counters and may reachmaxRestartAttemptsearlier than before. This only affects applications withrestartCounterResetMillis >= 0(default-1).spec.suspendtruefalserestartCounterResetMillisattempt durationSubmitted/ScheduledToRestart(includes backoff)DriverRequested(excludes backoff and suspended time)How was this patch tested?
Pass the CIs.
AppInitStepTestSubmittedandScheduledToRestart, resume aftersuspend: false, proceed inRunningHealthy, recover when the driver was created but the status update failed, previous attempt's driver pod does not bypass the hold, stale informer snapshot of a deleted pod does not bypass the holdSparkAppContextTestClusterInitStepTestSubmitted, proceed inRunningHealthy, recover when the master StatefulSet already existsApplicationStatusTestScheduledToRestartdoes not reset the restart counter (trimmed and untrimmed history)tests/e2e/suspend.status, no pod / StatefulSet, then patch and run to completionWas this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5.1