peter-toth commented on code in PR #854:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/854#discussion_r4060593561
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -70,12 +70,7 @@ public ReconcileProgress reconcile(
}
SparkApplication app = context.getResource();
if (app.getSpec().isSuspend() && !isDriverRequested(context)) {
- log.debug("Application is suspended, driver resources would not be
requested.");
- if (KueueWorkloadFactory.hasQueueName(app)) {
- // A resource suspended while queued must not keep holding the Kueue
quota.
- KueueWorkloadUtils.releaseWorkload(context.getClient(), app);
- }
- return completeAndDefaultRequeue();
+ return SuspendUtils.holdForSuspend(context, "driver");
Review Comment:
**Finding 1.** Both guards in front of this hold report "already requested"
as `false` when the read that answers it failed, so a resource whose driver or
master is live can be held. This PR stretches that hold from the default
interval to 30 minutes.
`SparkAppContext.getCurrentAttemptDriverPod()` catches
`KubernetesClientException` and returns empty, logging "considering it absent"
(`spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java:135`).
`ClusterInitStep.isMasterRequested` goes through the lenient
`ReconcilerUtils.getResource`, which does the same. That is the point I raised
on #853 at
[r4057681801](https://github.com/apache/spark-kubernetes-operator/pull/853#discussion_r4057681801),
deferred there as a follow-up.
Observed on this head. A suspended app whose current-attempt driver pod
exists in the API server, with the live verification read answering 503:
```
PROBE-PROGRESS >>> ReconcileProgress(completed=true, requeue=true,
requeueAfterDuration=PT30M)
PROBE-EVENT >>> The SparkApplication is suspended by spec.suspend, driver
would not be requested. Set spec.suspend to false to resume it.
PROBE-POD-STILL-THERE >>> true
```
Two consequences, both of them new here. The event asserts the driver was
not requested while it is running, and the republishing keeps it in `kubectl
describe` for half an hour. And the next look moves from `PT2M` on base to
`PT30M`, so an app stuck in `Submitted` with a live driver waits 15x longer
unless a pod update happens to arrive first. That window is exactly the one
`isDriverRequested` exists for: the app is in `Submitted` with a live driver
because an API write just failed, so the read failing on the same reconcile is
not an independent coincidence.
The shape I would use is to let the two checks report "could not tell" apart
from "absent", and on that answer return `completeAndDefaultRequeue()` and
publish no event. The app side is self-contained, `getCurrentAttemptDriverPod`
can surface the distinction instead of swallowing it. The cluster side needs
`getResourceStrictly` to be reachable, which is the #853 follow-up, so it can
wait for that.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -94,6 +94,9 @@ class AppInitStepTest {
private final ResourceEventRecorder eventRecorder =
mock(ResourceEventRecorder.class);
+ private static final ReconcileProgress SUSPEND_HOLD_PROGRESS =
+
ReconcileProgress.completeAndRequeueAfter(SuspendUtils.SUSPEND_HOLD_REQUEUE_INTERVAL);
Review Comment:
**Finding 4.** This derives the expected progress from the constant under
test, so the value itself is not pinned. I set `SUSPEND_HOLD_REQUEUE_INTERVAL =
Duration.ofSeconds(120)`, the default interval this deliberately departs from,
and the whole `:spark-operator` suite still passed.
Three places state 30 minutes to users: the `SuspendHeld` row in
`docs/configuration.md:85`, `docs/spark_custom_resources.md:541`, and the
field's own javadoc. All three drift silently today. One literal is enough,
here or in `ClusterInitStepTest.java:87`:
```java
private static final ReconcileProgress SUSPEND_HOLD_PROGRESS =
ReconcileProgress.completeAndRequeueAfter(Duration.ofMinutes(30));
```
`java.time.Duration` is not imported in either test class yet.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
+
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndRequeueAfter;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.spark.k8s.operator.context.BaseContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+
+/** Utilities to hold the resources of a suspended SparkApplication or
SparkCluster. */
+@Slf4j
+final class SuspendUtils {
+
+ /**
+ * Interval at which a suspended resource is reconciled, so that its event
is republished well
+ * within the one hour that the API server retains events by default. It is
deliberately much
+ * coarser than the steady-state reconcile interval: a suspended resource
has nothing to observe,
+ * while each republish costs a read and a write on the API server for as
long as the hold lasts.
+ * Nothing waits for this interval, since clearing spec.suspend arrives as a
watch event that
+ * reconciles the resource right away.
+ */
+ static final Duration SUSPEND_HOLD_REQUEUE_INTERVAL = Duration.ofMinutes(30);
Review Comment:
**Finding 3.** The one hour this is sized against is the kube-apiserver
`--event-ttl` flag, not a fixed property of Kubernetes. An operator on a
cluster that sets it below 30 minutes cannot shorten the republishing, so the
event disappears between repeats and the hold stops being visible.
`STALE_WORKLOAD_REQUEUE_INTERVAL` is a constant too, but nothing on a cluster
can make 5 seconds wrong, while the retention this one races is configurable.
It also costs the E2E its main assertion. Republishing is what the design
rests on, and with a 30-minute floor the `suspend-events` group cannot reach
`count > 1`, so nothing outside the unit tests exercises it.
A `ConfigOption<Long>` next to `MISSING_DRIVER_REQUEUE_INTERVAL_SECONDS` in
`SparkOperatorConf`, defaulting to 1800 and with `enableDynamicOverride(true)`,
addresses both: `tests/e2e/helm/events-config-values.yaml` can set it to a few
seconds and the suite can assert the count goes above one.
##########
docs/configuration.md:
##########
@@ -76,17 +76,20 @@ In addition, the operator publishes the following `Warning`
events.
| `StatusUpdateFailed` | A status patch is rejected. Transport-level errors
are skipped. |
| `KueueAdmissionRequestFailed` | Creating, reading or deleting a stale Kueue
`Workload` fails. Transport-level errors are skipped and retried every 5
seconds, while a persistent failure is retried with the default interval. |
-For a resource queued by [Kueue](spark_custom_resources.md#kueue), the
operator also publishes the
-following `Normal` events, since the resource stays in its initializing state
while it waits.
+For a resource held by [`spec.suspend`](spark_custom_resources.md#suspend) or
queued by
+[Kueue](spark_custom_resources.md#kueue), the operator also publishes the
following `Normal`
+events, since the resource stays in its initializing state without a state
transition.
| Reason | When |
|---|---|
+| `SuspendHeld` | The resource is held by `spec.suspend`, so the driver (or
master and worker) is not requested. It is republished every 30 minutes while
the hold lasts, so a repeat bumps the `count` of the one event rather than
creating another. Suspending a queued resource releases its Kueue `Workload`,
and the message says so, since the `KueueAdmissionPending` event it was queued
with outlives that `Workload`. |
| `KueueAdmissionPending` | The Kueue `Workload` waits for the admission. It
is republished while it waits, so a repeat bumps the `count` of the one event
rather than creating another. |
| `KueueAdmitted` | Kueue admitted the `Workload`, so the driver (or master
and worker) is requested. |
The `reason` values are stable, while the `message` values may change between
releases. Note that
Kubernetes retains events only for a limited time (one hour by default), so
the resource status
-remains the source of truth.
+remains the source of truth. The exception is a first attempt held by
`spec.suspend` or by Kueue,
Review Comment:
**Finding 6.** This narrows the exception to a first attempt, but an
application held in `ScheduledToRestart` is neither a first attempt nor without
a persisted status, and it gets the republished event too.
`docs/spark_custom_resources.md:542` and `SuspendUtils`'s javadoc both say so,
and `suspendedAppScheduledToRestartDoesNotRequestDriver` asserts the event.
The condition that actually holds is that the status does not report the
hold, which is also why the `ScheduledToRestart` case needs the event.
Something like "a resource held by `spec.suspend` or by Kueue, whose status
does not report the hold" covers both.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
+
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndRequeueAfter;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.spark.k8s.operator.context.BaseContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+
+/** Utilities to hold the resources of a suspended SparkApplication or
SparkCluster. */
+@Slf4j
+final class SuspendUtils {
+
+ /**
+ * Interval at which a suspended resource is reconciled, so that its event
is republished well
+ * within the one hour that the API server retains events by default. It is
deliberately much
+ * coarser than the steady-state reconcile interval: a suspended resource
has nothing to observe,
+ * while each republish costs a read and a write on the API server for as
long as the hold lasts.
+ * Nothing waits for this interval, since clearing spec.suspend arrives as a
watch event that
+ * reconciles the resource right away.
+ */
+ static final Duration SUSPEND_HOLD_REQUEUE_INTERVAL = Duration.ofMinutes(30);
+
+ private SuspendUtils() {}
+
+ /**
+ * Holds the resources of a resource suspended by {@code spec.suspend} and
reports the progress
+ * to return. A resource suspended while queued releases its Kueue Workload
first, so that it
+ * does not keep holding the quota, and its event then says so, since the
pending event it was
+ * queued with outlives the Workload. Callers keep their own guard for
resources requested
+ * before, which must complete their initialization instead of being held.
+ *
+ * <p>Like the Kueue pending event, the event is republished while the hold
lasts rather than
+ * once when it starts. The event sink keys the Event on the reason, so a
repeat bumps the count
+ * of the one Event instead of creating another and refreshes it, so that
the hold stays visible
+ * past the event retention of the API server. A suspended first attempt has
no persisted status
+ * to fall back on, since its initial Submitted state is never written. An
application held
+ * later, in ScheduledToRestart, does have the status of the previous
attempt, but that status
+ * does not say that the next attempt is withheld by spec.suspend, so it
gets the same event.
+ *
+ * <p>Unlike the Kueue hold, which ends when quota arrives, this one ends
only when a user clears
+ * spec.suspend, so the republishing is paced by {@link
#SUSPEND_HOLD_REQUEUE_INTERVAL} rather
+ * than by the steady-state reconcile interval.
+ *
+ * @param context The context of the suspended resource.
+ * @param requested The resources held until the resource is resumed, as
named in the event and
+ * the log, e.g. {@code "driver"}.
+ * @return The progress to return while the resource is suspended, requeued
after {@link
+ * #SUSPEND_HOLD_REQUEUE_INTERVAL}.
+ */
+ static ReconcileProgress holdForSuspend(final BaseContext<?> context, final
String requested) {
+ HasMetadata resource = context.getResource();
+ log.debug("{} is suspended, {} would not be requested.",
resource.getKind(), requested);
+ String message =
+ "The "
+ + resource.getKind()
+ + " is suspended by spec.suspend, "
+ + requested
+ + " would not be requested. Set spec.suspend to false to resume
it.";
+ if (KueueWorkloadFactory.hasQueueName(resource)) {
+ KueueWorkloadUtils.releaseWorkload(context.getClient(), resource);
+ // The pending event of a resource that was queued before stays until
the API server drops
+ // it, and the operator may not delete events, so say that it no longer
applies rather than
+ // leaving a contradicting pair behind.
+ message +=
Review Comment:
**Finding 2.** The suffix is appended on the queue-name label alone, so a
resource created with `suspend: true` and a queue name gets it even though it
was never queued and no `KueueAdmissionPending` event was ever published.
`suspendedAppWithQueueNameDoesNotCreateKueueWorkload`
(`spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:782`)
is exactly that case, and it asserts only the reason, so the message goes
unchecked. What it produces:
```
The SparkApplication is suspended by spec.suspend, driver would not be
requested. Set spec.suspend to false to resume it. It holds no Kueue Workload
while suspended, so an earlier KueueAdmissionPending event no longer applies.
```
An application reaching `ScheduledToRestart` is the same case, since
`AppCleanUpStep` already released the Workload on the way in.
`delete()` tells the two apart, so the suffix can be conditional. Measured
against the mock server with the same Workload name: `[]` when it is absent,
`[StatusDetails(... kind=Workload, name=sparkapplication-sparkapp1 ...)]` when
it is present. So `releaseWorkload` can report it:
```java
if (KueueWorkloadFactory.hasQueueName(resource)
&& KueueWorkloadUtils.releaseWorkload(context.getClient(), resource)) {
message += " It holds no Kueue Workload while suspended, so an earlier " +
...;
}
```
and `suspendedAppWithQueueNameDoesNotCreateKueueWorkload` can then assert
the full message, the way `suspendedAppDoesNotRequestDriver` already does.
##########
docs/spark_custom_resources.md:
##########
@@ -536,8 +536,12 @@ StatefulSets. Setting it back to `false` resumes the
regular lifecycle.
`Submitted` here is the operator's in-memory view. For a valid resource
created with
`suspend: true`, the initial `Submitted` status is not persisted to the API
server, so
-`kubectl get` shows an empty `Current State` until initialization resumes. An
application held
-later, in `ScheduledToRestart`, keeps the status its previous attempt already
wrote.
+`kubectl get` shows an empty `Current State` and no state transition events
are published until
+initialization resumes. Instead, the `SuspendHeld`
[event](configuration.md#kubernetes-events) is
+published when enabled. Since the status is not there to fall back on, it is
republished every 30
+minutes while the hold lasts, so that it outlives the event retention of the
API server. An application held later, in `ScheduledToRestart`, keeps
Review Comment:
**Finding 5.** 146 characters, where every other line of this paragraph is
100 or under. Re-wrapping 542-544 gives:
```
minutes 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.
```
--
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]
