peter-toth commented on code in PR #835: URL: https://github.com/apache/spark-kubernetes-operator/pull/835#discussion_r4029146874
########## spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtilsTest.java: ########## @@ -0,0 +1,212 @@ +/* + * 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.kueue; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.fabric8.kubernetes.api.model.ConditionBuilder; +import io.fabric8.kubernetes.api.model.KubernetesResourceList; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.dsl.MixedOperation; +import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation; +import io.fabric8.kubernetes.client.dsl.Resource; +import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.spark.k8s.operator.SparkApplication; +import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils.AdmissionResult; +import org.apache.spark.k8s.operator.kueue.v1beta2.PodSet; +import org.apache.spark.k8s.operator.kueue.v1beta2.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadSpec; +import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus; + +@EnableKubernetesMockClient(crud = true) +@SuppressFBWarnings( + value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD"}, + justification = "Unwritten fields are covered by Kubernetes mock client") +class KueueWorkloadUtilsTest { + private static final String NAME = "sparkapplication-app-1"; + + private KubernetesClient kubernetesClient; + + @Test + void newWorkloadIsCreatedAndPendingUntilAdmitted() { + Workload desired = workload("owner-uid-1", 1); + + Assertions.assertEquals( + AdmissionResult.PENDING, KueueWorkloadUtils.requestAdmission(kubernetesClient, desired)); + Workload created = getWorkload(); + Assertions.assertNotNull(created); + Assertions.assertEquals("test-queue", created.getSpec().getQueueName()); + Assertions.assertEquals( + KueueWorkloadUtils.hashPodSets(desired), + created.getMetadata().getAnnotations().get(KueueWorkloadUtils.ANNOTATION_POD_SETS_HASH)); + + // A second reconciliation reuses the same Workload and still waits for the admission + Assertions.assertEquals( + AdmissionResult.PENDING, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1))); + Assertions.assertEquals( + created.getMetadata().getUid(), getWorkload().getMetadata().getUid()); + } + + @Test + void admittedWorkloadIsReported() { + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1)); + admitWorkload(); + + Assertions.assertEquals( + AdmissionResult.ADMITTED, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1))); + } + + @Test + void admittedWorkloadIsKeptEvenIfPodSetsChanged() { + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1)); + admitWorkload(); + + Assertions.assertEquals( + AdmissionResult.ADMITTED, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 5))); + Assertions.assertNotNull(getWorkload()); + } + + @Test + void pendingWorkloadWithOutdatedPodSetsIsRecreated() { + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1)); + + // The spec changed while waiting for the admission + Assertions.assertEquals( + AdmissionResult.STALE, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 5))); + Assertions.assertNull(getWorkload()); + + Assertions.assertEquals( + AdmissionResult.PENDING, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 5))); + Assertions.assertEquals(5, getWorkload().getSpec().getPodSets().get(0).getCount()); + } + + @Test + void workloadOwnedByAnotherResourceIsDeleted() { + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("stale-owner-uid", 1)); + admitWorkload(); + + Assertions.assertEquals( + AdmissionResult.STALE, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1))); + Assertions.assertNull(getWorkload()); + } + + @Test + void terminatingWorkloadIsNotUsed() { + Workload terminating = workload("owner-uid-1", 1); + terminating.getMetadata().setFinalizers(List.of("kueue.x-k8s.io/resource-in-use")); + KueueWorkloadUtils.requestAdmission(kubernetesClient, terminating); + admitWorkload(); + kubernetesClient.resources(Workload.class).inNamespace("default").withName(NAME).delete(); + Assertions.assertNotNull(getWorkload().getMetadata().getDeletionTimestamp()); + + Assertions.assertEquals( + AdmissionResult.STALE, + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1))); + } + + @Test + void releaseWorkloadDeletesWorkload() { + KueueWorkloadUtils.requestAdmission(kubernetesClient, workload("owner-uid-1", 1)); + + KueueWorkloadUtils.releaseWorkload(kubernetesClient, owner()); + + Assertions.assertNull(getWorkload()); + // Releasing again is a no-op + KueueWorkloadUtils.releaseWorkload(kubernetesClient, owner()); + } + + @Test + @SuppressWarnings("unchecked") + void releaseWorkloadIgnoresFailures() { + KubernetesClient client = mock(KubernetesClient.class); + MixedOperation<Workload, KubernetesResourceList<Workload>, Resource<Workload>> operation = + mock(MixedOperation.class); + NonNamespaceOperation<Workload, KubernetesResourceList<Workload>, Resource<Workload>> + namespaced = mock(NonNamespaceOperation.class); + Resource<Workload> resource = mock(Resource.class); + when(client.resources(Workload.class)).thenReturn(operation); + when(operation.inNamespace("default")).thenReturn(namespaced); + when(namespaced.withName(NAME)).thenReturn(resource); + when(resource.delete()).thenThrow(new KubernetesClientException("forbidden", 403, null)); + + Assertions.assertDoesNotThrow(() -> KueueWorkloadUtils.releaseWorkload(client, owner())); + } + + private Workload getWorkload() { + return kubernetesClient.resources(Workload.class).inNamespace("default").withName(NAME).get(); + } + + private void admitWorkload() { + Workload workload = getWorkload(); + workload.setStatus( + WorkloadStatus.builder() + .conditions( + List.of(new ConditionBuilder().withType("Admitted").withStatus("True").build())) + .build()); + kubernetesClient.resource(workload).update(); + } + + private static SparkApplication owner() { + SparkApplication app = new SparkApplication(); + app.setMetadata(new ObjectMetaBuilder().withName("app-1").withNamespace("default").build()); + return app; + } + + private static Workload workload(final String ownerUid, final int executors) { + Workload workload = new Workload(); + workload.setMetadata( + new ObjectMetaBuilder() + .withName(NAME) + .withNamespace("default") + .withLabels(Map.of("spark.operator/spark-app-name", "app-1")) + .withOwnerReferences( + new OwnerReferenceBuilder() + .withName("app-1") + .withKind("SparkApplication") + .withUid(ownerUid) + .withController(true) + .build()) + .build()); + workload.setSpec( + WorkloadSpec.builder() + .queueName("test-queue") + .active(true) + .podSets(List.of(PodSet.builder().name("executor").count(executors).build())) Review Comment: **Finding 1.** No test gives a `PodSet` a `template`, so `hashPodSets` is only ever exercised on `count`. In production every pod set carries a full `PodTemplateSpec` from `buildPodSet`, and that is where everything non-trivial lives - containers, resource maps, and the node selector that `KueueWorkloadFactory.java:443` builds into a `new HashMap<>()`. That matters because "the same spec hashes the same on every reconcile" is what keeps the `STALE` branch from livelocking. If it ever stopped holding, a pending Workload would be found outdated, deleted, recreated, found outdated again, once per `STALE_WORKLOAD_REQUEUE_INTERVAL`, and never admitted. Nothing in the suite would notice. I checked that the invariant does hold today, so this is a coverage gap rather than a bug. The probe was: ```java @Test void hashIsStableAcrossIdenticalBuilds() { assertEquals( KueueWorkloadUtils.hashPodSets(KueueWorkloadFactory.buildWorkload(probeApp())), KueueWorkloadUtils.hashPodSets(KueueWorkloadFactory.buildWorkload(probeApp()))); } ``` with `probeApp()` setting `spark.executor.instances` plus twelve `spark.kubernetes.node.selector.*` entries, so both builds go through the `HashMap` node selector. It passes. Something along those lines in `KueueWorkloadFactoryTest`, where the app builders already live, would pin it. While there, the `.with(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)` on the writer is also unpinned - I removed it and all 34 tests under `org.apache.spark.k8s.operator.kueue` stayed green, including the probe above. It is doing no work today because both sides of the comparison come from the same factory path, but it is the line that would matter if a map ever reached the hash from a different source. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java: ########## @@ -0,0 +1,182 @@ +/* + * 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.kueue; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Duration; +import java.util.HashMap; +import java.util.HexFormat; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.SerializationFeature; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; +import lombok.extern.slf4j.Slf4j; + +import org.apache.spark.k8s.operator.kueue.v1beta2.Workload; +import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus; +import org.apache.spark.k8s.operator.utils.ModelUtils; +import org.apache.spark.k8s.operator.utils.ReconcilerUtils; + +/** Utilities to create, check and release Kueue Workloads. */ +@Slf4j +public final class KueueWorkloadUtils { + + /** Annotation holding the hash of the pod sets which the Workload was created with. */ + public static final String ANNOTATION_POD_SETS_HASH = "spark.operator/kueue-pod-sets-hash"; + + /** + * Requeue interval after {@link AdmissionResult#STALE}. It is short because the stale Workload + * goes away shortly, while an unchanged admission is watched with the default interval. + */ + public static final Duration STALE_WORKLOAD_REQUEUE_INTERVAL = Duration.ofSeconds(5); + + private KueueWorkloadUtils() {} + + /** Outcome of {@link #requestAdmission(KubernetesClient, Workload)}. */ + public enum AdmissionResult { + /** Kueue admitted the Workload, so the requested resources can be created. */ + ADMITTED, + /** The Workload waits for quota, so the resource creation is held. */ + PENDING, + /** + * The existing Workload cannot be used because it is owned by another resource, requests + * outdated pod sets, or is being deleted. It is deleted so that a later reconciliation creates + * the Workload of the current spec. + */ + STALE + } + + /** + * Creates the given Workload if it does not exist yet and reports whether Kueue admitted it. + * + * @param client The KubernetesClient. + * @param desired The Workload built for the resource to be admitted. + * @return The AdmissionResult for the Workload. Review Comment: **Finding 2.** The class exists to be called from `AppInitStep`, `ClusterInitStep` and `AppCleanUpStep`, so its throws contract is the part those callers have to design around, and right now it is only in the body: - `:93` throws `IllegalStateException` when `getOrCreateSecondaryResource` returns empty. - `:148` `deleteWorkload` does not catch, so the two `STALE` branches can throw `KubernetesClientException`. That is a deliberate difference from `releaseWorkload`, which swallows and explains why in its own javadoc, but nothing says so here. ```java * @param desired The Workload built for the resource to be admitted. * @return The AdmissionResult for the Workload. * @throws IllegalStateException if the Workload can neither be read nor created. * @throws io.fabric8.kubernetes.client.KubernetesClientException if a stale Workload cannot be * deleted. Unlike {@link #releaseWorkload}, this is not swallowed: the resource must not be * created while a Workload of the wrong shape still holds the quota. */ ``` Adjust the reasoning if I have the intent backwards - either way it is worth stating, since the difference between the two delete paths is the kind of thing a caller gets wrong once. The `desired` argument is also mutated in place at `:70` (the hash annotation is added to the caller's object). Harmless with the current factory, which builds a fresh Workload per call, but worth a `@param` note if this stays public. -- 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]
