peter-toth commented on code in PR #828: URL: https://github.com/apache/spark-kubernetes-operator/pull/828#discussion_r4018588596
########## docs/spark_custom_resources.md: ########## @@ -525,6 +525,34 @@ Note that `ttlAfterStopMillis` applies to the app as well as its secondary resou latter is smaller, then it takes higher precedence: operator would remove all resources related to this app after `ttlAfterStopMillis`. +## Suspend + +Both `SparkApplication` and `SparkCluster` support `.spec.suspend`. When it is set to `true`, the +operator keeps the resource in its initializing state (`Submitted`, or `ScheduledToRestart` for an +application that is scheduled to restart) and does not request the driver pod or the master / worker +StatefulSets. Setting it back to `false` resumes the regular lifecycle. Review Comment: **Finding 6.** `Submitted` is not something the user can observe here. With the dedicated `Suspended` state deferred to a follow-up ([#issuecomment-5684180823](https://github.com/apache/spark-kubernetes-operator/pull/828#issuecomment-5684180823)), the docs are the only place that can say so. For a `SparkApplication` created with `suspend: true`, no step in the pipeline writes a status. `AppValidateStep` persists only when the status is invalid and it never is. `AppCleanUpStep` returns `proceed()` for `Submitted`. The new branch returns before any persist. `toUpdateControl` is `noUpdate()`. So `.status` is absent on the API server and the `Current State` printer column is blank. That is indistinguishable from "the operator is not watching this namespace". `tests/e2e/watched-namespaces/chainsaw-test.yaml:54-58` asserts `.status == null` as the signature of exactly that case. Worth stating precisely, because the paragraph is only wrong for one of the two holds. An application suspended later, in `ScheduledToRestart`, does have a status on the server. `AppCleanUpStep` wrote it before the app got there. ```suggestion StatefulSets. Setting it back to `false` resumes the regular lifecycle. `Submitted` here is the operator's in-memory view. A resource created with `suspend: true` has no `.status` on the API server at all, so `kubectl get` shows an empty `Current State` until it resumes. An application suspended later, in `ScheduledToRestart`, keeps the status its previous attempt already wrote. ``` ########## tests/e2e/suspend/spark-application-suspended.yaml: ########## @@ -0,0 +1,24 @@ +# +# 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. +# + +apiVersion: spark.apache.org/v1 +kind: SparkApplication +metadata: + name: spark-job-suspend-test + namespace: default +spec: + suspend: true Review Comment: **Finding 7.** This file, and `spark-cluster-suspended.yaml`, now assert only `spec.suspend: true`. That is the value `spark-example-suspend.yaml` applied two steps earlier, on a field the operator never writes. The assertion cannot fail. Dropping the `status:` blocks was the right call. What is left is a 24-line file that adds nothing on top of the `error:` checks below it. Two options. Delete both files and let the `error:` checks carry the step. Or make them assert the hold, using the idiom already in the tree at `tests/e2e/watched-namespaces/chainsaw-test.yaml:54-58`: ```yaml - script: timeout: 30s content: kubectl get sparkapplication spark-job-suspend-test -n default -o json | jq ".status" check: (contains($stdout, 'null')): true ``` That pins today's behavior positively rather than by the absence of a pod. It will need updating when the `Suspended` state lands, which seems right. That follow-up is exactly the kind of change this test should notice. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java: ########## @@ -67,6 +67,10 @@ public ReconcileProgress reconcile( return proceed(); } SparkApplication app = context.getResource(); + if (app.getSpec().isSuspend()) { Review Comment: **Finding 8.** The gate asks "is the app initializing and suspended?", not "has anything been requested yet?". Those two differ in one window, and `appInitStepShouldBeIdempotentWhenStatusUpdateFails` (`AppInitStepTest.java:206`) exists because the codebase already knows about it. The sequence that produces it: 1. Reconcile N creates the pre-resources, the driver pod and the driver resources. 2. `attemptStatusUpdate` fails to persist `DriverRequested` and returns `completeAndImmediateRequeue()`. 3. `StatusRecorder.persistStatus` never updated `statusCache`, so `updateStatusFromCache` puts the resource back to `Submitted` on reconcile N+1. 4. A driver pod is live and the app reads as `Submitted`. An operator restart between the pod create and the status patch lands in the same place. The server has no `.status`, so `initStatus()` hands back a fresh `Submitted`. If `suspend` goes to `true` in that window, this branch fires and the app never leaves `Submitted`. `SparkAppReconciler.getReconcileSteps` adds only `AppInitStep` for `Submitted`, so no observer and no timeout ever looks at that pod. It runs to completion unnoticed and the resource is held forever. That contradicts `docs/spark_custom_resources.md:548`, "`suspend` only takes effect before the driver ... resources are requested". I reproduced it as a unit test at head. Reconcile 1 with `persistStatus` returning `false`, reset the status the way `updateStatusFromCache` would, set `suspend`, reconcile 2: ```java ReconcileProgress p2 = appInitStep.reconcile(ctx, recorder); Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), p2); Assertions.assertEquals( ApplicationStateSummary.Submitted, application.getStatus().getCurrentState().getCurrentStateSummary()); // the driver pod from reconcile 1 is still there Assertions.assertNotNull( kubernetesClient.pods().inNamespace("default").withName("driver-pod").get()); ``` It passes, so the hold wins over the live driver. The narrow fix here is `if (app.getSpec().isSuspend() && context.getDriverPod().isEmpty())`, which makes the code match the documented contract. One caveat I could not rule out. `getDriverPod()` reads the informer cache, and in `ScheduledToRestart` a just-deleted pod from the previous attempt could still be in it, which would make the operator start a driver for a suspended app. If that risk is not worth taking in this PR, the running-attempt suspend you have planned is the natural home for this window. A suspend that can stop a live attempt stops this one too. Worth carrying into that work rather than leaving it only in this thread. ########## docs/spark_custom_resources.md: ########## @@ -525,6 +525,34 @@ Note that `ttlAfterStopMillis` applies to the app as well as its secondary resou latter is smaller, then it takes higher precedence: operator would remove all resources related to this app after `ttlAfterStopMillis`. +## Suspend + +Both `SparkApplication` and `SparkCluster` support `.spec.suspend`. When it is set to `true`, the +operator keeps the resource in its initializing state (`Submitted`, or `ScheduledToRestart` for an +application that is scheduled to restart) and does not request the driver pod or the master / worker +StatefulSets. Setting it back to `false` resumes the regular lifecycle. + +``` yaml +apiVersion: spark.apache.org/v1 +kind: SparkApplication +metadata: + name: suspended-pi +spec: + suspend: true + mainClass: "org.apache.spark.examples.SparkPi" + jars: "local:///opt/spark/examples/jars/spark-examples.jar" + runtimeVersions: + sparkVersion: "4.2.0" +``` + +* `suspend` only takes effect before the driver (or master / worker) resources are requested. + Setting it to `true` on a running application does not stop the current attempt. If the + application is configured to restart, the next attempt is held until `suspend` is set back to + `false`. Setting it to `true` on a running cluster has no effect in the current version. +* Deleting a suspended resource works as usual. +* This is the building block for external job queueing systems such as + [Kueue](https://kueue.sigs.k8s.io/), which admit a workload by flipping `suspend` to `false`. Review Comment: **Finding 9.** Reads as if Kueue does this today. `KueueWorkloadFactory` is the only code outside the two init steps that touches `spec.isSuspend()`, and it has no caller in `src/main`. Nothing creates a `Workload`, and nothing flips `suspend` back. ```suggestion [Kueue](https://kueue.sigs.k8s.io/), which admit a workload by flipping `suspend` to `false`. The operator does not integrate with such a system yet. ``` -- 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]
