You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "dongjoon-hyun (via GitHub)" <gi...@apache.org> on 2026/09/23 15:33:06 UTC
[PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
dongjoon-hyun opened a new pull request, #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875
### What changes were proposed in this pull request?
This PR aims to support `spec.suspend` for a running `SparkCluster` via a new `Suspended` state.
- `RunningHealthy` with `spec.suspend=true` moves to `Suspended`, where the operator deletes the master/worker `StatefulSet`s, the worker HPA/PDB, and then the Kueue `Workload` after the pods are gone.
- Setting `spec.suspend=false` moves the cluster back to `Submitted` once everything is released, and the cluster is requested again from scratch.
- `ClusterInitStep` retries transient API failures (e.g. timeout, `429`, `500`) instead of failing the cluster with the terminal `SchedulingFailure`.
### Why are the changes needed?
Currently, `spec.suspend` has no effect on a running `SparkCluster`, so deleting the cluster is the only way to release its resources and its Kueue quota.
### Does this PR introduce _any_ user-facing change?
Yes. A running `SparkCluster` can be suspended and resumed, with a new `Suspended` state and `SuspendReleaseFailed` event.
### How was this patch tested?
Pass the CIs with the newly added unit tests and E2E tests (`tests/e2e/suspend`, `tests/e2e/kueue`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#issuecomment-5800113493
Merged to main
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4085339382
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.getMasterStatefulSetName;
+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.SparkClusterResourceSpec.getWorkerStatefulSetName;
+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. Everything is deleted by name rather than from the current spec. The spec may no longer
+ * ask for the HorizontalPodAutoscaler or the PodDisruptionBudget of the workers, and it may not
+ * even build after being edited while suspended, which would otherwise hold the pods and the
+ * quota until the spec is fixed.
+ *
+ * <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
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getWorkerStatefulSetName(name))
+ .delete();
+ client
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getMasterStatefulSetName(name))
+ .delete();
+ client
+ .autoscaling()
+ .v2()
+ .horizontalPodAutoscalers()
+ .inNamespace(namespace)
+ .withName(getWorkerHorizontalPodAutoscalerName(name))
+ .delete();
+ client
+ .policy()
+ .v1()
+ .podDisruptionBudget()
+ .inNamespace(namespace)
+ .withName(getWorkerPodDisruptionBudgetName(name))
+ .delete();
+ // Only whether any pod remains matters, so one is enough. A page may come back empty with a
+ // continue token, which still means that more pods remain.
+ PodList pods =
+ client
+ .pods()
+ .inNamespace(namespace)
+ .withLabel(LABEL_SPARK_CLUSTER_NAME, name)
+ .withLabelIn(
+ LABEL_SPARK_ROLE_NAME,
+ LABEL_SPARK_ROLE_MASTER_VALUE,
+ LABEL_SPARK_ROLE_WORKER_VALUE)
+ .list(new ListOptionsBuilder().withLimit(1L).build());
+ boolean morePods = StringUtils.isNotEmpty(pods.getMetadata().getContinue());
+ if (!pods.getItems().isEmpty() || morePods) {
+ // The deletion of each pod is observed by the pod informer, which reconciles again.
+ log.debug("Waiting for the pods of the suspended cluster to be deleted.");
+ return false;
+ }
+ // A Workload admitted before its queue label was removed is released as well, whenever the
+ // operator may access Workloads, which the Workload informer requires too.
+ if (KueueWorkloadFactory.hasQueueName(cluster)
+ || KUEUE_WORKLOAD_INFORMER_ENABLED.getValue()) {
+ KueueWorkloadUtils.deleteWorkloadOf(client, cluster);
+ }
+ } catch (KubernetesClientException e) {
+ log.warn("Failed to release the resources of the suspended cluster, will retry.", e);
+ if (!ReconcilerUtils.isRetryableError(e)) {
Review Comment:
Thank you, @peter-toth. I changed the gate to `isTransientError` in 9f38f1f, so a 429 or a 500 while releasing is now reported as `SuspendReleaseFailed`, consistent with `ClusterRequestFailed` and the docs. `rejectedReleaseIsReportedAndRetried` became `persistentReleaseFailureIsReportedAndRetried` over `{403, 429, 500}`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#issuecomment-5799435999
All tests passed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4085048164
##########
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:
Thank you, @peter-toth. I switched the release to delete the master and worker `StatefulSet`s by name, like the HPA and the PDB, so `ClusterSuspendStep` no longer builds the spec at all. An unbuildable spec now only fails the cluster with `SchedulingFailure` on resume, as for a new cluster. I also added `suspendedClusterWhoseSpecCannotBeBuiltIsReleased` in d742ee5.
##########
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:
Thanks. I added a `ClusterRequestFailed` warning event for the retried branch, skipping transport-level errors like the other reasons, and documented it in `configuration.md` in d742ee5.
##########
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:
Agreed. Since it is an existing issue on the application side, I will file a new JIRA issue for it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #875: [SPARK-59750] Support `spec.suspend` for a running `SparkCluster`
URL: https://github.com/apache/spark-kubernetes-operator/pull/875
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4085091745
##########
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:
I re-checked the code and realized that SPARK-53706 resolved it already.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4085278880
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.getMasterStatefulSetName;
+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.SparkClusterResourceSpec.getWorkerStatefulSetName;
+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. Everything is deleted by name rather than from the current spec. The spec may no longer
+ * ask for the HorizontalPodAutoscaler or the PodDisruptionBudget of the workers, and it may not
+ * even build after being edited while suspended, which would otherwise hold the pods and the
+ * quota until the spec is fixed.
+ *
+ * <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
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getWorkerStatefulSetName(name))
+ .delete();
+ client
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getMasterStatefulSetName(name))
+ .delete();
+ client
+ .autoscaling()
+ .v2()
+ .horizontalPodAutoscalers()
+ .inNamespace(namespace)
+ .withName(getWorkerHorizontalPodAutoscalerName(name))
+ .delete();
+ client
+ .policy()
+ .v1()
+ .podDisruptionBudget()
+ .inNamespace(namespace)
+ .withName(getWorkerPodDisruptionBudgetName(name))
+ .delete();
+ // Only whether any pod remains matters, so one is enough. A page may come back empty with a
+ // continue token, which still means that more pods remain.
+ PodList pods =
+ client
+ .pods()
+ .inNamespace(namespace)
+ .withLabel(LABEL_SPARK_CLUSTER_NAME, name)
+ .withLabelIn(
+ LABEL_SPARK_ROLE_NAME,
+ LABEL_SPARK_ROLE_MASTER_VALUE,
+ LABEL_SPARK_ROLE_WORKER_VALUE)
+ .list(new ListOptionsBuilder().withLimit(1L).build());
+ boolean morePods = StringUtils.isNotEmpty(pods.getMetadata().getContinue());
+ if (!pods.getItems().isEmpty() || morePods) {
+ // The deletion of each pod is observed by the pod informer, which reconciles again.
+ log.debug("Waiting for the pods of the suspended cluster to be deleted.");
+ return false;
+ }
+ // A Workload admitted before its queue label was removed is released as well, whenever the
+ // operator may access Workloads, which the Workload informer requires too.
+ if (KueueWorkloadFactory.hasQueueName(cluster)
+ || KUEUE_WORKLOAD_INFORMER_ENABLED.getValue()) {
+ KueueWorkloadUtils.deleteWorkloadOf(client, cluster);
+ }
+ } catch (KubernetesClientException e) {
+ log.warn("Failed to release the resources of the suspended cluster, will retry.", e);
+ if (!ReconcilerUtils.isRetryableError(e)) {
Review Comment:
**Finding 4.** This gate and the one you just added in `ClusterInitStep` disagree on exactly two codes.
`ClusterInitStep:170` skips the event when `isTransientError(e)`, so a 500 or a 429 there is reported as `ClusterRequestFailed`. This line skips it when `isRetryableError(e)`, which is `isTransientError` plus those same two codes, so a 500 or a 429 while releasing is reported nowhere. Measured on a `Suspended` cluster with `client.apps()` throwing:
| code | event today |
|---|---|
| 503 | none |
| 429 | none |
| 500 | none |
| 403 | `SuspendReleaseFailed` |
| 422 | `SuspendReleaseFailed` |
The asymmetry costs more here than there. `Submitted` with nothing happening is already anomalous, while `Suspended` is a legitimate steady state, so nothing tells a user whether a cluster is waiting for its pods to go or has been getting a 500 on the StatefulSet delete for an hour. The `docs/configuration.md` row describes the narrower set as well, "A failure which may clear on its own, such as a timeout, is skipped", which is `isTransientError`, not what the code skips.
```suggestion
if (!ReconcilerUtils.isTransientError(e)) {
```
I applied that and re-ran: 429 and 500 then report `SuspendReleaseFailed`, 503 still does not, and the whole `:spark-operator` suite stays green, including `failedReleaseIsRetriedWithoutReleasingKueueWorkload`, which throws 503 and asserts the recorder is never touched. Nothing pins the newly reported half though, so `rejectedReleaseIsReportedAndRetried` would want the `@CsvSource` treatment you gave `retryableFailureOfRequestingResourcesIsRetried`, over `{403, 429, 500}`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#discussion_r4085106347
##########
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:
Correction: this was already addressed by SPARK-53706 (e80dead). On `main`, `AppReconcileStep.updateStatusAndRequeueAfter` goes through `attemptStatusUpdate`, and `appendStateAndRequeueAfter` checks the result of `appendNewStateAndPersist`, both returning `completeAndImmediateRequeue()` when the status is not persisted. So `AppCleanUpStep` already does not report progress on a rejected patch, and no new JIRA issue is needed. The only difference is that the cluster side here retries with the default interval, since a status rejected by an older CRD, e.g. for `Suspended`, would be rejected again right away.
##########
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:
I re-checked the code and realized that SPARK-53706 resolved it already.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59750] Support `spec.suspend` for a running `SparkCluster` [spark-kubernetes-operator]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #875:
URL: https://github.com/apache/spark-kubernetes-operator/pull/875#issuecomment-5799694878
Thank you so much, @peter-toth !
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org