peter-toth commented on code in PR #851:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/851#discussion_r4052922179
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -124,9 +140,100 @@ public static AdmissionResult requestAdmission(
deleteWorkload(client, workload);
return AdmissionResult.STALE;
}
+ WorkloadSpec spec = workload.getSpec();
+ if (desired.getSpec().getPriority() != null
Review Comment:
**Finding 1.** This clause is what stops a permission blip from wiping a
pending Workload's priority, and nothing pins it.
I deleted just `desired.getSpec().getPriority() != null` and ran the whole
`spark-operator` suite: **it stays green**. Then, with a Workload already
carrying `WorkloadPriorityClass high` (priority 1000) and the priority-class
reads failing 403 on the next reconcile, the stored Workload came back
`priority=null, priorityClassRef=null`. So the operator would hand it back to
Kueue at the default priority 0 because it briefly could not read the class.
The CEL rules allow that write, since presence is only frozen once quota is
reserved, so nothing downstream catches it either.
This test fails on that mutant and passes as written:
```java
@Test
void pendingWorkloadKeepsItsPriorityWhenThePermissionIsLost() {
createWorkloadPriorityClass("high", 1000);
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workloadWithPriorityClass("high"));
Assertions.assertEquals(1000, getWorkload().getSpec().getPriority());
// The operator loses the RBAC rules while the Workload waits for quota
KubernetesClient forbiddenClient =
mock(KubernetesClient.class,
withSettings().defaultAnswer(delegatesTo(kubernetesClient)));
KubernetesClientException forbidden = new
KubernetesClientException("forbidden", 403, null);
doThrow(forbidden).when(forbiddenClient).resources(WorkloadPriorityClass.class);
doThrow(forbidden).when(forbiddenClient).scheduling();
Assertions.assertEquals(
AdmissionResult.PENDING,
KueueWorkloadUtils.requestAdmission(forbiddenClient,
workloadWithPriorityClass("high")));
Assertions.assertEquals(1000, getWorkload().getSpec().getPriority());
Assertions.assertEquals("high",
getWorkload().getSpec().getPriorityClassRef().getName());
}
```
It needs `org.mockito.AdditionalAnswers.delegatesTo`, `Mockito.doThrow` and
`Mockito.withSettings`. `priorityIsNotSetWithoutPermission` covers the same 403
but only through `setPriority`, so it never reaches this branch.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -76,16 +90,18 @@ public enum AdmissionResult {
* 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. The
pod sets hash annotation
- * is added to it in place.
+ * @param desired The Workload built for the resource to be admitted. The
priority and the pod
+ * sets hash annotation are added to it in place.
* @return The AdmissionResult for the Workload.
- * @throws IllegalStateException if the Workload can neither be read nor
created.
+ * @throws IllegalStateException if the Workload can neither be read nor
created, or if its
+ * priority class does not exist.
* @throws KubernetesClientException if a stale Workload cannot be deleted.
Unlike {@link
* #releaseWorkload}, this is not swallowed so that the resource is not
created until the
* stale Workload is gone.
*/
public static AdmissionResult requestAdmission(
final KubernetesClient client, final Workload desired) {
+ setPriority(client, desired);
Review Comment:
**Finding 2.** Resolving here means every reconcile of a queued resource
pays for it, including the ones where nothing changed.
I counted the mock-server requests for a steady-state `PENDING` reconcile
with no priority class anywhere: **2 with this line, 1 without it**. The extra
one is a cluster-wide `priorityclasses` LIST, because `getPodPriorityClass`
falls through to the `globalDefault` lookup, and that is the default shape for
a resource with no priority class at all. On the `STALE` path it repeats every
5 s.
Kueue's own integration gates exactly this, in
`jobframework.updateWorkloadPriorities`:
```go
// Resolving only when at least one workload needs a transition keeps
// steady-state reconciles from re-resolving and overwriting the
// otherwise-mutable priority value.
if len(needsClassChange) == 0 {
return nil
}
```
`classifyWorkloadsForPriorityUpdate` decides that by comparing class
*names*, which costs nothing. The same gate is available here:
`KueueWorkloadFactory` copies the owner's labels onto the Workload
(`spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:167`),
so the stored Workload carries the same `kueue.x-k8s.io/priority-class` label,
and its `priorityClassRef.name` is already in hand. So resolve only when the
Workload has to be created, or when the existing `priorityClassRef` name
differs from the label. That is also exactly the documented behaviour that "a
changed value of the same class does not affect the existing `Workload`".
It does need the read to happen before the create, i.e. splitting
`getOrCreateSecondaryResource` into a get and a create in this method.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -56,6 +65,11 @@ public final class KueueWorkloadUtils {
*/
public static final Duration STALE_WORKLOAD_REQUEUE_INTERVAL =
Duration.ofSeconds(5);
+ /** Like Kueue, the priority of a Workload without any priority class. */
+ private static final int DEFAULT_PRIORITY = 0;
+
+ private static final int HTTP_FORBIDDEN = 403;
Review Comment:
**Finding 5.** The JDK has this one, and the module static-imports it
elsewhere: `ReconcilerUtils.java:27` for `HTTP_NOT_FOUND`,
`SparkExceptionUtils.java:22` and `StatusRecorder.java:22` for `HTTP_CONFLICT`,
and `ReadinessProbe.java:22` already pulls in `HTTP_FORBIDDEN` that way.
`Constants.HTTP_TOO_MANY_REQUESTS` exists only because 429 has no JDK constant.
```java
import static java.net.HttpURLConnection.HTTP_FORBIDDEN;
```
and drop the field.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -124,9 +140,100 @@ public static AdmissionResult requestAdmission(
deleteWorkload(client, workload);
return AdmissionResult.STALE;
}
+ WorkloadSpec spec = workload.getSpec();
+ if (desired.getSpec().getPriority() != null
+ && !Objects.equals(spec.getPriorityClassRef(),
desired.getSpec().getPriorityClassRef())
+ && (workload.getStatus() == null ||
!workload.getStatus().isQuotaReserved())) {
+ // Like Kueue, a changed priority class is applied in place so that the
Workload keeps its
+ // position in the queue. Kueue does not allow the change once the quota
is reserved.
+ log.info(
+ "Updating the priority class of the pending Kueue Workload {}.",
+ workload.getMetadata().getName());
+ spec.setPriorityClassRef(desired.getSpec().getPriorityClassRef());
+ spec.setPriority(desired.getSpec().getPriority());
+ client.resource(workload).update();
+ }
return AdmissionResult.PENDING;
}
+ /**
+ * Sets the priority of the desired Workload in the same way as Kueue
built-in integrations. The
+ * WorkloadPriorityClass of the `kueue.x-k8s.io/priority-class` label takes
precedence over the
+ * PriorityClass of the first pod set which has one, and the global default
PriorityClass is used
+ * without both. Without any of them, the priority is 0 without a priority
class. The priority is
+ * left unset if the operator is not allowed to read the cluster-scoped
priority classes.
+ *
+ * @param client The KubernetesClient.
+ * @param workload The Workload whose priority is set in place.
+ * @throws IllegalStateException if the priority class does not exist.
+ */
+ static void setPriority(final KubernetesClient client, final Workload
workload) {
+ WorkloadSpec spec = workload.getSpec();
+ try {
+ Map<String, String> labels = workload.getMetadata().getLabels();
+ String workloadPriorityClassName =
+ labels == null ? null :
labels.get(Constants.LABEL_WORKLOAD_PRIORITY_CLASS);
+ if (StringUtils.isNotEmpty(workloadPriorityClassName)) {
+ WorkloadPriorityClass workloadPriorityClass =
+
client.resources(WorkloadPriorityClass.class).withName(workloadPriorityClassName).get();
+ if (workloadPriorityClass == null) {
+ throw new IllegalStateException(
+ "Kueue WorkloadPriorityClass " + workloadPriorityClassName + "
is not found.");
Review Comment:
**Finding 4.** A typo in the label leaves the resource stuck with nothing
for the user to look at.
`AppInitStep.holdForKueueAdmission` catches `IllegalStateException`, logs a
warning and requeues after 5 s. The first attempt's `Submitted` status is never
persisted, so `kubectl get` shows an empty `Current State` and `kubectl
describe` shows nothing. The result is a resource that retries forever, every 5
s, with the reason only in the operator log, for a plain user error that will
not fix itself.
`docs/spark_custom_resources.md` is honest about it ("The resource keeps
waiting without a `Workload` while the named priority class does not exist"),
but that does not help someone working out why their application never starts.
The pod `PriorityClass` case at line 226 is the worse half, because it is
unambiguously permanent: without Kueue the same spec reaches the driver pod and
the Priority admission plugin rejects it, so the user gets a
`SchedulingFailure` naming the class. Throwing `UnsupportedOperationException`
there would restore that. `KueueWorkloadFactory.buildWorkload` already uses it
for an unsupported spec and the init steps turn it into `SchedulingFailure`.
For the `WorkloadPriorityClass` case retrying is the right outcome, since an
admin may still create the class, so that one needs a signal rather than a
different outcome.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -124,9 +140,100 @@ public static AdmissionResult requestAdmission(
deleteWorkload(client, workload);
return AdmissionResult.STALE;
}
+ WorkloadSpec spec = workload.getSpec();
+ if (desired.getSpec().getPriority() != null
+ && !Objects.equals(spec.getPriorityClassRef(),
desired.getSpec().getPriorityClassRef())
+ && (workload.getStatus() == null ||
!workload.getStatus().isQuotaReserved())) {
+ // Like Kueue, a changed priority class is applied in place so that the
Workload keeps its
+ // position in the queue. Kueue does not allow the change once the quota
is reserved.
Review Comment:
**Finding 3.** "Kueue does not allow the change once the quota is reserved"
is narrower than that, and the guard above is correspondingly stricter than
Kueue.
The Workload CRD freezes four things while `QuotaReserved` is true
(`apis/kueue/v1beta2/workload_types.go:1191-1194`): the *presence* of
`priorityClassRef`, its `group`, its `kind`, and its `name` **only** for
`scheduling.k8s.io/PriorityClass`. The `name` of a `WorkloadPriorityClass` and
the `priority` value both stay mutable, and Kueue relies on that:
`hasSameOrEmptyPriorityClass` returns true for two `WorkloadPriorityClass` refs
with different names, so `updateWorkloadPriorities` goes ahead and writes.
The part of the guard that matters is real. A Workload that reserved quota
with no ref can never be given one, which is the first skip in
`classifyWorkloadsForPriorityUpdate`. What it also blocks is raising the
`WorkloadPriorityClass` on a Workload waiting behind admission checks, which
can take minutes with a `ProvisioningRequest`, and that is the case the label
exists for.
Mirroring `hasSameOrEmptyPriorityClass` keeps the safe part:
```java
if (desired.getSpec().getPriority() != null
&& !Objects.equals(spec.getPriorityClassRef(),
desired.getSpec().getPriorityClassRef())
&& (workload.getStatus() == null
|| !workload.getStatus().isQuotaReserved()
|| isSameGroupAndKind(
spec.getPriorityClassRef(),
desired.getSpec().getPriorityClassRef()))) {
```
where the helper requires both refs present with equal `group` and `kind`,
and equal `name` too when the group is `scheduling.k8s.io`.
`quotaReservedWorkloadKeepsPriorityClass` then needs its two classes to differ
in group or kind (or to go from a ref to none) so that it keeps testing the
case that really is frozen.
--
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]
