peter-toth commented on code in PR #851:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/851#discussion_r4057064988
##########
docs/spark_custom_resources.md:
##########
@@ -586,6 +586,13 @@ spec:
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.
+* Like Kueue built-in integrations, the `Workload` gets the priority of the
`WorkloadPriorityClass`
+ named by the `kueue.x-k8s.io/priority-class` label. Without the label, the
`priorityClassName` of
+ the driver (or master) pod template is used, then the one of the executor
(or worker) pod
+ template, and then the `globalDefault` `PriorityClass`. The resource keeps
waiting without a
+ `Workload` while the named priority class does not exist. Changing the label
before the
Review Comment:
**Finding 8.** "before the `Workload` is admitted" is now broader than what
`isPriorityClassChangeAllowed` permits.
Once quota is reserved but before admission, two label changes are silently
dropped by the presence and group/kind freeze: adding the label to a `Workload`
that has no ref at all, and adding it to one backed by a `globalDefault`
`PriorityClass`. Both fall inside the window this sentence promises. The
previous "while the `Workload` waits for quota" was accurate, so the new
wording needs the exception spelled out:
```
`Workload` while the named priority class does not exist. Changing the
label before the
`Workload` reserves quota updates its priority in place. After that Kueue
freezes the
presence, group and kind of the priority class, so only the name of a
`WorkloadPriorityClass` still changes. A changed value of the same class
does not affect
the existing `Workload`.
```
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -124,9 +143,121 @@ public static AdmissionResult requestAdmission(
deleteWorkload(client, workload);
return AdmissionResult.STALE;
}
+ WorkloadSpec spec = workload.getSpec();
+ PriorityClassRef desiredPriorityClassRef =
desired.getSpec().getPriorityClassRef();
+ if (desired.getSpec().getPriority() != null
+ && !Objects.equals(spec.getPriorityClassRef(), desiredPriorityClassRef)
+ && isPriorityClassChangeAllowed(workload, desiredPriorityClassRef)) {
+ // Like Kueue, a changed priority class is applied in place so that the
Workload keeps its
+ // position in the queue.
+ log.info(
+ "Updating the priority class of the pending Kueue Workload {}.",
+ workload.getMetadata().getName());
+ spec.setPriorityClassRef(desiredPriorityClassRef);
+ spec.setPriority(desired.getSpec().getPriority());
+ client.resource(workload).update();
+ }
return AdmissionResult.PENDING;
}
+ /**
+ * Checks whether Kueue accepts the priority class change of the given
Workload. Like the Workload
+ * CEL rules, the presence, the group and the kind of the priorityClassRef
are frozen once the
+ * quota is reserved, and so is the name of a Kubernetes PriorityClass. The
name of a
+ * WorkloadPriorityClass stays mutable, which is what Kueue relies on to
raise the priority of a
+ * Workload waiting for its admission checks.
+ */
+ private static boolean isPriorityClassChangeAllowed(
+ final Workload workload, final PriorityClassRef desired) {
+ if (workload.getStatus() == null ||
!workload.getStatus().isQuotaReserved()) {
+ return true;
+ }
+ PriorityClassRef current = workload.getSpec().getPriorityClassRef();
+ return current != null
Review Comment:
**Finding 6.** Three of this helper's four conditions have no test, and
every mutant survives the whole suite.
I removed each in turn and ran `KueueWorkloadUtilsTest`:
| removed | result |
|---|---|
| the `getGroup()` **and** `getKind()` equality checks | suite green |
| `!SCHEDULING_API_GROUP.equals(desired.getGroup())` | suite green |
| `current != null` | suite green |
Group and kind are not separable, since the type's own CEL rule makes `group
== 'scheduling.k8s.io'` imply `kind == 'PriorityClass'`, so either one alone
still blocks. `quotaReservedWorkloadKeepsFrozenPriorityClass` only covers the
presence freeze in the ref-to-none direction.
All three are reachable. These three tests pass as written and kill all
three mutants — I ran both arms:
```java
@Test
void quotaReservedWorkloadKeepsItsPodPriorityClassName() {
createPriorityClass("default-a", 50, true);
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workload("owner-uid-1", 1));
reserveQuota();
// The cluster-wide default changed, which is a PriorityClass name change
kubernetesClient.resource(priorityClass("default-a", 50,
false)).update();
createPriorityClass("default-b", 70, true);
Assertions.assertEquals(
AdmissionResult.PENDING,
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workload("owner-uid-1", 1)));
Assertions.assertEquals("default-a",
getWorkload().getSpec().getPriorityClassRef().getName());
Assertions.assertEquals(50, getWorkload().getSpec().getPriority());
}
@Test
void quotaReservedWorkloadDoesNotSwitchPriorityClassGroup() {
createPriorityClass("default-a", 50, true);
createWorkloadPriorityClass("high", 1000);
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workload("owner-uid-1", 1));
reserveQuota();
// Adding the label would switch the group and the kind of the ref
Assertions.assertEquals(
AdmissionResult.PENDING,
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workloadWithPriorityClass("high")));
Assertions.assertEquals(
"scheduling.k8s.io",
getWorkload().getSpec().getPriorityClassRef().getGroup());
Assertions.assertEquals(50, getWorkload().getSpec().getPriority());
}
@Test
void quotaReservedWorkloadWithoutPriorityClassDoesNotGainOne() {
createWorkloadPriorityClass("high", 1000);
// The Workload is created while the operator cannot read the classes,
so it has no ref
KueueWorkloadUtils.requestAdmission(forbiddenClient(),
workloadWithPriorityClass("high"));
Assertions.assertNull(getWorkload().getSpec().getPriorityClassRef());
reserveQuota();
// The permission is back, but Kueue no longer accepts adding a priority
class
Assertions.assertEquals(
AdmissionResult.PENDING,
KueueWorkloadUtils.requestAdmission(kubernetesClient,
workloadWithPriorityClass("high")));
Assertions.assertNull(getWorkload().getSpec().getPriorityClassRef());
}
```
Two helpers to extract: `priorityClass(name, value, globalDefault)` out of
`createPriorityClass`, and `forbiddenClient()` out of
`pendingWorkloadKeepsItsPriorityWithoutPermission`.
The last one matters most. Without `current != null` it does not merely
allow a bad write, it throws `NullPointerException` out of `requestAdmission`
on `current.getGroup()`. `holdForKueueAdmission` catches only
`IllegalStateException` and `KubernetesClientException`, so the resource ends
up in `SchedulingFailure`.
One note on reachability while writing these: a pod template's
`priorityClassName` change can never reach this helper, because it changes the
pod sets hash and so takes the STALE/recreate path. A `scheduling.k8s.io` ref
only changes name through a `globalDefault` swap, which is what the first test
uses.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -124,9 +143,121 @@ public static AdmissionResult requestAdmission(
deleteWorkload(client, workload);
return AdmissionResult.STALE;
}
+ WorkloadSpec spec = workload.getSpec();
+ PriorityClassRef desiredPriorityClassRef =
desired.getSpec().getPriorityClassRef();
+ if (desired.getSpec().getPriority() != null
+ && !Objects.equals(spec.getPriorityClassRef(), desiredPriorityClassRef)
+ && isPriorityClassChangeAllowed(workload, desiredPriorityClassRef)) {
+ // Like Kueue, a changed priority class is applied in place so that the
Workload keeps its
+ // position in the queue.
+ log.info(
+ "Updating the priority class of the pending Kueue Workload {}.",
+ workload.getMetadata().getName());
+ spec.setPriorityClassRef(desiredPriorityClassRef);
+ spec.setPriority(desired.getSpec().getPriority());
+ client.resource(workload).update();
+ }
return AdmissionResult.PENDING;
}
+ /**
+ * Checks whether Kueue accepts the priority class change of the given
Workload. Like the Workload
+ * CEL rules, the presence, the group and the kind of the priorityClassRef
are frozen once the
+ * quota is reserved, and so is the name of a Kubernetes PriorityClass. The
name of a
+ * WorkloadPriorityClass stays mutable, which is what Kueue relies on to
raise the priority of a
+ * Workload waiting for its admission checks.
+ */
+ private static boolean isPriorityClassChangeAllowed(
+ final Workload workload, final PriorityClassRef desired) {
+ if (workload.getStatus() == null ||
!workload.getStatus().isQuotaReserved()) {
+ return true;
+ }
+ PriorityClassRef current = workload.getSpec().getPriorityClassRef();
+ return current != null
+ && desired != null
+ && Objects.equals(current.getGroup(), desired.getGroup())
+ && Objects.equals(current.getKind(), desired.getKind())
+ && !SCHEDULING_API_GROUP.equals(desired.getGroup());
+ }
+
+ /**
+ * 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.");
+ }
+ spec.setPriorityClassRef(
+ new PriorityClassRef(
+ Constants.KUEUE_API_GROUP, "WorkloadPriorityClass",
workloadPriorityClassName));
+ spec.setPriority(workloadPriorityClass.getValue());
+ return;
+ }
+ PriorityClass priorityClass = getPodPriorityClass(client, spec);
+ if (priorityClass == null) {
+ spec.setPriorityClassRef(null);
+ spec.setPriority(DEFAULT_PRIORITY);
+ } else {
+ spec.setPriorityClassRef(
+ new PriorityClassRef(
+ SCHEDULING_API_GROUP, "PriorityClass",
priorityClass.getMetadata().getName()));
+ spec.setPriority(priorityClass.getValue());
+ }
+ } catch (KubernetesClientException e) {
+ if (e.getCode() != HTTP_FORBIDDEN) {
+ throw e;
+ }
+ log.warn(
+ "Requesting the Kueue Workload {} without priority because the
operator is not allowed "
+ + "to read the priority classes.",
+ workload.getMetadata().getName());
+ }
+ }
+
+ /**
+ * Returns the PriorityClass of the first pod set which has one, or the
global default
+ * PriorityClass. Like Kueue, the lowest one wins if there are more than one
global default.
+ */
+ private static PriorityClass getPodPriorityClass(
+ final KubernetesClient client, final WorkloadSpec spec) {
+ for (PodSet podSet : spec.getPodSets()) {
+ PodTemplateSpec template = podSet.getTemplate();
+ String name =
+ template == null || template.getSpec() == null
+ ? null
+ : template.getSpec().getPriorityClassName();
+ if (StringUtils.isNotEmpty(name)) {
+ PriorityClass priorityClass =
+ client.scheduling().v1().priorityClasses().withName(name).get();
+ if (priorityClass == null) {
+ throw new IllegalStateException("PriorityClass " + name + " is not
found.");
+ }
+ return priorityClass;
+ }
+ }
+ return
client.scheduling().v1().priorityClasses().list().getItems().stream()
Review Comment:
**Finding 7.** This lookup runs on every reconcile, so a `Workload` with no
label and no pod priority class gets re-priced whenever an admin changes the
cluster's default. Kueue never does that.
Observed on the mock server, with nothing about the Spark resource changed
between the two reconciles:
```
ref after create =
PriorityClassRef(group=scheduling.k8s.io, kind=PriorityClass, name=default-a)
priority=50
ref after the globalDefault change =
PriorityClassRef(group=scheduling.k8s.io, kind=PriorityClass, name=default-b)
priority=70
```
`classifyWorkloadsForPriorityUpdate` drops such a Workload from the update
set outright:
```go
if !workload.HasNoPriority(wl) &&
!workload.IsWorkloadPriorityClass(wl) {
continue
}
```
`HasNoPriority` is `PriorityClassRef == nil` and `IsWorkloadPriorityClass`
requires the `kueue.x-k8s.io` group, so a `scheduling.k8s.io`-backed Workload
never has its priority rewritten after creation. The comment above the function
says it outright: "a Pod PriorityClass-backed workload does not follow the
label at all". The effect here is that Spark `Workload`s would re-rank inside a
`ClusterQueue` while every other workload in it holds still, one write each,
which is the opposite of what this PR is trying to match.
This also corrects the fix I sketched for finding 2. Kueue's gate has two
parts and I only described the second:
1. skip the Workload entirely when its ref is a Pod `PriorityClass`;
2. otherwise compare the label to the ref's name, and resolve only when they
differ.
Part 1 is what fixes this finding, and it is also what makes part 2
well-defined — without it a Pod-`PriorityClass`-backed Workload has no label to
compare against, so a name comparison would resolve on every reconcile anyway.
--
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]
