[SPARK-60025] Support spec.suspend for a running SparkApplication

build-tools/helm/spark-kubernetes-operator/templates/kueue-validating-admission-policy.yaml

@@ -15,7 +15,7 @@
 
 {{/*
 Keep the queue-name label of a SparkApplication or SparkCluster from changing once it started,
-until the application is scheduled to restart or the cluster is Suspended. See the Kueue section
+until the application is scheduled to restart or the resource is Suspended. See the Kueue section
 of docs/spark_custom_resources.md.
 */}}
 {{- if .Values.operatorRbac.kueue.enabled }}
@@ -60,14 +60,14 @@ spec:
         ? oldObject.status.currentState.currentStateSummary : ''
   validations:
     # The states in which ApplicationStateSummary or ClusterStateSummary#isInitializing holds, i.e.
-    # before an attempt starts, and ClusterStateSummary.Suspended. Keep them in sync.
+    # before an attempt starts, and the Suspended state of both. Keep them in sync.
     - expression: >-
         variables.oldQueue == variables.newQueue
         || variables.oldState in ['', 'Submitted', 'ScheduledToRestart', 'Suspended']
       message: >-
         The kueue.x-k8s.io/queue-name label can be changed only before the resource starts, i.e.
-        while it is Submitted or ScheduledToRestart, or while a SparkCluster is Suspended, since
-        the Kueue Workload admitted for it keeps its quota in that queue.
+        while it is Submitted or ScheduledToRestart, or while it is Suspended, since the Kueue
+        Workload admitted for it keeps its quota in that queue.
       reason: Forbidden
 ---
 apiVersion: admissionregistration.k8s.io/v1

docs/architecture.md

@@ -138,6 +138,9 @@ stateDiagram-v2
     RunningWithPartialCapacity --> Succeeded
     RunningWithPartialCapacity --> Failed
 
