peter-toth commented on code in PR #881:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/881#discussion_r4096444681
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -365,6 +376,49 @@ public static ReconcileProgress retryAfterRequestFailure(
return ReconcileProgress.completeAndDefaultRequeue();
}
+ /**
+ * Releases the pending Workload of a resource whose queue label was removed
while it waited for
+ * the admission. The resource now starts without Kueue, so Kueue would
otherwise admit the
+ * Workload later into quota which nothing uses. An admitted Workload is
kept, since the
+ * resources it was admitted for may be running already, and it is released
with them like the
+ * Workload of a queued resource. Like the admission request, a failed
release is retried before
+ * the resources are requested.
+ *
+ * @param context The context of the resource without a queue name label.
+ * @return The progress to return while the release fails, or empty to
proceed.
+ */
+ public static Optional<ReconcileProgress> releaseDequeuedWorkload(
+ final BaseContext<?> context) {
+ Optional<Workload> workload = context.getCachedKueueWorkload();
+ if (workload.isEmpty() || isAdmitted(workload.get())) {
Review Comment:
**Finding 2.** Keeping the admitted `Workload` is right, but the branch also
skips `applyAdmittedFlavors`, so the resources it was admitted for are
re-applied without the flavors Kueue assigned.
`releaseDequeuedWorkload` returns `Optional.empty()` for an admitted
`Workload`, and both init steps then carry on and re-apply their secondary
resources. Nothing sets the flavors on the context on this path, so the specs
are rebuilt without them:
- `AppInitStep:137` server-side-applies `getDriverPreResourcesSpec()`, which
holds the executor pod template ConfigMap: `SparkAppResourceSpec:96` takes
Spark's driver pre-resources, and
`SparkAppResourceSpecFactory.applyKueuePodSetFlavors` is what puts the executor
flavor into `spec.executorSpec.podTemplateSpec`. Rewriting that ConfigMap
without the flavor drops the node selector and tolerations from every executor
the driver requests afterwards.
- `ClusterInitStep:112-127` server-side-applies the master and worker
StatefulSets, and `SparkClusterContext.applyKueuePodSetFlavor` is the only
thing that decorates them. Applying them without the flavor changes the pod
template, so the master and worker pods roll.
This is the case the labeled branch guards against, in its own words
"instead of dropping them from the resources" and "from the pod templates",
through `isDriverRequested`/`isMasterRequested` and `applyAdmittedFlavors`. It
is reachable on the same window that guard exists for: the driver or master was
created after the admission, the status update to `DriverRequested` /
`RunningHealthy` failed, and the queue label was removed before the next
reconcile.
```java
if (workload.isEmpty()) {
return Optional.empty();
}
if (isAdmitted(workload.get())) {
// The resources it was admitted for may be running already, so the
Workload is kept and
// released with them, and the flavors it was admitted with are
applied again.
return applyAdmittedFlavors(context);
}
```
One trade-off to weigh with that: `applyAdmittedFlavors` throws
`IllegalArgumentException` on a node-selector conflict, which `ClusterInitStep`
turns into the terminal `SchedulingFailure`. That is what the labeled path
already does, so it is consistent, but it turns a silent drift into a hard
failure. The gap predates this PR, so a follow-up ticket is fine by me.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -165,6 +165,18 @@ public static AdmissionResponse requestAdmission(
return new AdmissionResponse(AdmissionResult.STALE, workload);
}
WorkloadSpec spec = workload.getSpec();
+ boolean changed = false;
+ String desiredQueueName = desired.getSpec().getQueueName();
+ if (!Objects.equals(spec.getQueueName(), desiredQueueName)) {
Review Comment:
**Finding 1.** Kueue freezes `spec.queueName` on the quota reservation, not
on the admission, so this clause issues a PUT the API server rejects.
The `Workload` CRD at the version CI installs (`KUEUE_VERSION: v0.19.4`,
`.github/workflows/build_and_test.yml:10`) carries this rule on the `Workload`
root:
```
+kubebuilder:validation:XValidation:rule="((has(oldSelf.status) && ...
oldSelf.status.conditions.exists(c, c.type == 'QuotaReserved' && c.status ==
'True')) && (... self.status.conditions.exists(c, c.type == 'QuotaReserved' &&
c.status == 'True'))) ? ((has(oldSelf.spec.queueName) ==
has(self.spec.queueName)) && (!has(oldSelf.spec.queueName) ||
oldSelf.spec.queueName == self.spec.queueName)) : true", message="queueName is
immutable while workload quota reserved"
```
`QuotaReserved` precedes `Admitted` whenever the ClusterQueue has admission
checks, and that window is exactly the one the priority clause below is written
for: `quotaReservedWorkloadFollowsWorkloadPriorityClassChange` says "so that
the priority of a Workload waiting for its admission checks can still be
raised". `isPriorityClassChangeAllowed` therefore gates on `isQuotaReserved()`,
and its javadoc names the same CEL family. This clause gates on
`isAdmitted(workload)` only, and the doc sentence the PR adds states the
invariant the clause does not enforce: "since it holds no quota yet"
(`docs/spark_custom_resources.md:640`).
What I measured, on a `QuotaReserved=True` `Workload` with no `Admitted`
condition, with a probe added to `KueueWorkloadUtilsTest` using
`reserveQuota()` and a desired without a priority-class label so only this
clause can fire:
```
queueName=another-queue priorityClassRef=(group=kueue.x-k8s.io,
kind=WorkloadPriorityClass, name=low) rv 3 -> 4
```
So the queue is rewritten and the PUT goes out, while the sibling clause
correctly leaves the priority class alone. With `!quotaReserved &&` added, the
probe reports `queueName=test-queue` and the resourceVersion stays at 3, and
the rest of `KueueWorkloadUtilsTest` still passes.
Two consequences. `422 Invalid` is not in
`ReconcilerUtils.isTransientError`, so `holdForAdmission` publishes
`KueueAdmissionRequestFailed` ("Failed to request Kueue admission") and
requeues on every reconcile for the whole admission-check window.
`KueueAdmissionPending` is not published on those reconciles, and the doc calls
that event the only signal a queued first attempt has. Second, both clauses
share one `update()`, so the rejected PUT takes the priority raise with it.
With both labels changed on a quota-reserved `Workload` I measured
`queueName=another-queue priorityClass=high priority=1000` going out in a
single PUT, so the behavior
`quotaReservedWorkloadFollowsWorkloadPriorityClassChange` covers regresses.
```suggestion
boolean quotaReserved = workload.getStatus() != null &&
workload.getStatus().isQuotaReserved();
if (!quotaReserved && !Objects.equals(spec.getQueueName(),
desiredQueueName)) {
```
That passes `spotlessCheck`, `checkstyleMain`, `pmdMain` and `spotbugsMain`.
Both sibling clauses have a `quotaReserved*` test and this one has none, so a
`quotaReservedWorkloadKeepsItsQueue` next to
`quotaReservedWorkloadKeepsFrozenPriorityClass` would pin it. The doc then
reads "if the `kueue.x-k8s.io/queue-name` label changes before the `Workload`
reserves quota", matching the priority bullet above it, plus a sentence for
what happens after: the change is ignored and the resource keeps running in the
old queue.
One honesty note: I could not get the rejection itself out of a real API
server here, so that half is read off the CRD rule rather than observed. The
experiment that settles it is applying the v0.19.4 `Workload` CRD to a cluster,
patching a `Workload` status to `QuotaReserved=True`, then patching
`spec.queueName`.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -165,6 +165,18 @@ public static AdmissionResponse requestAdmission(
return new AdmissionResponse(AdmissionResult.STALE, workload);
}
WorkloadSpec spec = workload.getSpec();
+ boolean changed = false;
+ String desiredQueueName = desired.getSpec().getQueueName();
+ if (!Objects.equals(spec.getQueueName(), desiredQueueName)) {
+ // Like Kueue, a queue label changed while waiting moves the Workload to
the new queue in
+ // place, since it holds no quota yet.
+ log.info(
+ "Moving the pending Kueue Workload {} to queue {}.",
+ workload.getMetadata().getName(),
+ desiredQueueName);
+ spec.setQueueName(desiredQueueName);
Review Comment:
**Finding 3.** The moved `Workload` keeps the `kueue.x-k8s.io/queue-name`
label of the old queue.
`KueueWorkloadFactory.buildWorkload` copies the owner's labels onto the
`Workload` (`KueueWorkloadFactory:165-169`), so a `Workload` created for `q1`
carries `kueue.x-k8s.io/queue-name: q1`. The move rewrites `spec.queueName`
only, so afterwards `kubectl get workload -l kueue.x-k8s.io/queue-name=q1`
still lists a `Workload` queued in `q2`.
Nothing reads that label off the `Workload` — Kueue goes by
`spec.queueName`, and the operator's own mapper goes by the name label — so
this is cosmetic. It is worth a line anyway because `kubectl get workload` is
what the doc points users at to follow a queued resource.
```java
Map<String, String> labels = new
HashMap<>(workload.getMetadata().getLabels());
labels.put(Constants.LABEL_QUEUE_NAME, desiredQueueName);
workload.getMetadata().setLabels(labels);
```
--
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]
