dongjoon-hyun commented on code in PR #843:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/843#discussion_r4047767928
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java:
##########
@@ -100,6 +103,20 @@ public ReconcileProgress reconcile(
}
}
try {
+ // Like the suspend hold, a driver requested before must not be left
unobserved.
+ if (KueueWorkloadFactory.hasQueueName(app) &&
!isDriverRequested(context)) {
Review Comment:
Thank you for the reproduction. I applied the smallest fix in 34e083c. When
`spec.suspend` holds a queued resource, `AppInitStep` and `ClusterInitStep` now
delete its `Workload` via `KueueWorkloadUtils.releaseWorkload`, so it no longer
keeps the quota. The resource is queued again when it is resumed.
- `suspendingQueuedAppReleasesKueueWorkload` and
`suspendingQueuedClusterReleasesKueueWorkload` cover it.
- `suspendedAppWithQueueNameDoesNotCreateKueueWorkload` now stubs the client
instead of asserting `never().getClient()`.
- The `Kueue` section of the docs describes the new behavior.
I'd like to leave the `spec.active` approach, which keeps the queue position
across a suspend, as a follow-up. Until then, `.active(!suspend)` in
`KueueWorkloadFactory` remains unreachable.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java:
##########
@@ -598,4 +612,182 @@ void staleInformerSnapshotDoesNotBypassSuspend() {
ApplicationStateSummary.ScheduledToRestart,
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+
+ @Test
+ void kueueWorkloadIsCreatedAndDriverIsHeldUntilAdmitted() {
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ Workload workload = getWorkload();
+ Assertions.assertNotNull(workload);
+ Assertions.assertEquals("test-queue", workload.getSpec().getQueueName());
+ Assertions.assertTrue(workload.getSpec().getActive());
+ verify(mockContext, never()).getDriverPodSpec();
+ verifyNoInteractions(recorder);
+ Assertions.assertEquals(
+ ApplicationStateSummary.Submitted,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
+ @Test
+ void admittedKueueWorkloadRequestsDriver() {
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+ when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec);
+ when(mockContext.getDriverResourcesSpec()).thenReturn(List.of());
+ when(recorder.persistStatus(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ application.setStatus(invocation.getArgument(1));
+ return true;
+ });
+
+ // Not admitted yet: the driver is not requested
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ appInitStep.reconcile(mockContext, recorder));
+ Assertions.assertNull(
+
kubernetesClient.pods().inNamespace("default").withName("driver-pod").get());
+
+ admitWorkload();
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ appInitStep.reconcile(mockContext, recorder));
+ Assertions.assertNotNull(
+
kubernetesClient.pods().inNamespace("default").withName("driver-pod").get());
+ Assertions.assertEquals(
+ ApplicationStateSummary.DriverRequested,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
+ @Test
+ void suspendedAppWithQueueNameDoesNotCreateKueueWorkload() {
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+ application.getSpec().setSuspend(true);
+ when(mockContext.getResource()).thenReturn(application);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ Assertions.assertNull(getWorkload());
+ verify(mockContext, never()).getClient();
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void staleKueueWorkloadIsDeletedBeforeRequestingAdmission() {
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+ // A Workload of a deleted application that had the same name is not
garbage collected yet
+ Workload stale = KueueWorkloadFactory.buildWorkload(application);
+ stale.getMetadata().getOwnerReferences().get(0).setUid("stale-uid");
+ kubernetesClient.resource(stale).create();
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndRequeueAfter(
+ KueueWorkloadUtils.STALE_WORKLOAD_REQUEUE_INTERVAL),
+ progress);
+ Assertions.assertNull(getWorkload());
+ verify(mockContext, never()).getDriverPodSpec();
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void unsupportedKueueSpecFailsScheduling() {
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(kueueApplicationMetadata);
+
application.getSpec().getSparkConf().put("spark.dynamicAllocation.enabled",
"true");
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndImmediateRequeue(),
progress);
+ Assertions.assertNull(getWorkload());
+ ArgumentCaptor<ApplicationStatus> captor =
ArgumentCaptor.forClass(ApplicationStatus.class);
+ verify(recorder).persistStatus(any(), captor.capture());
+ Assertions.assertEquals(
+ ApplicationStateSummary.SchedulingFailure,
+ captor.getValue().getCurrentState().getCurrentStateSummary());
Review Comment:
Good catch. I added the message assertion as you suggested in 34e083c.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java:
##########
@@ -77,6 +78,19 @@ public ReconcileProgress reconcile(
}
}
try {
+ // Like the suspend hold, a master requested before must complete its
initialization.
+ if (KueueWorkloadFactory.hasQueueName(cluster) &&
!isMasterRequested(context)) {
+ AdmissionResult admission =
+ KueueWorkloadUtils.requestAdmission(
+ context.getClient(),
KueueWorkloadFactory.buildWorkload(cluster));
Review Comment:
Agreed. In 34e083c, `buildWorkload` stays inside the `try`, so an
unsupported spec still ends up in `SchedulingFailure`. An
`IllegalStateException` or `KubernetesClientException` from `requestAdmission`
is logged and retried after `STALE_WORKLOAD_REQUEUE_INTERVAL` instead. I
applied the same to `AppInitStep`, and `kueueApiFailureIsRetried` in both step
tests covers it.
For now, the retry is visible only in the operator log. I'd like to surface
it as a Kubernetes `Warning` event in a follow-up.
##########
docs/spark_custom_resources.md:
##########
@@ -557,9 +557,52 @@ spec:
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`.
- The operator does not integrate with such a system yet.
+* This is the building block for external job queueing systems. See
[Kueue](#kueue) for the
+ built-in integration.
+
+## Kueue
+
+A `SparkApplication` or a `SparkCluster` labeled with
`kueue.x-k8s.io/queue-name` is queued by
+[Kueue](https://kueue.sigs.k8s.io/). The operator creates a Kueue `Workload`
that describes the
+driver and executor (or master and worker) pod sets, and holds the creation of
those resources
+until Kueue admits the `Workload`.
+
+```yaml
+apiVersion: spark.apache.org/v1
+kind: SparkApplication
+metadata:
+ name: spark-pi
+ labels:
+ kueue.x-k8s.io/queue-name: spark-queue
+spec:
+ mainClass: "org.apache.spark.examples.SparkPi"
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ runtimeVersions:
+ sparkVersion: "4.2.0"
+```
+
+* The label alone enables the integration. Kueue and its `LocalQueue` must
exist, and the operator
+ needs the Kueue RBAC rules which the Helm chart grants when
`operatorRbac.kueue.enabled` is set.
+ See [Optional Prerequisites](operations.md#optional-prerequisites).
+* The `Workload` is named `<lower-cased kind>-<resource name>` and is owned by
the Spark resource,
+ so it is garbage collected along with it.
+* While the `Workload` waits for quota, the resource stays in its initializing
state (`Submitted`,
+ or `ScheduledToRestart` for a restarted attempt) and no driver (or master /
worker) is created.
+ Like `spec.suspend`, the initial `Submitted` status of the first attempt is
not persisted to the
+ API server, so `kubectl get` shows an empty `Current State` and no events
are published until the
+ `Workload` is admitted. Use `kubectl get workload` to see the admission
status. If the spec
+ changes while waiting, the `Workload` is recreated with the new resource
requests.
+* When a `SparkApplication` attempt stops and its resources are released, the
operator deletes the
+ `Workload` so that Kueue releases the quota. A restarted attempt is queued
again. Resources
+ retained by `resourceRetainPolicy` keep the quota until they are released.
+* A `SparkCluster` requests the resources set on the `master` and `worker`
containers of its pod
+ templates. A missing request defaults to the limit, or else to 1 CPU and
`SPARK_DAEMON_MEMORY`
+ plus overhead. A worker uses `SPARK_WORKER_CORES` for the CPU and adds
`SPARK_WORKER_MEMORY` to
+ the memory when they are set. A `SparkCluster` keeps the quota until it is
deleted.
+* `spec.suspend` takes precedence. A suspended resource does not get a
`Workload` at all.
+* Dynamic allocation, a `SparkCluster` with `minWorkers < maxWorkers`, and pod
template files set
+ through `spark.kubernetes.{driver,executor}.podTemplateFile` are not
supported yet. Such a
+ resource fails with `SchedulingFailure` instead of being queued.
Review Comment:
Added your suggested bullet to the `Kueue` section in 34e083c. Honoring an
eviction needs an observer beyond the init steps, so I'll handle it in a
follow-up.
##########
docs/spark_custom_resources.md:
##########
@@ -557,9 +557,52 @@ spec:
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`.
- The operator does not integrate with such a system yet.
+* This is the building block for external job queueing systems. See
[Kueue](#kueue) for the
+ built-in integration.
+
+## Kueue
+
+A `SparkApplication` or a `SparkCluster` labeled with
`kueue.x-k8s.io/queue-name` is queued by
+[Kueue](https://kueue.sigs.k8s.io/). The operator creates a Kueue `Workload`
that describes the
+driver and executor (or master and worker) pod sets, and holds the creation of
those resources
+until Kueue admits the `Workload`.
+
+```yaml
+apiVersion: spark.apache.org/v1
+kind: SparkApplication
+metadata:
+ name: spark-pi
+ labels:
+ kueue.x-k8s.io/queue-name: spark-queue
+spec:
+ mainClass: "org.apache.spark.examples.SparkPi"
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ runtimeVersions:
+ sparkVersion: "4.2.0"
+```
+
+* The label alone enables the integration. Kueue and its `LocalQueue` must
exist, and the operator
+ needs the Kueue RBAC rules which the Helm chart grants when
`operatorRbac.kueue.enabled` is set.
+ See [Optional Prerequisites](operations.md#optional-prerequisites).
+* The `Workload` is named `<lower-cased kind>-<resource name>` and is owned by
the Spark resource,
+ so it is garbage collected along with it.
+* While the `Workload` waits for quota, the resource stays in its initializing
state (`Submitted`,
+ or `ScheduledToRestart` for a restarted attempt) and no driver (or master /
worker) is created.
+ Like `spec.suspend`, the initial `Submitted` status of the first attempt is
not persisted to the
+ API server, so `kubectl get` shows an empty `Current State` and no events
are published until the
+ `Workload` is admitted. Use `kubectl get workload` to see the admission
status. If the spec
+ changes while waiting, the `Workload` is recreated with the new resource
requests.
+* When a `SparkApplication` attempt stops and its resources are released, the
operator deletes the
+ `Workload` so that Kueue releases the quota. A restarted attempt is queued
again. Resources
+ retained by `resourceRetainPolicy` keep the quota until they are released.
+* A `SparkCluster` requests the resources set on the `master` and `worker`
containers of its pod
+ templates. A missing request defaults to the limit, or else to 1 CPU and
`SPARK_DAEMON_MEMORY`
+ plus overhead. A worker uses `SPARK_WORKER_CORES` for the CPU and adds
`SPARK_WORKER_MEMORY` to
+ the memory when they are set. A `SparkCluster` keeps the quota until it is
deleted.
Review Comment:
Added the guidance to the `SparkCluster` bullet in 34e083c, based on your
suggestion.
##########
tests/e2e/kueue/spark-example-queued.yaml:
##########
@@ -0,0 +1,37 @@
+#
+# 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-kueue-queued-test
+ namespace: default
+ labels:
+ kueue.x-k8s.io/queue-name: ($LOCAL_QUEUE)
+spec:
+ mainClass: "org.apache.spark.examples.SparkPi"
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ # The Kueue admission takes one more reconciliation, which uses up the
per-resource rate limit
+ # (5 reconciliations per 15 seconds) before the driver starts. Run longer
than that window so
+ # that the driver transitions are observed after the rate limit is refreshed.
+ driverArgs: ["10000"]
Review Comment:
Thank you for checking that the filters do not affect the informer cache. In
34e083c, I added a bullet about the extra reconciliation to the `Kueue` section
and reworded the comment here to say that the test avoids the window. The
underlying rate limit issue is not specific to Kueue, so I'm looking into it
separately.
--
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]