+    RunningHealthy --> Suspended : spec.suspend=true
+    Suspended --> Submitted : spec.suspend=false, new attempt
+
     state Failures {
         SchedulingFailure
         DriverStartTimedOut
@@ -165,6 +168,10 @@ stateDiagram-v2
 * Once the driver is requested, an application moves to `Succeeded`, `Failed` or `DriverEvicted`
   from any of the states above whenever the driver pod terminates or is evicted. It also moves to
   `Failed` if the driver pod is removed unexpectedly.
+* Likewise, an application moves to `Suspended` from any of the states from `DriverRequested` to
+  `RunningWithBelowThresholdExecutors` when [`spec.suspend`](spark_custom_resources.md#suspend) is
+  set to `true`. Its driver and executors are released, and setting it back to `false` starts a
+  new attempt from `Submitted`, which does not count as a restart.
 * User may configure the app CR to time-out after given threshold of time if it cannot reach healthy
   state after given threshold. The timeout can be configured for different lifecycle stages,
   when driver starting, when driver becoming ready, and when requesting executor pods.

docs/config_properties.md

@@ -53,7 +53,7 @@
  | spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds | Integer | 5 | false | Initial interval(in seconds) of retries on unhandled controller errors. | 
  | spark.kubernetes.operator.reconciler.retry.intervalMultiplier | Double | 1.5 | false | Interval multiplier of retries on unhandled controller errors. Setting this to 1 for linear retry. | 
  | spark.kubernetes.operator.reconciler.retry.maxIntervalSeconds | Integer | -1 | false | Max interval(in seconds) of retries on unhandled controller errors. Set to non-positive for unlimited. | 
- | spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds | Long | 1800 | true | Requeue interval (in seconds) at which a resource held by spec.suspend is reconciled, so that its SuspendHeld event is republished. A SparkCluster suspended while running publishes no such event, since its Suspended status says so, and uses this interval only to look at its released resources again. The hold ends only when a user clears spec.suspend, which arrives as a watch event and reconciles right away, so nothing waits for this interval. It is deliberately much coarser than spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended resource has nothing to observe while each republish costs a read and a write on the API server. Keep it below the '--event-ttl' of the API server (one hour by default), or the event expires between repeats and the hold stops being visible, and keep 'spark.kubernetes.operator.events.minIntervalSeconds' within the rest of that TTL, that is at most the TTL minus this interval, or a repeat is dropped and the next one lands after the event expired. | 
+ | spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds | Long | 1800 | true | Requeue interval (in seconds) at which a resource held by spec.suspend is reconciled, so that its SuspendHeld event is republished. A SparkApplication or SparkCluster suspended while running publishes no such event, since its Suspended status says so, and uses this interval only to look at its released resources again. The hold ends only when a user clears spec.suspend, which arrives as a watch event and reconciles right away, so nothing waits for this interval. It is deliberately much coarser than spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended resource has nothing to observe while each republish costs a read and a write on the API server. Keep it below the '--event-ttl' of the API server (one hour by default), or the event expires between repeats and the hold stops being visible, and keep 'spark.kubernetes.operator.events.minIntervalSeconds' within the rest of that TTL, that is at most the TTL minus this interval, or a repeat is dropped and the next one lands after the event expired. | 
  | spark.kubernetes.operator.reconciler.terminationTimeoutSeconds | Integer | 30 | false | Grace period for operator shutdown before reconciliation threads are killed. | 
  | spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled | Boolean | true | true | When enabled, operator would trim state transition history when a new attempt starts, keeping previous attempt summary only. It also drops the history of a suspended SparkCluster when it is resumed. | 
  | spark.kubernetes.operator.watchedNamespaces | String | default | true | Comma-separated list of namespaces that the operator would be watching for Spark resources. If set to '*', operator would watch all namespaces. | 

docs/configuration.md

@@ -100,9 +100,9 @@ replacing its `message`. A repeat carrying the same `message` may be paced by
 | `Normal` | All other states, e.g. `ScheduledToRestart`, `DriverRequested`, `DriverStarted`, `DriverReady`, `InitializedBelowThresholdExecutors`, `RunningHealthy`, `RunningWithPartialCapacity`, `Succeeded`, `ResourceReleased` and `Suspended` |
 
 The initial `Submitted` status of a new resource is not persisted to the API server on its own, so
-no event is published for it. A `Submitted` event is published only when a `SparkCluster` moves
-from `Suspended` back to `Submitted`, after `spec.suspend` is set back to `false` or after its
-eviction by Kueue.
+no event is published for it. A `Submitted` event is published only when a `SparkApplication` or a
+`SparkCluster` moves from `Suspended` back to `Submitted`, after `spec.suspend` is set back to
+`false`, or after the eviction of a `SparkCluster` by Kueue.
 
 In addition, the operator publishes the following `Warning` events.
 
@@ -113,11 +113,11 @@ 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, 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. |
+| `SuspendReleaseFailed` | Deleting the driver pod (`SparkApplication`), the master and worker StatefulSets or the HorizontalPodAutoscaler or PodDisruptionBudget of the workers (`SparkCluster`), or the Kueue `Workload` of a resource 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. |
 | `KueueResourceFlavorReadFailed` | Reading the `ResourceFlavor`s which Kueue assigned to an admitted `Workload` fails, so the node selector and the tolerations of the flavors cannot be applied. Transport-level errors are skipped and retried every 5 seconds, while a persistent failure is retried with the default interval. |
 | `KueuePodsReadyUpdateFailed` | Recording the `PodsReady` condition on the Kueue `Workload` of a running `SparkApplication` or `SparkCluster` fails, e.g. without the permission for the `workloads/status` subresource, so Kueue may evict it by a `PodsReadyTimeout`, which releases a running `SparkCluster` and queues it again even if its master and workers are ready. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure with the default interval, while a `SparkApplication` retries any failure with its next reconciliation, so that the recording does not hold back the observation of its driver. See [Kueue](spark_custom_resources.md#kueue). |
 | `KueueDisabled` | A `SparkApplication` or `SparkCluster` labeled with `kueue.x-k8s.io/queue-name` is not queued, since `spark.kubernetes.operator.kueue.enabled` is disabled, so the driver (or master and worker) is requested without the Kueue admission. |
-| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a `SparkApplication` whose driver is requested and whose attempt has not stopped. The operator does not act on it, so the driver and executors keep running. An evicted `Workload` keeps holding its quota, while a deactivated one no longer counts against it. It is republished on every reconciliation while it lasts, subject to [`minIntervalSeconds`](#event-frequency). See [Kueue](spark_custom_resources.md#kueue). |
+| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a `SparkApplication` whose driver is requested and whose attempt has neither stopped nor been suspended. The operator does not act on it, so the driver and executors keep running. An evicted `Workload` keeps holding its quota, while a deactivated one no longer counts against it. It is republished on every reconciliation while it lasts, subject to [`minIntervalSeconds`](#event-frequency). See [Kueue](spark_custom_resources.md#kueue). |
 
 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`