peter-toth commented on code in PR #966:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/966#discussion_r4267510855
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -301,17 +321,49 @@ static PodSet buildDriverPodSet(
}
static PodSet buildExecutorPodSet(
- final SparkApplication app, final Map<String, String> sparkConf) {
+ final SparkApplication app,
+ final Map<String, String> sparkConf,
+ final boolean dynamicAllocation) {
ApplicationSpec spec = app.getSpec();
return buildPodSet(
PODSET_EXECUTOR,
- parseInt(sparkConf.get("spark.executor.instances"),
DEFAULT_EXECUTOR_INSTANCES),
+ dynamicAllocation
+ ? getReservedMaxExecutors(app, sparkConf)
Review Comment:
**Finding 3.** An idea for a follow-up, not for this PR. Kueue partial
admission could replace the first drawback, waiting until the maximum fits:
- The executor pod set gets `minCount` set to Spark's initial executors, and
`count` set to `maxExecutors`.
- Kueue admits the largest count between the two that fits
(`PodSetReducer`). `PartialAdmission` is Beta and on by default since Kueue
v0.5. Only one pod set of a `Workload` may set `minCount`, which suits the
executors.
- After the admission, the operator sets
`spark.dynamicAllocation.maxExecutors` of the driver to the admitted
`podSetAssignments[executor].count`. Kueue does the same with `parallelism` for
a batch `Job` (`RunWithPodSetsInfo`).
The lower bound has to be the initial executors, `max(minExecutors,
initialExecutors, executor.instances)`, not `minExecutors`. Otherwise
`SchedulerBackendUtils.getInitialTargetExecutorNumber` fails the driver when
the admitted count is below the initial one.
The counter-arguments:
- It quietly lowers the user's `maxExecutors` for the attempt.
- The admitted count has to reach the driver conf build, next to the pod set
infos.
- Preemption and fair sharing still count the admitted number for the whole
attempt.
It would also need its own opt-in, since the annotation is named after
reserving the maximum. WDYT?
##########
docs/spark_custom_resources.md:
##########
@@ -810,9 +810,20 @@ relates it to the reconciliation.
ones with `kubectl get workloads -A -l spark.operator/spark-app-name` and
`kubectl get workloads -A -l spark.operator/spark-cluster-name`, and delete
only those whose
owner has no running pods, e.g. of a `Failed` `SparkCluster`.
-* 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.
+* 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.
+* Since Kueue cannot resize the quota of an admitted `Workload` yet, a
`SparkApplication` with
+ dynamic allocation fails with `SchedulingFailure` unless it is annotated with
+ `spark.operator/kueue-reserve-max-executors: "true"`, which is ignored
without dynamic
+ allocation. As a stopgap, its `Workload` then reserves the quota of
+ `spark.dynamicAllocation.maxExecutors` executors, which must be set since it
is unlimited by
+ default, for the whole attempt: Kueue admits it only once the maximum fits,
and preemption, fair
+ sharing and a `ProvisioningRequest` admission check count the maximum even
while fewer executors
+ run. Like the integration of Kueue for the Kubeflow Spark Operator, the
`PodsReady` condition
Review Comment:
**Finding 1.** `spark.dynamicAllocation.maxExecutors` is a limit per
`ResourceProfile`, not per application. So the `Workload` bounds only the
executors of the default profile.
I read the chain in Spark master (not run on a cluster):
- `ExecutorAllocationManager.addExecutors` compares `maxNumExecutors` with
the target of one `rpId` ("this is doing a max check per ResourceProfile"). Its
test `add executors multiple profiles` sets `maxExecutors` to 10 and reaches a
target of 10 for the default profile plus 10 for a profile with 4 cores and 4
GPUs.
- `KubernetesClusterSchedulerBackend.doRequestTotalExecutors` passes these
targets per profile to `ExecutorPodsAllocator`. It builds each pod with
`rpIdToResourceProfile(resourceProfileId)`, so a pod of another profile gets
that profile's cores, memory and custom resources.
- `ResourceProfileManager.isSupported` rejects an executor profile other
than the default on K8s without dynamic allocation. So a queued application
could not get here before this PR.
So an application with `maxExecutors: "10"` and one `RDD.withResources`
stage can run 10 more executors with GPUs that no `Workload` counts. That is
the CPU ETL then GPU ML example of Spark's stage-level scheduling docs. GPU
quota is also a common reason to put Spark behind Kueue.
The operator can't see a `ResourceProfile` in the spec, so I'd document it.
For example, after "count the maximum even while fewer executors run.":
```markdown
The reservation bounds only the executors of the default resource profile.
Spark applies
`spark.dynamicAllocation.maxExecutors` to each `ResourceProfile`
separately, so with stage-level
scheduling, e.g. `RDD.withResources`, the executors of another profile run
outside the quota.
```
The `Dynamic allocation` row of the feature table could say it as well. A
job with one `withResources` stage under the E2E queue would show it, by
counting its executor pods against `podSetAssignments`.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java:
##########
@@ -1034,17 +1054,23 @@ private static void deleteWorkload(final
KubernetesClient client, final Workload
}
/**
- * Hashes the pod sets of the desired Workload with sorted map keys. The
hash is stored as an
- * annotation at creation, so the comparison does not depend on the
defaulting applied to the
- * stored pod sets.
+ * Hashes the pod sets of the desired Workload with sorted map keys,
together with the annotations
+ * which it is built with, e.g. how many pods of a pod set must be ready, so
that a change of
+ * either recreates a pending Workload. The hash is stored as an annotation
at creation, so the
+ * comparison does not depend on the defaulting applied to the stored pod
sets.
*/
static String hashPodSets(final Workload workload) {
try {
+ ObjectWriter writer =
+
ModelUtils.objectMapper.writer().with(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
+ Map<String, String> annotations =
+ new HashMap<>(
+
Objects.requireNonNullElse(workload.getMetadata().getAnnotations(), Map.of()));
+ // Added to the desired Workload once it is computed
+ annotations.remove(ANNOTATION_POD_SETS_HASH);
String json =
- ModelUtils.objectMapper
- .writer()
- .with(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)
- .writeValueAsString(workload.getSpec().getPodSets());
+ writer.writeValueAsString(workload.getSpec().getPodSets())
+ + (annotations.isEmpty() ? "" :
writer.writeValueAsString(annotations));
Review Comment:
**Finding 2.** Kueue never reads `spark.operator/kueue-pods-ready-executor`.
The operator reads it only after the admission, in `shouldRecordPodsReady`.
Hashing it still makes a change of only `spark.dynamicAllocation.minExecutors`
delete the pending `Workload` (lines 206-213). The new `Workload` gets a new
`creationTimestamp`, which is what Kueue orders a queue by
(`Ordering.GetQueueOrderTimestamp` in v0.20.0). So the application goes to the
back of its queue although nothing Kueue looks at changed.
The queue name and the priority class are applied in place for this reason
(lines 219-245), and the annotation can take the same path:
- `hashPodSets` hashes the pod sets only, like before this PR.
- The in-place block of `requestAdmission` copies the
`ANNOTATION_PODS_READY_PREFIX` annotations of `desired` over those of the
stored `Workload`:
```java
Map<String, String> desiredPodsReady = podsReadyAnnotations(desired);
if (!desiredPodsReady.equals(podsReadyAnnotations(workload))) {
// Kueue does not read them, so the Workload keeps its position in the
queue
Map<String, String> annotations = new
HashMap<>(workload.getMetadata().getAnnotations());
annotations
.keySet()
.removeIf(key ->
key.startsWith(KueueWorkloadFactory.ANNOTATION_PODS_READY_PREFIX));
annotations.putAll(desiredPodsReady);
workload.getMetadata().setAnnotations(annotations);
changed = true;
}
```
`podsReadyAnnotations` returns the annotations with that prefix. I tried
this locally with a test that requests the admission, sets `minExecutors` to 1
and requests it again:
- On this head, the second request returns `STALE` and the `Workload` is
gone.
- With the change, it returns `PENDING`, and the `Workload` keeps its `uid`
and gets the new annotation.
The other Kueue tests pass. Only
`hashPodSetsCoversTheExecutorsWhichPodsReadyRequires` fails, since it asserts
the hash change, so it would turn into a test of the in-place update. The PR
description would need the same update.
--
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]
