dongjoon-hyun commented on code in PR #895:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/895#discussion_r4119290980
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -73,17 +74,24 @@ public ReconcileProgress reconcile(
}
SparkApplication app = context.getResource();
if (app.getSpec().isSuspend()) {
- Optional<Pod> currentAttemptDriverPod;
+ boolean driverRequested = false;
try {
- currentAttemptDriverPod = context.getCurrentAttemptDriverPod();
+ driverRequested = isDriverRequested(context);
} catch (KubernetesClientException e) {
// Whether a driver of this attempt is live is unknown, not answered.
Holding would claim
// in an event that none was requested, and would do so for the whole
suspend hold
- // interval, so look again with the steady-state interval instead.
+ // interval. It would also release the Kueue quota of a running
driver, so look again with
+ // the steady-state interval instead.
log.warn("Failed to verify the driver pod of a suspended application,
will retry.", e);
return completeAndDefaultRequeue();
+ } catch (Exception e) {
Review Comment:
Thank you, @MaxGekk. You're right. Addressed in ea8b9997 by narrowing the
catch to the spec build of the API lookup in `isDriverRequested`, which is
reached only after the cache check came back empty, and only for a suspended
application. A spec failure for a cached driver propagates again like before
this PR and `ClusterInitStep`, which the new
`suspendedAppWithCachedDriverAndUnbuildableSpecKeepsKueueWorkload` verifies. A
non-suspended application still gets `SchedulingFailure` for it, see
`driverMissingFromCacheWithUnbuildableSpecFailsScheduling`.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -239,12 +250,19 @@ private boolean isDriverRequested(SparkAppContext
context) {
Optional<Workload> workload = context.getCachedKueueWorkload();
if (workload.isEmpty()
Review Comment:
Yes, the cached `Workload` can lag behind the live read of
`requestAdmission` too, so this PR narrows that pre-existing window rather than
closing it. Looking the driver up whenever a `Workload` may exist is affordable
only if the lookup does not build the driver spec (see your comment on the spec
build), so I'll handle both in a follow-up JIRA: look up the driver by a name
derived without the spec, and then do it for every suspended application like
`ClusterInitStep.isMasterRequested`.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -73,17 +74,24 @@ public ReconcileProgress reconcile(
}
SparkApplication app = context.getResource();
if (app.getSpec().isSuspend()) {
- Optional<Pod> currentAttemptDriverPod;
+ boolean driverRequested = false;
try {
- currentAttemptDriverPod = context.getCurrentAttemptDriverPod();
+ driverRequested = isDriverRequested(context);
Review Comment:
Addressed in ea8b9997. `driverRequested` is computed once in the suspend
branch and passed to `holdForKueueAdmission`, so a driver found there is not
looked up again. `verify(mockContext).getCurrentAttemptDriverPod()` in
`suspendedAppWithDriverMissingFromCacheKeepsItsKueueWorkload` fails without it.
Note that a pod deleted between the two lookups would still be created again by
`getOrCreateSecondaryResource` in the same reconcile, so the single evaluation
saves the call rather than closing that window.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -239,12 +250,19 @@ private boolean isDriverRequested(SparkAppContext
context) {
Optional<Workload> workload = context.getCachedKueueWorkload();
if (workload.isEmpty()
|| !KueueWorkloadUtils.isAdmitted(workload.get())
- || (KueueWorkloadUtils.findEviction(workload.get()).isEmpty()
+ || (!context.getResource().getSpec().isSuspend()
+ && KueueWorkloadUtils.findEviction(workload.get()).isEmpty()
&& !KueueWorkloadUtils.isDeactivated(workload.get()))) {
return false;
}
Pod driverPod =
context.getClient().resource(context.getDriverPodSpec()).get();
Review Comment:
Thank you for the analysis. Right, the spec build uploads the dependencies
and runs again after `setKueuePodSetFlavors`. The same already happens for the
lookup of SPARK-59800 and for a cached driver in
`getCurrentAttemptDriverPod()`, so I'll fix them together in a follow-up JIRA
by deriving the driver pod name (`spark.kubernetes.driver.pod.name`, or
`<appId>-driver`) without building the spec. That removes the spec-build
failure mode behind the catch as well.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -239,12 +250,19 @@ private boolean isDriverRequested(SparkAppContext
context) {
Optional<Workload> workload = context.getCachedKueueWorkload();
if (workload.isEmpty()
|| !KueueWorkloadUtils.isAdmitted(workload.get())
- || (KueueWorkloadUtils.findEviction(workload.get()).isEmpty()
+ || (!context.getResource().getSpec().isSuspend()
+ && KueueWorkloadUtils.findEviction(workload.get()).isEmpty()
&& !KueueWorkloadUtils.isDeactivated(workload.get()))) {
return false;
}
Pod driverPod =
context.getClient().resource(context.getDriverPodSpec()).get();
- return driverPod != null && driverPod.getMetadata().getDeletionTimestamp()
== null;
+ return driverPod != null
Review Comment:
Addressed in ea8b9997 with
`SparkAppContext.isLiveDriverPod(SparkApplication, Pod)`, which the cached
candidates and the API server pod in `getCurrentAttemptDriverPod()` and the
lookup in `isDriverRequested` share now. It is static so that the mocked
contexts in the tests keep the real check. The new
`podOfAnotherApplicationOnApiServerIsNotCurrentAttemptDriver` covers the API
server side of `getCurrentAttemptDriverPod()`, which checked only the deletion
before.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -1428,6 +1431,173 @@ void neverAdmittedKueueWorkloadDoesNotLookUpTheDriver()
{
verify(recorder, never()).persistStatus(any(), any());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void suspendedAppWithDriverMissingFromCacheKeepsItsKueueWorkload(boolean
queueLabelRemoved) {
+ // The driver was created, but the status update to DriverRequested did
not land, and the app
+ // is suspended before the informer has seen the pod. The suspend hold
would release the
+ // Workload under the driver, even if the queue label was removed
meanwhile.
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+
kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(application)).create();
+ admitWorkload();
+ createDriverPod(application);
+ application.getSpec().setSuspend(true);
+ if (queueLabelRemoved) {
+ application.getMetadata().setLabels(Map.of());
+ }
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
when(mockContext.getCachedKueueWorkload()).thenReturn(Optional.of(getWorkload()));
+
when(mockContext.getCurrentAttemptDriverPod()).thenReturn(Optional.empty());
+ when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec);
+ when(mockContext.getDriverResourcesSpec()).thenReturn(List.of());
+ when(recorder.persistStatus(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ application.setStatus(invocation.getArgument(1));
+ return true;
+ });
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ // The live driver is found, so the initialization completes and the
Workload which holds its
+ // quota is not released
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ Assertions.assertNotNull(getWorkload());
+ Assertions.assertEquals(
+ ApplicationStateSummary.DriverRequested,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ // Like a driver found in the cache, it gets the flavors of the admitted
Workload again
+ verify(mockContext).setKueuePodSetFlavors(any());
+ }
+
+ @Test
+ void suspendedAppWithUnbuildableDriverSpecReleasesKueueWorkload() {
Review Comment:
Addressed in ea8b9997. The new
`suspendedAppWithCachedDriverAndUnbuildableSpecKeepsKueueWorkload` uses the
real `SparkAppContext` with a cached driver pod whose spec build throws, and
asserts that the failure propagates and the `Workload` is kept. It fails before
the change. This test now states that it covers the case without a cached
driver.
--
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]
