peter-toth commented on code in PR #828: URL: https://github.com/apache/spark-kubernetes-operator/pull/828#discussion_r4017891284
########## tests/e2e/suspend/spark-application-suspended.yaml: ########## @@ -0,0 +1,30 @@ +# +# 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 +status: Review Comment: **Finding 1.** Nothing in a suspended reconcile ever writes `.status` to the API server, so this assertion (and the `spark-cluster-suspended.yaml` twin) has no `status` to match. The trace, for a `SparkApplication` created with `suspend: true`: - `AppValidateStep` persists only when the status is *invalid*, and it never is. `CustomResource`'s constructor calls `initStatus()`, so `getStatus()` returns a `Submitted` `ApplicationStatus` even for a CR the server stored without one. `isValidApplicationStatus` passes and the step returns `proceed()`. - `AppCleanUpStep` returns `proceed()` for `Submitted` without persisting. - the new branch returns `completeAndDefaultRequeue()` before any persist. - `ReconcilerUtils.toUpdateControl` returns `UpdateControl.noUpdate()`, and the CRD declares `subresources: status: {}`, so JOSDK writes nothing either. `SparkCluster` is the same, except `ClusterValidateStep` is an unconditional `proceed()`, so there is not even a reset path. I ran the three steps of each pipeline against a mock recorder: ```java SparkApplication app = new SparkApplication(); app.setMetadata(new ObjectMetaBuilder().withName("a").withNamespace("default").build()); app.getSpec().setSuspend(true); SparkAppContext ctx = mock(SparkAppContext.class); when(ctx.getResource()).thenReturn(app); SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class); new AppValidateStep().reconcile(ctx, recorder); new AppCleanUpStep().reconcile(ctx, recorder); new AppInitStep().reconcile(ctx, recorder); verifyNoInteractions(recorder); // passes; the ClusterValidate/Terminated/Init trio passes too ``` So a held resource has no `status` on the server and an empty `Current State` printer column. Chainsaw then fails here on `status.currentState`, and on `(*.currentStateSummary)` over a missing `stateTransitionHistory`. I have no cluster here, so that last hop is read from the assertion files rather than observed. Two ways out: - Persist the initial status once when entering the hold, so the resource really is `Submitted` on the server. Careful: `StatusRecorder.updateStatusFromCache` seeds `statusCache` with the current status on a cache miss, and `patchAndStatusWithVersionLocked` short-circuits on `newStatusNode.equals(previousStatusNode)`. A plain `persistStatus(context, app.getStatus())` in the new branch is therefore dropped silently on the first reconcile. - Or drop the `status:` blocks from `spark-application-suspended.yaml` and `spark-cluster-suspended.yaml` and assert only `spec.suspend: true` plus the existing `error:` checks on the driver pod and the StatefulSets. The first also fixes the operator-facing half. With Kueue holding a queue of workloads, `kubectl get sparkapp` currently shows a blank state for every one of them. ########## docs/spark_custom_resources.md: ########## @@ -525,6 +525,32 @@ 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 or cluster has no effect in the current version. Review Comment: **Finding 2.** Not true for a `SparkApplication` configured with a restart policy. `AppCleanUpStep` has no suspend check, so a running attempt that fails still goes through `terminateOrRestart` and lands in `ScheduledToRestart` (`AppCleanUpStep.java:186-193`). The next reconcile hits the new gate and holds there until someone flips the flag back. `suspendedAppScheduledToRestartDoesNotRequestDriver` asserts exactly that hold. So `suspend: true` on a running app does not stop the current attempt, but it does stop every attempt after it. For `SparkCluster` the bullet is accurate: `SparkClusterReconciler.getReconcileSteps` only adds `ClusterInitStep` for `Submitted`, and a `RunningHealthy` cluster never goes back there. ```suggestion 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. ``` The PR description's "Suspending an already running application or cluster is out of scope" reads the same way and is worth the same correction. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java: ########## @@ -59,6 +59,10 @@ public ReconcileProgress reconcile( return proceed(); } SparkCluster cluster = context.getResource(); + if (cluster.getSpec().isSuspend()) { + log.info("Cluster is suspended, master and worker resources would not be requested."); Review Comment: **Finding 4.** A held resource logs this on every reconcile, so once per `spark.kubernetes.operator.reconciler.intervalSeconds` (120s by default) for as long as it stays suspended. With Kueue holding a few hundred workloads that is steady INFO traffic for a no-op. The other steady-state no-op paths report at debug (`AppCleanUpStep.java:132`, `AppReconcileStep.java:86`). Same line at `AppInitStep.java:71`. ########## spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java: ########## @@ -342,4 +344,102 @@ void nonTrimModeRestartBackoffActiveRequeuesWithDelay() { ApplicationStateSummary.ScheduledToRestart, application.getStatus().getCurrentState().getCurrentStateSummary()); } + + @Test + void suspendedAppDoesNotRequestDriver() { + AppInitStep appInitStep = new AppInitStep(); + SparkAppContext mockContext = mock(SparkAppContext.class); + SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class); + SparkApplication application = new SparkApplication(); + application.setMetadata(applicationMetadata); + application.getSpec().setSuspend(true); + when(mockContext.getResource()).thenReturn(application); + + ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder); + + Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), progress); + verify(mockContext, never()).getDriverPreResourcesSpec(); + verify(mockContext, never()).getDriverPodSpec(); + verify(mockContext, never()).getClient(); + verifyNoInteractions(recorder); + Assertions.assertEquals( + ApplicationStateSummary.Submitted, + application.getStatus().getCurrentState().getCurrentStateSummary()); + } + + @Test + void suspendedAppScheduledToRestartDoesNotRequestDriver() { + // ScheduledToRestart with an elapsed backoff: suspend takes precedence over restart. + AppInitStep appInitStep = new AppInitStep(); + SparkAppContext mockContext = mock(SparkAppContext.class); + SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class); + SparkApplication application = new SparkApplication(); + application.setMetadata(applicationMetadata); + application.getSpec().setSuspend(true); + application.getSpec().setApplicationTolerations( + ApplicationTolerations.builder() + .restartConfig(RestartConfig.builder().restartBackoffMillis(5000L).build()) + .build()); + ApplicationState timedOutState = + new ApplicationState(ApplicationStateSummary.DriverStartTimedOut, "timed out"); + ApplicationState scheduledState = + new ApplicationState(ApplicationStateSummary.ScheduledToRestart, "restarting"); + scheduledState.setLastTransitionTime(Instant.now().minusMillis(60000L).toString()); + Map<Long, ApplicationState> history = new TreeMap<>(); + history.put(0L, timedOutState); + history.put(1L, scheduledState); + application.setStatus(new ApplicationStatus( + scheduledState, history, + new ApplicationAttemptSummary(), new ApplicationAttemptSummary())); + when(mockContext.getResource()).thenReturn(application); + + ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder); + + Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), progress); + verify(mockContext, never()).getDriverPodSpec(); + verify(mockContext, never()).getClient(); + verifyNoInteractions(recorder); + Assertions.assertEquals( + ApplicationStateSummary.ScheduledToRestart, + application.getStatus().getCurrentState().getCurrentStateSummary()); + } + + @Test + void unsuspendedAppRequestsDriverOnNextReconcile() { + AppInitStep appInitStep = new AppInitStep(); + SparkAppContext mockContext = mock(SparkAppContext.class); + SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class); + SparkApplication application = new SparkApplication(); + application.setMetadata(applicationMetadata); + application.getSpec().setSuspend(true); + when(mockContext.getResource()).thenReturn(application); + when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of()); + when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec); + when(mockContext.getDriverResourcesSpec()).thenReturn(List.of()); + when(mockContext.getClient()).thenReturn(kubernetesClient); + when(recorder.persistStatus(any(), any())).thenAnswer(invocation -> { + ApplicationStatus newStatus = invocation.getArgument(1); + application.setStatus(newStatus); + return true; + }); + + // Suspended: nothing is created and the app stays Submitted + ReconcileProgress progress1 = appInitStep.reconcile(mockContext, recorder); + Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), progress1); + Assertions.assertNull( + kubernetesClient.pods().inNamespace("default").withName("driver-pod").get()); + Assertions.assertEquals( + ApplicationStateSummary.Submitted, + application.getStatus().getCurrentState().getCurrentStateSummary()); + + // Unsuspended: the regular init path requests the driver + application.getSpec().setSuspend(false); + ReconcileProgress progress2 = appInitStep.reconcile(mockContext, recorder); + Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(), progress2); + Assertions.assertNotNull( + kubernetesClient.pods().inNamespace("default").withName("driver-pod").get()); + Assertions.assertEquals( + ApplicationStateSummary.DriverRequested, + application.getStatus().getCurrentState().getCurrentStateSummary()); + } Review Comment: **Finding 3.** Worth an app-side twin of `ClusterInitStepTest.nonInitializingClusterProceeds`: a `RunningHealthy` app with `suspend: true` must still return `proceed()`. That is the boundary finding 2 is about, and right now it is pinned only for `SparkCluster`. I ran this against the PR head and it passes: ```java @Test void suspendedNonInitializingAppProceeds() { AppInitStep appInitStep = new AppInitStep(); SparkAppContext mockContext = mock(SparkAppContext.class); SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class); SparkApplication application = new SparkApplication(); application.setMetadata(applicationMetadata); application.getSpec().setSuspend(true); application.setStatus( application .getStatus() .appendNewState( new ApplicationState(ApplicationStateSummary.RunningHealthy, "running"))); when(mockContext.getResource()).thenReturn(application); Assertions.assertEquals( ReconcileProgress.proceed(), appInitStep.reconcile(mockContext, recorder)); verifyNoInteractions(recorder); } ``` ########## tests/e2e/suspend/spark-application-state-transition.yaml: ########## @@ -0,0 +1,34 @@ +# +# 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: false +status: + stateTransitionHistory: Review Comment: **Finding 5.** This is `tests/e2e/assertions/spark-application/spark-state-transition.yaml` with a different resource name, and `spark-cluster-state-transition.yaml` is the cluster one. AGENTS.md puts shared assertions in `tests/e2e/assertions/`. Binding the name there (`name: ($SPARK_APPLICATION_NAME)`, as the namespace already is) would let this group reuse both files the way `state-transition/chainsaw-test.yaml` does, leaving only the two `*-suspended.yaml` files here. The extra `spec.suspend: false` check can move into an inline `resource:` assert. -- 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]
