dongjoon-hyun commented on code in PR #909:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/909#discussion_r4201068302
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -88,4 +92,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:
`isTransientError` returns `false` for `429`, so a throttled read (API
Priority and Fairness) publishes `SuspendCheckFailed`. The javadoc above skips
transport-level failures because writing an event would only add load to an API
server that is often the cause, which applies to `429` as well. The comment in
`ClusterInitStep` even mentions a single throttled read. Shall we skip `429`
here too?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -88,4 +92,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(
Review Comment:
Once the check succeeds again, nothing supersedes this Warning. Since the
following `SuspendHeld` can be dropped by `minIntervalSeconds`, a stale `Failed
to check ... will retry` can stay the newest event in `kubectl describe` for up
to the suspend hold interval (30 minutes by default), e.g. after an RBAC fix.
The docs document it, but could we avoid it, e.g. by publishing a recovery or
`SuspendHeld` event that is not deduplicated when the previous reconcile failed
the check?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,29 +68,31 @@ 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.
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
- try {
- masterRequested = isMasterRequested(context);
- } catch (KubernetesClientException e) {
- // 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();
- }
- if (!masterRequested) {
- // Unlike a first attempt, a cluster resumed from Suspended has
persisted this Submitted
- // state, which would keep saying that it is resumed. It goes back to
Suspended instead,
- // where ClusterSuspendStep releases whatever it holds only after its
pods are gone.
- if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
- return appendStateAndImmediateRequeue(
- context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ if (cluster.getStatus().getStateTransitionHistory().lastKey() == 0) {
+ final boolean masterRequested;
+ try {
+ masterRequested = isMasterRequested(context);
+ } catch (KubernetesClientException e) {
+ // 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.
+ // Going to Suspended would show a cluster which may never have
started as being
+ // released, and keep it there until it is resumed, even after a
single throttled read.
+ return SuspendUtils.retryAfterCheckFailure(context, e, "master");
+ }
+ if (!masterRequested) {
+ return SuspendUtils.holdForSuspend(context, "master and workers");
}
- return SuspendUtils.holdForSuspend(context, "master and workers");
}
+ // Only a first attempt whose master is known to be absent is held. 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, and a first
attempt whose master
+ // exists was requested before, e.g. the status update to RunningHealthy
failed. Either goes
+ // to Suspended, where ClusterSuspendStep releases the master and
workers whether or not they
+ // were requested, rather than applying every resource again. So the
master of a resumed or
+ // requeued cluster is not looked up.
+ return appendStateAndImmediateRequeue(
+ context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
Review Comment:
A first attempt whose master StatefulSet exists now goes to `Suspended`
directly. Since `isMasterRequested` looks the StatefulSet up by name only, a
StatefulSet left by a deleted `SparkCluster` with the same name, which is still
being garbage-collected, makes a recreated `suspend: true` cluster `Suspended`
(released), although it never ran, and it stays there until it is resumed. This
is the outcome the comment in the catch block wants to avoid. Should we check
the owner reference (UID) as well?
##########
docs/architecture.md:
##########
@@ -203,7 +203,7 @@ stateDiagram-v2
RunningHealthy --> Suspended : spec.suspend=true or Kueue eviction
Suspended --> Submitted : spec.suspend=false or after Kueue eviction
- Submitted --> Suspended : spec.suspend=true after resume
+ Submitted --> Suspended : spec.suspend=true, unless a first attempt has no
master yet
Review Comment:
A first attempt whose master cannot be checked stays in `Submitted` as well,
while this label reads as if it goes to `Suspended`. Maybe `spec.suspend=true,
unless a first attempt has no master yet or cannot check it`?
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java:
##########
@@ -325,17 +325,19 @@ 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
Review Comment:
This test covers a cluster resumed from `Suspended`
(`CLUSTER_RESUMED_MESSAGE`) only. The PR description also mentions a cluster
queued again after a Kueue eviction (`CLUSTER_REQUEUED_MESSAGE`). Could we
parameterize this test with that message, ideally with a Kueue cluster, and
verify that neither the master nor the `Workload` is touched?
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -88,4 +92,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);
Review Comment:
nit: The pattern of `log.warn` followed by `EventUtils.warn` unless
`ReconcilerUtils.isTransientError` now exists in `StatusRecorder`,
`ClusterSuspendStep`, `ClusterInitStep`, and here. A small helper such as
`EventUtils.warnUnlessTransient(recorder, reason, message, e)` would keep the
classification consistent, e.g. if `429` is reclassified. It can be a follow-up.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -68,29 +68,31 @@ 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.
if (cluster.getSpec().isSuspend()) {
- final boolean masterRequested;
- try {
- masterRequested = isMasterRequested(context);
- } catch (KubernetesClientException e) {
- // 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();
- }
- if (!masterRequested) {
- // Unlike a first attempt, a cluster resumed from Suspended has
persisted this Submitted
- // state, which would keep saying that it is resumed. It goes back to
Suspended instead,
- // where ClusterSuspendStep releases whatever it holds only after its
pods are gone.
- if (cluster.getStatus().getStateTransitionHistory().lastKey() > 0) {
- return appendStateAndImmediateRequeue(
- context, statusRecorder, new ClusterState(Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ if (cluster.getStatus().getStateTransitionHistory().lastKey() == 0) {
Review Comment:
The key of the state transition history, used as a proxy for a first
attempt, now also decides whether the master is looked up at all. If a later
change persists another `Submitted` state for a first attempt, e.g. while it
waits for the Kueue admission, this branch flips silently. Could we name this
condition, e.g. with a small helper or by checking the message of the
`Submitted` state (`CLUSTER_RESUMED_MESSAGE` / `CLUSTER_REQUEUED_MESSAGE`), so
that the intent is explicit?
##########
docs/configuration.md:
##########
@@ -112,6 +112,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 until then. The operator log
shows whether the check still fails. |
Review Comment:
After this PR, only a first attempt of a `SparkCluster` looks up its master,
while a resumed or requeued cluster goes to `Suspended` without the lookup and
never gets this event. The same applies to the new paragraph in
`spark_custom_resources.md`. How about saying `the master of a suspended
SparkCluster which has not started yet` or similar?
--
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]
