dongjoon-hyun commented on code in PR #909:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/909#discussion_r4146679904
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,27 +68,30 @@ public ReconcileProgress reconcile(
return proceed();
}
SparkCluster cluster = context.getResource();
- // A cluster whose master StatefulSet already exists has been requested
before (e.g. the status
- // update to RunningHealthy failed), so let it complete its initialization
even if suspended.
+
+ boolean masterRequested = false;
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
+ // Unlike a first attempt, a cluster resumed from Suspended or queued
again after an eviction
+ // has persisted this Submitted state, which would keep saying that it
is resumed. It goes
+ // back to Suspended instead, where ClusterSuspendStep deletes the
master and worker
+ // StatefulSets by name and releases whatever it holds only after its
pods are gone. That
+ // happens whether or not the master was requested, so it is not looked
up.
+ if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
+ return appendStateAndImmediateRequeue(
+ context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ }
+ // A cluster whose master StatefulSet already exists has been requested
before (e.g. the
+ // status update to RunningHealthy failed), so let it complete its
initialization even if
+ // suspended.
Review Comment:
**[P1]** The new comment above says that `ClusterSuspendStep` deletes the
master and worker `StatefulSet`s by name and releases the `Workload` only after
the pods are gone, whether or not the master was requested. That holds for a
first attempt whose master exists as well, yet such a cluster still completes
its initialization here. It applies every resource again and persists
`RunningHealthy`, and only then does `ClusterSuspendStep` move it to
`Suspended` and delete what was just applied. For example, suppose the previous
reconciliation applied the master `StatefulSet` but failed to apply the worker
one with a retryable error, or the status update to `RunningHealthy` failed,
and `spec.suspend` is set to `true` afterwards.
- The worker `StatefulSet`, which never existed, is created, and its pods
start only to be deleted in the next reconciliation.
- If an admission webhook keeps failing the `NetworkPolicy`, HPA or PDB with
500, only `ClusterRequestFailed` repeats. The suspension is never applied,
while the master and its Kueue `Workload` keep the quota.
- A non-retryable failure, e.g. 422 for a spec edited while suspended, or a
node selector conflict of the Kueue flavors ends the cluster in
`SchedulingFailure`, although the user only suspended it.
Could we move such a cluster to `Suspended` as well, like the `lastKey() >
0` branch? Then `holdForKueueAdmission` is reached only when `spec.suspend` is
`false`, so the new `masterRequested` parameter and both `masterRequested ||`
are no longer needed. `suspendAfterMasterRequestedCompletesInitialization` and
the new `suspendedQueuedClusterWithRequestedMasterIsLookedUpOnce`, which assert
`RunningHealthy` for this case, would change accordingly. Since this path
predates this PR, a follow-up JIRA issue is also fine.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -221,19 +225,22 @@ private ReconcileProgress failScheduling(
*
* @param context The SparkClusterContext for the cluster.
* @param cluster The SparkCluster.
+ * @param masterRequested Whether the master was found requested already in
this reconcile, e.g.
+ * by the suspend check, so that it is not looked up again.
* @return The progress to return while the admission is not granted, or
empty to proceed.
*/
private Optional<ReconcileProgress> holdForKueueAdmission(
- SparkClusterContext context, SparkCluster cluster) {
+ SparkClusterContext context, SparkCluster cluster, boolean
masterRequested) {
try {
if (!KueueWorkloadFactory.hasQueueName(cluster)) {
- return KueueWorkloadUtils.handleDequeuedWorkload(context, () ->
isMasterRequested(context));
+ return KueueWorkloadUtils.handleDequeuedWorkload(
+ context, () -> masterRequested || isMasterRequested(context));
Review Comment:
**[P2]** No test covers this `masterRequested ||`. With this line reverted
to `() -> isMasterRequested(context)`, all 555 tests of `:spark-operator:test`
still pass.
- `suspendedQueuedClusterWithRequestedMasterIsLookedUpOnce` uses a cluster
with the queue label, so it reaches only L243.
- `kueueFlavorsAreAppliedAgainWhenTheMasterExists(queueLabelRemoved=true)`
is not suspended, so `masterRequested` is `false` there.
- `suspendAfterMasterRequestedCompletesInitialization` doesn't stub
`getCachedKueueWorkload()`, so `handleDequeuedWorkload` returns before it
evaluates the supplier.
The uncovered case is a suspended first attempt whose queue label was
removed after its admission. Without this line, a 403 of the second read is
reported as `KueueAdmissionRequestFailed` (`... before requesting Kueue
admission`), which is what the comment of the new test wants to avoid. A `null`
applies the `StatefulSet`s again without the flavors of the admitted
`Workload`. P9 suggests a way to cover both L237 and L243.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -405,6 +408,61 @@ void suspendAfterMasterRequestedCompletesInitialization() {
verify(statefulSetApplicable, times(2)).serverSideApply();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void suspendedQueuedClusterWithRequestedMasterIsLookedUpOnce() {
+ // The suspend check already found the master, so the Kueue admission must
not read it again:
+ // a failure of that second read would be reported as
KueueAdmissionRequestFailed rather than
+ // SuspendCheckFailed, and a null would request the admission of a running
master
+ ClusterInitStep clusterInitStep = new ClusterInitStep();
+ SparkClusterContext mockContext = mock(SparkClusterContext.class);
+ SparkClusterStatusRecorder recorder =
mock(SparkClusterStatusRecorder.class);
+ SparkCluster cluster = buildKueueCluster();
+ cluster.getSpec().setSuspend(true);
+ KubernetesClient mockClient = mock(KubernetesClient.class,
RETURNS_DEEP_STUBS);
+ when(mockClient.resource(masterStatefulSetSpec).get())
+ .thenReturn(masterStatefulSetSpec)
+ .thenThrow(new KubernetesClientException("forbidden", 403, null));
+ stubWorkloadRead(mockClient, admittedWorkload(cluster, Map.of()));
+ ServerSideApplicable<Service> serviceApplicable =
mock(ServerSideApplicable.class);
+ ServiceResource<Service> serviceResource = mock(ServiceResource.class);
+ when(serviceResource.forceConflicts()).thenReturn(serviceApplicable);
+
when(mockClient.services().resource(any(Service.class))).thenReturn(serviceResource);
+ ServerSideApplicable<StatefulSet> statefulSetApplicable =
mock(ServerSideApplicable.class);
+ RollableScalableResource<StatefulSet> statefulSetResource =
+ mock(RollableScalableResource.class);
+
when(statefulSetResource.forceConflicts()).thenReturn(statefulSetApplicable);
+ when(mockClient.apps().statefulSets().resource(any(StatefulSet.class)))
+ .thenReturn(statefulSetResource);
+ ServerSideApplicable<NetworkPolicy> networkPolicyApplicable =
mock(ServerSideApplicable.class);
+ Resource<NetworkPolicy> networkPolicyResource = mock(Resource.class);
+
when(networkPolicyResource.forceConflicts()).thenReturn(networkPolicyApplicable);
+
when(mockClient.network().networkPolicies().resource(any(NetworkPolicy.class)))
+ .thenReturn(networkPolicyResource);
+ when(mockContext.getResource()).thenReturn(cluster);
+ when(mockContext.getClient()).thenReturn(mockClient);
+
when(mockContext.getMasterServiceSpec()).thenReturn(service("cluster1-master-svc"));
+
when(mockContext.getWorkerServiceSpec()).thenReturn(service("cluster1-worker-svc"));
+
when(mockContext.getMasterStatefulSetSpec()).thenReturn(masterStatefulSetSpec);
+
when(mockContext.getWorkerStatefulSetSpec()).thenReturn(workerStatefulSetSpec);
+
when(mockContext.getWorkerNetworkPolicySpec()).thenReturn(networkPolicy("cluster1-worker"));
+
when(mockContext.getHorizontalPodAutoscalerSpec()).thenReturn(Optional.empty());
+
when(mockContext.getPodDisruptionBudgetSpec()).thenReturn(Optional.empty());
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
+ ReconcileProgress progress = clusterInitStep.reconcile(mockContext,
recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ verify(mockClient.resource(masterStatefulSetSpec)).get();
+ verify(mockClient, never()).resource(any(Workload.class));
Review Comment:
**[P3]** Two of these assertions cannot fail for the regression which this
test targets. With L243 of `ClusterInitStep` reverted to `if
(isMasterRequested(context))`, the second `get()` throws the stubbed 403, which
`holdForKueueAdmission` turns into `retryAfterRequestFailure`. It returns
`completeAndDefaultRequeue()` as well, and nothing reaches
`client.resource(workload)`. So only L456, L458 and L460 catch it, and the test
passes with L455 and L457 alone.
The `null` case in the comment (`a null would request the admission of a
running master`) is not exercised either. Even with `.thenReturn(null)` as the
second answer, the test fails earlier, since the unstubbed
`priorityClasses().list()` in `KueueWorkloadPriority` throws a
`NullPointerException`, which ends in `SchedulingFailure`. How about saying in
the comment that the `get()` count at L456 guards this, and either dropping
L457 or covering the `null` case with `priorityClasses().list()` stubbed as
well?
##########
docs/spark_custom_resources.md:
##########
@@ -544,7 +544,12 @@ 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 driver (or master) cannot be read to check
whether it was requested
+already, e.g. since the API server rejects the read, the resource is neither
held nor started, and
+the `SuspendCheckFailed` event is published instead, until the check succeeds.
A failure which may
+clear on its own, such as a timeout, is retried without the event. Once the
check succeeds again,
+the `SuspendHeld` event may follow only with its next repeat, so a
`SuspendCheckFailed` event which
+is no longer refreshed no longer applies.
Review Comment:
**[P4]** This doesn't hold once the check fails only at the transport level,
and neither does `One which is no longer refreshed has cleared.` in
`configuration.md`. For example, the master read fails with 500, and
`SuspendCheckFailed` is published. If the same outage then continues with 503,
504 or connection resets, `retryAfterCheckFailure` skips the event but still
neither holds nor starts the cluster, so its pending `Workload` is not released
either. A user who follows this sentence would conclude that the failure is
over.
```suggestion
the `SuspendHeld` event may follow only with its next repeat, so a
`SuspendCheckFailed` event which
is no longer refreshed no longer applies, unless the check now fails only at
the transport level.
```
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,27 +68,30 @@ public ReconcileProgress reconcile(
return proceed();
}
SparkCluster cluster = context.getResource();
- // A cluster whose master StatefulSet already exists has been requested
before (e.g. the status
- // update to RunningHealthy failed), so let it complete its initialization
even if suspended.
+
+ boolean masterRequested = false;
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
+ // Unlike a first attempt, a cluster resumed from Suspended or queued
again after an eviction
+ // has persisted this Submitted state, which would keep saying that it
is resumed. It goes
+ // back to Suspended instead, where ClusterSuspendStep deletes the
master and worker
+ // StatefulSets by name and releases whatever it holds only after its
pods are gone. That
+ // happens whether or not the master was requested, so it is not looked
up.
+ if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
+ return appendStateAndImmediateRequeue(
+ context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ }
Review Comment:
**[P8]** This changes a behavior which the PR description calls unchanged
(`The requeue behavior is unchanged: ...`). A suspended `SparkCluster` resumed
from `Suspended` or queued again after an eviction now goes back to `Suspended`
without the lookup. Before, a failed lookup was retried in `Submitted` with the
default interval, and an existing master completed the initialization to
`RunningHealthy` first. Could you update the PR description, including the
user-facing change section? It becomes the commit message when merged.
##########
docs/spark_custom_resources.md:
##########
@@ -544,7 +544,12 @@ 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 driver (or master) cannot be read to check
whether it was requested
+already, e.g. since the API server rejects the read, the resource is neither
held nor started, and
+the `SuspendCheckFailed` event is published instead, until the check succeeds.
A failure which may
Review Comment:
**[P6]** Events are disabled by default
(`spark.kubernetes.operator.events.enabled=false`), and this document says
`when enabled` for the other events, e.g. `SuspendHeld` at L542-543. With the
defaults, a suspended first attempt whose check fails has neither a persisted
status nor this event, only the operator log.
```suggestion
the `SuspendCheckFailed` event is published instead when enabled, until the
check succeeds. A failure which may
```
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,27 +68,30 @@ public ReconcileProgress reconcile(
return proceed();
}
SparkCluster cluster = context.getResource();
- // A cluster whose master StatefulSet already exists has been requested
before (e.g. the status
- // update to RunningHealthy failed), so let it complete its initialization
even if suspended.
+
+ boolean masterRequested = false;
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
+ // Unlike a first attempt, a cluster resumed from Suspended or queued
again after an eviction
+ // has persisted this Submitted state, which would keep saying that it
is resumed. It goes
+ // back to Suspended instead, where ClusterSuspendStep deletes the
master and worker
+ // StatefulSets by name and releases whatever it holds only after its
pods are gone. That
+ // happens whether or not the master was requested, so it is not looked
up.
Review Comment:
**[P7]** nit. A cluster queued again after an eviction persisted `Submitted`
with `CLUSTER_REQUEUED_MESSAGE` (`Cluster is submitted again after its eviction
by Kueue.`), not with the resumed one, see `String message = evicted ?
CLUSTER_REQUEUED_MESSAGE : CLUSTER_RESUMED_MESSAGE;` in `ClusterSuspendStep`.
```suggestion
// Unlike a first attempt, a cluster resumed from Suspended or queued
again after an eviction
// has persisted this Submitted state, which would keep saying that it
is resumed or queued
// again. It goes back to Suspended instead, where ClusterSuspendStep
deletes the master and
// worker StatefulSets by name and releases whatever it holds only
after its pods are gone.
// That happens whether or not the master was requested, so it is not
looked up.
```
##########
docs/spark_custom_resources.md:
##########
@@ -544,7 +544,12 @@ 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 driver (or master) cannot be read to check
whether it was requested
+already, e.g. since the API server rejects the read, the resource is neither
held nor started, and
Review Comment:
**[P12]** The Kueue section still says that `suspending a queued resource
deletes its Workload to release the quota` (L691), and the guide for disabling
the integration asks to suspend the queued resources `so that the operator
releases their Workloads itself` (L720-722). With this paragraph, that no
longer holds while the check fails. For example, if the read keeps failing with
403, the `Workload` stays until the read succeeds, and a pending one may even
be admitted meanwhile. The behavior predates this PR, but now that it is
documented here, how about qualifying those two places too, e.g. by linking to
this section?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -321,17 +321,20 @@ void resumedClusterSuspendedAgainGoesBackToSuspended() {
new ClusterState(ClusterStateSummary.Submitted,
Constants.CLUSTER_RESUMED_MESSAGE),
true));
cluster.getSpec().setSuspend(true);
+ // Not stubbed: a read would find a master, but the cluster must not look
it up at all
KubernetesClient mockClient = mock(KubernetesClient.class,
RETURNS_DEEP_STUBS);
- when(mockClient.resource(masterStatefulSetSpec).get()).thenReturn(null);
when(mockContext.getResource()).thenReturn(cluster);
when(mockContext.getClient()).thenReturn(mockClient);
when(mockContext.getMasterStatefulSetSpec()).thenReturn(masterStatefulSetSpec);
Review Comment:
**[P10]** nit. This stub is no longer used, since the cluster now goes to
`Suspended` before the lookup, and it contradicts `verify(mockContext,
never()).getMasterStatefulSetSpec()` at L336. It passes only because these
tests don't use strict stubs.
```suggestion
```
##########
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)) {
+ EventUtils.warn(
+ context.getEventRecorder(),
+ EventUtils.REASON_SUSPEND_CHECK_FAILED,
+ what + ", will retry. " + EventUtils.describe(e));
+ }
+ return completeAndDefaultRequeue();
Review Comment:
**[P14]** nit. Except for the reason and the interval of a transient
failure, this is the same as `KueueWorkloadUtils.retryAfterRequestFailure`: the
same log line, the same `what + ", will retry. " + EventUtils.describe(e)` and
the same gate. On top of the current `main`, the `log.warn` plus `if
(!ReconcilerUtils.isTransientError(e)) EventUtils.warn(...)` pattern appears in
seven places with this PR, e.g. `ClusterSuspendStep`, `ClusterInitStep`
(`ClusterRequestFailed`), `StatusRecorder` and
`KueueWorkloadUtils.recordPodsReady`. A helper like
`EventUtils.warnUnlessTransient(recorder, reason, message, e)` would keep the
policy in one place. This can be a follow-up.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,27 +68,30 @@ public ReconcileProgress reconcile(
return proceed();
}
SparkCluster cluster = context.getResource();
- // A cluster whose master StatefulSet already exists has been requested
before (e.g. the status
- // update to RunningHealthy failed), so let it complete its initialization
even if suspended.
+
+ boolean masterRequested = false;
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
+ // Unlike a first attempt, a cluster resumed from Suspended or queued
again after an eviction
+ // has persisted this Submitted state, which would keep saying that it
is resumed. It goes
+ // back to Suspended instead, where ClusterSuspendStep deletes the
master and worker
+ // StatefulSets by name and releases whatever it holds only after its
pods are gone. That
+ // happens whether or not the master was requested, so it is not looked
up.
+ if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
+ return appendStateAndImmediateRequeue(
+ context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ }
+ // A cluster whose master StatefulSet already exists has been requested
before (e.g. the
+ // status update to RunningHealthy failed), so let it complete its
initialization even if
+ // suspended.
try {
masterRequested = isMasterRequested(context);
} catch (KubernetesClientException e) {
Review Comment:
**[P5]** This catches only a `KubernetesClientException`, but
`isMasterRequested` builds the whole master `StatefulSet` spec only for its
name. If the spec cannot be built, e.g. after it was edited while suspended,
the exception leaves the reconciliation, since the `try` at L106 is not reached
yet. Then a suspended first attempt gets neither `SuspendHeld` nor
`SuspendCheckFailed`, only a `ReconcileError` while the JOSDK retries it, and
`holdForSuspend` never releases its pending `Workload`.
`AppInitStep` handles the same case with `hasLiveDriverPod`, and
`ClusterSuspendStep` deletes the `StatefulSet`s by name, since the spec `may
not even build after being edited while suspended`. How about reading the
master by name in `isMasterRequested` as well, i.e.
`apps().statefulSets().inNamespace(namespace).withName(getMasterStatefulSetName(name)).get()`?
This is not introduced by this PR, and I couldn't find a spec which passes the
validation and still fails to build, so a follow-up JIRA issue is fine.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -321,17 +321,20 @@ void resumedClusterSuspendedAgainGoesBackToSuspended() {
new ClusterState(ClusterStateSummary.Submitted,
Constants.CLUSTER_RESUMED_MESSAGE),
Review Comment:
**[P11]** The new branches are tested only for this resumed history, a 503,
and an application in `Submitted`.
- A cluster queued again after an eviction (`CLUSTER_REQUEUED_MESSAGE`),
which the new comment names, is not tested. The branch doesn't read the message
today, but `ClusterSuspendStep` tells the two apart, so a parameter over both
messages would guard a later change.
- A transport-level failure, e.g. `new KubernetesClientException("closed",
new SocketException("reset"))`, is not tested for the suspend check of either
step. Only `ReconcilerUtilsTest` checks its classification.
- `SuspendCheckFailed` of an application in `ScheduledToRestart`, which the
new docs sentence follows, is not tested either.
##########
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` | 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. While the failure lasts, it is
republished on every reconciliation, subject to
[`minIntervalSeconds`](#event-frequency). Once the check succeeds again, the
unchanged `SuspendHeld` event may be dropped by the same interval until its
next repeat, so this event can stay the newest one for up to 30 minutes
(`spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds`). One
which is no longer refreshed has cleared. |
Review Comment:
**[P13]** `up to 30 minutes` holds only while `minIntervalSeconds` is at
most `suspendHoldRequeueIntervalSeconds`. With `minIntervalSeconds=2400`, which
the Event frequency section uses as an example, the recovering `SuspendHeld` is
dropped at its next repeat as well, so this event stays the newest one for
about 60 minutes. The interval is also in the `SuspendHeld` row already. How
about `until its next repeat` only? The suggestion also rewords the last
sentence for P4.
```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. While the failure lasts, it is
republished on every reconciliation, subject to
[`minIntervalSeconds`](#event-frequency). Once the check succeeds again, the
unchanged `SuspendHeld` event may be dropped by the same interval until its
next repeat, so this event can stay the newest one until then. One which is no
longer refreshed has cleared, unless the check now fails only at the transport
level. |
```
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -405,6 +408,61 @@ void suspendAfterMasterRequestedCompletesInitialization() {
verify(statefulSetApplicable, times(2)).serverSideApply();
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void suspendedQueuedClusterWithRequestedMasterIsLookedUpOnce() {
Review Comment:
**[P9]** This test copies the block of the apply and spec stubs, which this
file already has six times. How about a `suspended` parameter of
`kueueFlavorsAreAppliedAgainWhenTheMasterExists` instead? For example:
```java
@ParameterizedTest
@CsvSource({"false, false", "true, false", "false, true", "true, true"})
@SuppressWarnings("unchecked")
void kueueFlavorsAreAppliedAgainWhenTheMasterExists(
boolean queueLabelRemoved, boolean suspended) {
...
SparkCluster cluster = buildKueueCluster();
cluster.getSpec().setSuspend(suspended);
...
verify(mockClient.resource(masterStatefulSetSpec)).get();
```
I tried it locally. It passes with this PR, while it fails only for `(true,
true)` with L237 reverted and only for `(false, true)` with L243 reverted, so
it covers P2 as well.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -562,6 +562,36 @@ void suspendedAppWithUnverifiableDriverIsNotHeld() {
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+ @ParameterizedTest
+ @ValueSource(ints = {403, 429, 500})
+ void suspendedAppWithPersistentlyUnverifiableDriverPublishesEvent(int code) {
Review Comment:
**[P15]** nit. This test and
`suspendedClusterWithPersistentlyUnverifiableMasterPublishesEvent` differ from
the 503 tests right above them only in the code and the event assertions. They
could be one test each, like
`retryableFailureOfRequestingResourcesIsRetried(int code, int events)` in
`ClusterInitStepTest`, e.g. with `@CsvSource({"503, 0", "403, 1", "429, 1",
"500, 1"})`. Then the `Submitted` assertion of the 503 tests would cover the
other codes as well. The current split is also fine, since
`ClusterSuspendStepTest` does the same.
--
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]
