peter-toth commented on code in PR #875:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4084901724
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -151,19 +161,40 @@ public ReconcileProgress reconcile(
.appendNewState(new ClusterState(RunningHealthy,
CLUSTER_READY_MESSAGE));
statusRecorder.persistStatus(context, updatedStatus);
return completeAndDefaultRequeue();
- } catch (Exception e) {
- if (log.isErrorEnabled()) {
- log.error("Failed to request master resource.", e);
+ } catch (KubernetesClientException e) {
+ if (ReconcilerUtils.isRetryableError(e)) {
+ // SchedulingFailure is terminal for a cluster, so a request which may
yet succeed is sent
+ // again, e.g. for a cluster resumed from Suspended, while a rejected
one fails the cluster.
+ log.warn("Failed to request master resource, will retry.", e);
Review Comment:
**Finding 2.** This is the only retry path in the operator that a user
cannot see.
Before this change a 500 or a 429 while applying the resources produced
`SchedulingFailure` with `CLUSTER_SCHEDULE_FAILURE_MESSAGE` plus the stack
trace in `.status`, which is terminal but visible. Now a failure that keeps
repeating, such as a mutating webhook on StatefulSets that stays down, leaves
the cluster in `Submitted` with an empty-looking status and nothing but this
`log.warn` in the operator's own logs. The requeue is the default interval, so
it goes round forever.
Every sibling reports itself: `KueueWorkloadUtils.retryAfterRequestFailure`
publishes `KueueAdmissionRequestFailed`, `ClusterSuspendStep` publishes
`SuspendReleaseFailed` on the same classification, and a rejected status patch
publishes `StatusUpdateFailed`. Making this branch the exception is what hides
it.
```java
if (ReconcilerUtils.isRetryableError(e)) {
log.warn("Failed to request master resource, will retry.", e);
EventUtils.warn(
context.getEventRecorder(),
EventUtils.REASON_CLUSTER_REQUEST_FAILED,
"Failed to request the master and workers of the cluster, will
retry. "
+ EventUtils.describe(e));
return completeAndDefaultRequeue();
}
```
with the new reason added to the `Warning` table in `docs/configuration.md`,
next to the rows this one reads like. The `events.minIntervalSeconds` pacing
already keeps a repeating reason from becoming a write per reconcile, so the
cost is bounded.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java:
##########
@@ -48,15 +48,19 @@ public abstract ReconcileProgress reconcile(
* @param statusRecorder The SparkClusterStatusRecorder for recording status
updates.
* @param updatedStatus The updated ClusterStatus.
* @param requeueAfter The duration after which to re-queue.
- * @return The ReconcileProgress indicating the re-queue.
+ * @return The ReconcileProgress indicating the re-queue, or the default
re-queue if the status
+ * is not persisted.
*/
protected ReconcileProgress updateStatusAndRequeueAfter(
SparkClusterContext context,
SparkClusterStatusRecorder statusRecorder,
ClusterStatus updatedStatus,
Duration requeueAfter) {
- statusRecorder.persistStatus(context, updatedStatus);
- return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
+ // A rejected update, e.g. of a state unknown to an older CRD, would be
rejected again right
+ // away, so it is retried with the default interval, not immediately.
+ return statusRecorder.persistStatus(context, updatedStatus)
Review Comment:
**Finding 3.** The `AppReconcileStep` twins of these two helpers still throw
the result away, so the same rejected patch is silent on the application side.
`AppReconcileStep.updateStatusAndRequeueAfter` and
`appendStateAndRequeueAfter` call `persistStatus` / `appendNewStateAndPersist`
as statements and return the requeue regardless. `AppCleanUpStep` calls them
four times, at `:143`, `:175`, `:194` and `:216`, so a patch the API server
rejects still reads as a completed step there. Mirroring the two ternaries you
added here, at `:61` and `:84`, would close it, and a follow-up ticket is fine
given the size of this PR.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * 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.Constants.CLUSTER_RESUMED_MESSAGE;
+import static
org.apache.spark.k8s.operator.Constants.CLUSTER_SUSPENDED_MESSAGE;
+import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_CLUSTER_NAME;
+import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_MASTER_VALUE;
+import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_NAME;
+import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_WORKER_VALUE;
+import static
org.apache.spark.k8s.operator.SparkClusterResourceSpec.getWorkerHorizontalPodAutoscalerName;
+import static
org.apache.spark.k8s.operator.SparkClusterResourceSpec.getWorkerPodDisruptionBudgetName;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.KUEUE_WORKLOAD_INFORMER_ENABLED;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.TRIM_ATTEMPT_STATE_TRANSITION_HISTORY;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndRequeueAfter;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed;
+
+import java.time.Duration;
+
+import io.fabric8.kubernetes.api.model.ListOptionsBuilder;
+import io.fabric8.kubernetes.api.model.PodList;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.spark.k8s.operator.SparkCluster;
+import org.apache.spark.k8s.operator.context.SparkClusterContext;
+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.status.ClusterState;
+import org.apache.spark.k8s.operator.status.ClusterStateSummary;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
+import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
+import org.apache.spark.k8s.operator.utils.StringUtils;
+
+/**
+ * Suspends a running cluster by spec.suspend, and resumes it. A cluster runs
until it is deleted,
+ * so suspending it is the only way to give its quota back without deleting it.
+ *
+ * <p>A running cluster enters Suspended before anything is released, and the
master and workers
+ * are released only in Suspended. Releasing them first would leave a cluster
without them in
+ * RunningHealthy when spec.suspend is cleared in the meantime, and nothing
recreates them there.
+ * A resumed cluster leaves Suspended only after everything is released, and
then starts over from
+ * Submitted, where {@link ClusterInitStep} applies all of its resources again
and requests a new
+ * Kueue admission. Resuming earlier would let it keep the StatefulSets or the
admitted Workload of
+ * the previous spec, so that a spec changed in the meantime would run on the
old admission.
+ */
+@Slf4j
+public final class ClusterSuspendStep extends ClusterReconcileStep {
+ /**
+ * Reconciles a running or suspended cluster against spec.suspend.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status
updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
+ @Override
+ public ReconcileProgress reconcile(
+ SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
+ SparkCluster cluster = context.getResource();
+ boolean suspend = cluster.getSpec().isSuspend();
+ ClusterStateSummary summary =
cluster.getStatus().getCurrentState().getCurrentStateSummary();
+ if (summary == ClusterStateSummary.RunningHealthy && suspend) {
+ return appendStateAndImmediateRequeue(
+ context,
+ statusRecorder,
+ new ClusterState(ClusterStateSummary.Suspended,
CLUSTER_SUSPENDED_MESSAGE));
+ }
+ if (summary == ClusterStateSummary.Suspended) {
+ if (!releaseResources(context)) {
+ return completeAndDefaultRequeue();
+ }
+ if (!suspend) {
+ // Like a restarted application attempt, the resumed cluster drops the
history of the run
+ // which is over, so that suspending and resuming it again and again
keeps it bounded.
+ return updateStatusAndRequeueAfter(
+ context,
+ statusRecorder,
+ cluster
+ .getStatus()
+ .appendNewState(
+ new ClusterState(ClusterStateSummary.Submitted,
CLUSTER_RESUMED_MESSAGE),
+ TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue()),
+ Duration.ZERO);
+ }
+ return completeAndRequeueAfter(
+
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
+ }
+ return proceed();
+ }
+
+ /**
+ * Releases the master and workers of a suspended cluster, and then its
Kueue Workload. The
+ * Services and the NetworkPolicy are kept, since they hold no quota and are
applied again on
+ * resume. The HorizontalPodAutoscaler and the PodDisruptionBudget of the
workers are deleted by
+ * name rather than from the current spec, which may no longer ask for them,
so that the resumed
+ * cluster gets only those which its spec asks for.
+ *
+ * <p>The Workload is released only after the master and worker pods are
gone, since Kueue would
+ * admit another workload into the quota which the terminating pods still
occupy. Other pods which
+ * carry the cluster label, e.g. to reach the workers through their
NetworkPolicy, do not count.
+ * Everything here is idempotent, so pods that are not gone yet, or a
release that failed, are
+ * simply looked at again. A failure that sending the request again would
not fix is reported.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @return True once everything is released, false while pods remain or a
release failed.
+ */
+ private boolean releaseResources(SparkClusterContext context) {
+ SparkCluster cluster = context.getResource();
+ String namespace = cluster.getMetadata().getNamespace();
+ String name = cluster.getMetadata().getName();
+ KubernetesClient client = context.getClient();
+ try {
+ client.resource(context.getWorkerStatefulSetSpec()).delete();
Review Comment:
**Finding 1.** This call builds the whole resource spec, and a build failure
is not a `KubernetesClientException`, so it escapes the `catch` at `:175` and
leaves the reconciler.
`getWorkerStatefulSetSpec()` goes through
`SparkClusterContext.getSecondaryResourceSpec()` to
`SparkClusterResourceSpecFactory.buildResourceSpec`, which runs the submission
worker over the current spec. `ClusterInitStep` treats that as fallible and
wraps the same call in `catch (Exception e)` at `:172`, and
`ClusterValidateStep.reconcile` is a bare `return proceed()`, so nothing
rejects such a spec earlier.
The path is one this feature invites rather than an exotic one. Your own
class javadoc gives editing the spec while suspended as the reason resume must
start over, so an edit that the worker can no longer build lands squarely here.
I ran it: stub `getWorkerStatefulSetSpec()` to throw an
`IllegalArgumentException` on a `Suspended` cluster and
`ClusterSuspendStep.reconcile` propagates it, with the state still `Suspended`.
So the master and worker StatefulSets are never deleted, the pods keep running,
the Kueue quota is never given back, and clearing `spec.suspend` reaches the
same call first, so the cluster cannot leave `Suspended` either. All that
surfaces is JOSDK's generic `ReconcileError`.
A second catch keeps the invariant that nothing is reported as released
until it is, and reports the failure on the reason you already added:
```java
} catch (Exception e) {
// The spec may have been edited into one which cannot be built while
the cluster was
// suspended, and its resources must not be left behind, so the
failure is reported and the
// release is looked at again rather than escaping the reconciler.
log.warn("Failed to build the resources of the suspended cluster, will
retry.", e);
EventUtils.warn(
context.getEventRecorder(),
EventUtils.REASON_SUSPEND_RELEASE_FAILED,
"Failed to build the master and workers of the suspended cluster,
will retry. "
+ EventUtils.describe(e));
return false;
}
```
I applied exactly that and re-ran `:spark-operator:test`: 429 tests green,
and the probe returns `completeAndDefaultRequeue()` instead of throwing. The
`docs/configuration.md` row for `SuspendReleaseFailed` would want "or its
resources cannot be built" added to its list.
Deleting the two StatefulSets by name is the other way, and it matches what
you already do one line below for the HPA and the PDB:
`SparkClusterResourceSpec` builds them as `<cluster>-master` and
`<cluster>-worker` from `getMetadata().getName()`, so two more helpers next to
`getWorkerHorizontalPodAutoscalerName` would remove the spec dependency
entirely. That one moves the
`failedReleaseIsRetriedWithoutReleasingKueueWorkload` stub off
`resource(workerStatefulSetSpec)`, so it is the larger change of the two.
--
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]
