dongjoon-hyun commented on code in PR #909:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/909#discussion_r4137431972
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -78,8 +78,7 @@ public ReconcileProgress reconcile(
// Whether the master is live is unknown, not answered. Holding would
claim in an event
// that none was requested, and would release the Kueue quota of a
running master, so
// look again with the steady-state interval instead.
- log.warn("Failed to check whether the master of a suspended cluster
exists.", e);
- return completeAndDefaultRequeue();
+ return SuspendUtils.retryAfterCheckFailure(context, e, "master and
workers");
Review Comment:
**[P1]** With `"master and workers"`, the event message renders as follows,
and the log message starts the same way.
```
Failed to check whether the master and workers of the suspended SparkCluster
was requested, will retry. KubernetesClientException: ...
```
The verb doesn't agree with the subject, and `isMasterRequested` reads only
the master `StatefulSet`, so the workers are never checked. The new docs row,
the Javadoc of `REASON_SUSPEND_CHECK_FAILED` and the sibling message at L249
(`Failed to check whether the master exists ...`) all say `master`, and the log
message which this replaces (`Failed to check whether the master of a suspended
cluster exists.`) was accurate as well.
```suggestion
return SuspendUtils.retryAfterCheckFailure(context, e, "master");
```
`ClusterInitStepTest` L899 needs the same change (`"master of the suspended
SparkCluster"`).
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -88,4 +91,37 @@ static ReconcileProgress holdForSuspend(final BaseContext<?>
context, final Stri
return completeAndRequeueAfter(
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
}
+
+ /**
+ * Reports a failed check of whether the resources of a suspended resource
were requested, and
+ * returns the progress to retry it with. Whether they are live is unknown
then, so the resource
+ * is neither held, which would claim in an event that none was requested
and release the Kueue
+ * quota of a running driver or master, nor started. Like a failed Kueue
admission request, a
+ * transport level failure is not published, since writing an event would
only add load to an API
+ * server that is often the cause of it. Anything else is, since a suspended
resource may have no
+ * persisted status to show it and would otherwise be retried without any
signal.
+ *
+ * @param context The context of the suspended resource.
+ * @param e The failure to report.
+ * @param requested The resources whose request could not be checked, as
named in the event and
+ * the log, e.g. {@code "driver"}.
+ * @return The progress to return, requeued after the steady-state interval.
+ */
+ static ReconcileProgress retryAfterCheckFailure(
+ final BaseContext<?> context, final KubernetesClientException e, final
String requested) {
+ String what =
+ "Failed to check whether the "
+ + requested
+ + " of the suspended "
+ + context.getResource().getKind()
+ + " was requested";
+ log.warn("{}, will retry.", what, e);
+ if (!ReconcilerUtils.isTransientError(e)) {
Review Comment:
**[P3]** Nothing retracts this warning once the check succeeds again. The
recovering `holdForSuspend` publishes the same `SuspendHeld` message as before,
which `ConfigurableEventRecorder` drops if it was published within
`minIntervalSeconds` (300s), and the next one comes only after
`suspendHoldRequeueIntervalSeconds` (1800s). For example, for a suspended
`SparkCluster` on its first attempt, whose master is read from the API server
on every reconciliation:
- t=0s: It is held, and `SuspendHeld` is published.
- t=60s: A spec edit triggers a reconciliation, the read fails with 403, and
`SuspendCheckFailed` is published. It is requeued after 120s.
- t=180s: The read succeeds, and `SuspendHeld` with the same message is
dropped.
- t=1980s: `SuspendHeld` is published again.
So, for about 30 minutes, the newest event says "will retry" although the
check succeeded, while before this PR the newest event stayed the accurate
`SuspendHeld`. The trigger is narrow, but since `holdForSuspend` already
retracts a stale `KueueAdmissionPending` in its message, it may be worth
handling here too, e.g. by saying so in the recovering `SuspendHeld`, or at
least by documenting it.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -78,8 +78,7 @@ public ReconcileProgress reconcile(
// Whether the master is live is unknown, not answered. Holding would
claim in an event
// that none was requested, and would release the Kueue quota of a
running master, so
// look again with the steady-state interval instead.
- log.warn("Failed to check whether the master of a suspended cluster
exists.", e);
- return completeAndDefaultRequeue();
+ return SuspendUtils.retryAfterCheckFailure(context, e, "master and
workers");
}
if (!masterRequested) {
// Unlike a first attempt, a cluster resumed from Suspended has
persisted this Submitted
Review Comment:
**[P4]** For a cluster with a persisted history (`lastKey() > 0`, i.e.
resumed from `Suspended` or requeued after an eviction), both outcomes of the
lookup normally end in `Suspended`: a requested master completes the
initialization and is suspended from `RunningHealthy` by `ClusterSuspendStep`,
and an unrequested one goes to `Suspended` right here. Since
`ClusterSuspendStep` deletes the `StatefulSet`s by name and releases the
`Workload` only after the pods are gone, whether or not the cluster went
through `RunningHealthy`, we could check `lastKey() > 0` before
`isMasterRequested` and skip the lookup for such a cluster. Then the lookup and
`SuspendCheckFailed` remain only for the first attempt, which is exactly the
case without a persisted status that this PR targets. Currently, while the read
keeps failing (e.g. 403), such a cluster republishes `SuspendCheckFailed` every
120s, and its persisted status keeps saying `Cluster is resumed as spec.suspend
is set to false.`, which is wh
at this comment wants to avoid.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -78,8 +78,7 @@ public ReconcileProgress reconcile(
// Whether the master is live is unknown, not answered. Holding would
claim in an event
// that none was requested, and would release the Kueue quota of a
running master, so
// look again with the steady-state interval instead.
- log.warn("Failed to check whether the master of a suspended cluster
exists.", e);
- return completeAndDefaultRequeue();
+ return SuspendUtils.retryAfterCheckFailure(context, e, "master and
workers");
}
if (!masterRequested) {
Review Comment:
**[P2]** When this check finds the master, e.g. since the status update to
`RunningHealthy` was lost, `holdForKueueAdmission` calls `isMasterRequested`
again (L229 and L235). This double lookup predates this PR, but now that the
lookup of a suspended cluster has its own reason, a failure of the second read
is reported differently: as `KueueAdmissionRequestFailed` (`Failed to check
whether the master exists before requesting Kueue admission`), with a 5-second
retry of a transient failure, instead of `SuspendCheckFailed` with the default
interval, as the new docs row says. In the rare case where the second read
returns `null`, a suspended cluster would even request the Kueue admission.
`AppInitStep` avoids this by keeping `driverRequested` outside the suspend
branch and passing it to `holdForKueueAdmission`. Could we do the same here,
i.e. `masterRequested || isMasterRequested(context)` at L229 and L235?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -88,4 +91,37 @@ static ReconcileProgress holdForSuspend(final BaseContext<?>
context, final Stri
return completeAndRequeueAfter(
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
}
+
+ /**
+ * Reports a failed check of whether the resources of a suspended resource
were requested, and
+ * returns the progress to retry it with. Whether they are live is unknown
then, so the resource
+ * is neither held, which would claim in an event that none was requested
and release the Kueue
+ * quota of a running driver or master, nor started. Like a failed Kueue
admission request, a
+ * transport level failure is not published, since writing an event would
only add load to an API
+ * server that is often the cause of it. Anything else is, since a suspended
resource may have no
+ * persisted status to show it and would otherwise be retried without any
signal.
Review Comment:
**[P6]** This rationale applies to `holdForSuspend` above as well, which is
not changed by this PR, so a follow-up JIRA issue is fine.
`KueueWorkloadUtils.releaseWorkload` swallows a failed `Workload` deletion, and
the hold then publishes a normal `SuspendHeld` without the release addendum and
requeues after 30 minutes, without any warning. For example, if the Role of the
operator lacks `delete` on `workloads`, or an admission webhook fails the
deletion, the pending `Workload` of a suspended resource can be admitted into
quota which nothing uses, while only the operator log says so. The dequeued
path reports the same failure as `KueueAdmissionRequestFailed` (see
`failedReleaseOfPendingKueueWorkloadHoldsDequeuedCluster`), and the reason
given in the Javadoc of `releaseWorkload`, "the Workload is garbage collected
with its owner anyway", doesn't hold for a suspended owner.
##########
docs/configuration.md:
##########
@@ -93,6 +93,7 @@ In addition, the operator publishes the following `Warning`
events.
| `KueueAdmissionRequestFailed` | Creating, reading or deleting a stale Kueue
`Workload` fails, or the driver or master cannot be read before the admission
is requested. Transport-level errors are skipped and retried every 5 seconds,
while a persistent failure is retried with the default interval. |
| `ClusterRequestFailed` | Applying the Services, StatefulSets, NetworkPolicy,
HorizontalPodAutoscaler or PodDisruptionBudget of a `SparkCluster` fails in a
way which may yet succeed, such as a throttled request or an internal server
error from an admission webhook, so it is retried with the default interval
instead of failing the cluster. Transport-level errors are skipped. |
| `SuspendReleaseFailed` | Deleting the master and worker StatefulSets, the
HorizontalPodAutoscaler or PodDisruptionBudget of the workers, or the Kueue
`Workload` of a `SparkCluster` suspended while running fails, or its pods
cannot be listed. A failure which may clear on its own, such as a timeout, is
skipped. Every failure is retried with the default interval. |
+| `SuspendCheckFailed` | Whether the driver of a suspended `SparkApplication`,
or the master of a suspended `SparkCluster`, was requested cannot be checked,
so the resource is neither held, which would release its Kueue `Workload`, nor
started. Transport-level errors are skipped. Every failure is retried with the
default interval. |
Review Comment:
**[P5]** This row covers any failure to check, but only a
`KubernetesClientException` reaches `retryAfterCheckFailure`. For a
`SparkApplication`, a driver spec which cannot be built fails the same check
differently. For example, when a suspended application has a live driver pod in
the informer cache and its `sparkConf` was edited so that the spec no longer
builds, `SparkAppContext.getCurrentAttemptDriverPod` (L117) throws a
`SparkException` out of the reconciliation. It then gets `ReconcileError` on
the first attempt only and the JOSDK retries, but neither `SuspendCheckFailed`
nor the default interval, as
`suspendedAppWithCachedDriverAndUnbuildableSpecKeepsKueueWorkload` asserts. How
about narrowing it to a failed read, like the `KueueAdmissionRequestFailed` row?
```suggestion
| `SuspendCheckFailed` | The driver of a suspended `SparkApplication`, or
the master of a suspended `SparkCluster`, cannot be read to check whether it
was requested, so the resource is neither held, which would release its Kueue
`Workload`, nor started. Transport-level errors are skipped. Every such failure
is retried with the default interval. |
```
##########
docs/spark_custom_resources.md:
##########
@@ -544,7 +544,10 @@ published when enabled. Since the status is not there to
fall back on, it is rep
minutes by default while the hold lasts, so that it outlives the event
retention of the API
server. An application held later, in `ScheduledToRestart`, keeps the status
its previous attempt
already wrote and gets the same event, since that status says that a restart
is due, not that the
-next attempt is withheld.
+next attempt is withheld. If the operator cannot check whether the driver (or
master) was requested
Review Comment:
**[P5]** Same as in `configuration.md`, a driver spec which cannot be built
is not reported by `SuspendCheckFailed`.
```suggestion
next attempt is withheld. If the driver (or master) cannot be read to check
whether it was requested
```
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -562,6 +562,38 @@ void suspendedAppWithUnverifiableDriverIsNotHeld() {
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+ @ParameterizedTest
+ @ValueSource(ints = {403, 429, 500})
+ void suspendedAppWithPersistentlyUnverifiableDriverPublishesEvent(int code) {
+ // Unlike a transport level failure, a persistent one is reported, since a
suspended app has no
+ // persisted status to show it. The event must not claim that no driver
was requested.
+ 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);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getCurrentAttemptDriverPod())
+ .thenThrow(new KubernetesClientException("rejected", code, null));
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ verify(mockContext, never()).getDriverPodSpec();
+ verifyNoInteractions(recorder);
+ ArgumentCaptor<EventRecord> event =
ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder).record(event.capture());
+ Assertions.assertEquals(EventType.WARNING, event.getValue().type());
+ Assertions.assertEquals(EventUtils.REASON_SUSPEND_CHECK_FAILED,
event.getValue().reason());
+ Assertions.assertTrue(
+ event.getValue().message().contains("driver of the suspended
SparkApplication"),
+ event.getValue().message());
+ Assertions.assertFalse(
+ event.getValue().message().contains("would not be requested"),
event.getValue().message());
Review Comment:
**[P7]** These assertions pin neither the full message nor its cause. For
instance, both new tests still pass when `+ EventUtils.describe(e)` is dropped
from `retryAfterCheckFailure`, and the `contains` check did not reveal the P1
message. Also, the `assertFalse(...)` can hardly fail, since the reason is
asserted above and the template is fixed. How about comparing the full message,
like the `SuspendHeld` tests in this file?
```java
Assertions.assertEquals(
"Failed to check whether the driver of the suspended
SparkApplication was requested, "
+ "will retry. KubernetesClientException: rejected",
event.getValue().message());
```
The same applies to `ClusterInitStepTest` L898-902.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -562,6 +562,38 @@ void suspendedAppWithUnverifiableDriverIsNotHeld() {
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+ @ParameterizedTest
+ @ValueSource(ints = {403, 429, 500})
+ void suspendedAppWithPersistentlyUnverifiableDriverPublishesEvent(int code) {
+ // Unlike a transport level failure, a persistent one is reported, since a
suspended app has no
+ // persisted status to show it. The event must not claim that no driver
was requested.
+ 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);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getCurrentAttemptDriverPod())
+ .thenThrow(new KubernetesClientException("rejected", code, null));
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ verify(mockContext, never()).getDriverPodSpec();
+ verifyNoInteractions(recorder);
+ ArgumentCaptor<EventRecord> event =
ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder).record(event.capture());
Review Comment:
**[P8]** nit. This file and `ClusterInitStepTest` already have
`captureEvents(int)`, which every other event assertion uses, including
`refusedDriverLookupBeforeKueueAdmissionPublishesEvent`. `EventRecord event =
captureEvents(1).get(0);` is equivalent and avoids the repeated
`event.getValue()`.
--
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]
