MaxGekk commented on code in PR #895:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/895#discussion_r4118601506
##########
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:
This catch also covers the case the comment says can't happen:
`getCurrentAttemptDriverPod()` builds the driver spec exactly when the informer
cache holds a live driver-labelled pod (it needs the spec for the pod name). If
the build fails there - the user changed `sparkConf` in the same edit that set
`suspend: true`, a transient `IOException` from the upload filesystem, an NPE -
`driverRequested` stays false and `holdForSuspend` releases the admitted
Workload while the cached driver and its executors keep running outside the
quota. Before this change the exception propagated and the reconcile was
retried, and `ClusterInitStep.isMasterRequested` still lets spec-build failures
propagate.
Could the catch be narrowed to the API-lookup branch of `isDriverRequested`,
after the cache check has come back empty?
##########
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:
For a suspended app whose driver is found here, the lookup runs a second
time in the same reconcile, in `holdForKueueAdmission` (line 208, or through
`handleDequeuedWorkload`). If the second answer differs - the pod was deleted
in between, or the cache and API answers change - it returns false, and the
code falls through to `holdForAdmission`, which with an admitted Workload
returns empty, so the driver is created for an application with
`spec.suspend=true`. In the normal case the second GET is just a wasted call.
Could `driverRequested` be computed once and passed down?
##########
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:
Nit: "carries the driver labels and is not being deleted" is now written out
a third time - `SparkAppContext.getDriverPod` and `getCurrentAttemptDriverPod`
have the same `containsAll(driverLabels(...))` + `deletionTimestamp == null`
filter. If how a pod is recognised as the current attempt's driver changes
later (say, an attempt label) and one copy is missed, the cache path and the
API path disagree about what the driver is, which is the disagreement this PR
removes. Maybe a shared helper such as
`SparkAppContext.isCurrentAttemptDriver(Pod)`?
##########
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:
The API lookup is gated on the *cached* Workload being admitted, but
admission itself is read from the API server (`requestAdmission` ->
`getOrCreateSecondaryResource`). So the same cache-lag race this PR fixes for
pods can come back through the Workload informer: reconcile N reads the
admitted Workload from the API, creates the driver, and the `DriverRequested`
status update fails; the user sets `suspend: true`; at N+1 neither the pod
cache nor the Workload cache has caught up (the Workload may still be the
pre-admission copy, or not cached at all if it was created in N-1).
`getCachedKueueWorkload()` is then empty or not admitted, this returns false
without a lookup, and `releaseWorkload` deletes the admitted Workload by name
through the API while the driver runs.
For a suspended app, would it be safer to look the driver up whenever a
Workload may exist (queue label present or Workload cached), as
`ClusterInitStep.isMasterRequested` does?
##########
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:
Building the full driver spec here has side effects for a suspended app: the
submission worker uploads dependencies to `spark.kubernetes.file.upload.path`
and cleans up temp resources, only to decide whether to hold it. And when a
driver is found, `applyAdmittedFlavors` -> `setKueuePodSetFlavors` resets the
memoized spec, so it's built (and uploaded) again in the same reconcile. The
lookup only needs the driver pod's name, which is deterministic
(`SparkAppSubmissionWorker.generateSparkAppId`); a label-selected GET like
`SparkAppContext.getDriverPodFromApi()`, or the name derived without building
the whole spec, would avoid both the side effects and the spec-build failure
mode behind the catch-all at line 87.
##########
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:
This test stubs `getCurrentAttemptDriverPod()` to return empty, so the real
path where a driver-labelled pod is cached and `getDriverPodSpec()` throws
inside `getCurrentAttemptDriverPod()` isn't exercised (see the comment at
`AppInitStep.java:87`). The `assertNull(getWorkload())` also makes "release on
spec failure" the intended behaviour for every path. Could there be a test with
a cached driver pod whose spec build throws, asserting the Workload is kept?
--
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]
