viirya commented on code in PR #966:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/966#discussion_r4267583165
##########
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)
+ : parseInt(sparkConf.get("spark.executor.instances"),
DEFAULT_EXECUTOR_INSTANCES),
spec.getExecutorSpec(),
sparkConf,
Constants.EXECUTOR_SPARK_CONTAINER_PROP_KEY,
calculateExecutorMemoryMiB(sparkConf, isNonJvmApp(spec),
isPythonApp(spec)));
}
+ /**
+ * Returns `spark.dynamicAllocation.maxExecutors`, whose quota the Workload
of an application with
+ * dynamic allocation reserves if the application is annotated to, since the
quota of an admitted
+ * Workload cannot grow or shrink.
+ */
+ private static int getReservedMaxExecutors(
+ final SparkApplication app, final Map<String, String> sparkConf) {
+ Map<String, String> annotations = app.getMetadata().getAnnotations();
+ if (annotations == null
+ || !"true".equalsIgnoreCase(
+
annotations.get(Constants.ANNOTATION_KUEUE_RESERVE_MAX_EXECUTORS))) {
+ throw new UnsupportedOperationException(
+ "Kueue does not support SparkApplication with dynamic allocation "
+ + "(spark.dynamicAllocation.enabled=true) unless it is annotated
with "
+ + Constants.ANNOTATION_KUEUE_RESERVE_MAX_EXECUTORS
+ + "=true to reserve the quota of
spark.dynamicAllocation.maxExecutors executors.");
+ }
+ // Like Spark, the maximum is unlimited by default
+ int maxExecutors =
Review Comment:
Spark does not bound the initial target by the maximum.
`Utils.getDynamicAllocationInitialExecutors` returns `max(minExecutors,
initialExecutors, spark.executor.instances)`, and
`ExecutorAllocationManager.validateSettings` only checks `min <= max`.
`start()` then requests `initialNumExecutors` as-is, and the target is not
lowered while `initializing`. With e.g. `spark.executor.instances=5` and
`maxExecutors=2`, the Workload reserves 2 executors, but Spark creates 5
executor pods until the first stage or the idle timeout. Since the point of the
Workload is accurate quota accounting, could we reject the case where that max
exceeds `maxExecutors`, like the unset-maximum case below? Reserving the larger
of the two would also work, but rejecting seems simpler.
##########
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) {
Review Comment:
Nit: with this, the "DA requires the annotation" check runs while the pod
set is being built (inside `getReservedMaxExecutors`). How about computing the
executor count in `buildWorkload(SparkApplication)`, next to the `minExecutors`
annotation, and passing the count in? Then all the DA logic lives in one place
and `buildExecutorPodSet` only builds the pod set.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactoryTest.java:
##########
@@ -330,15 +333,65 @@ void testBuildWorkloadDynamicAllocation() {
.build());
ApplicationSpec spec = new ApplicationSpec();
+ // Like Spark, the initial executors are within the minimum and the maximum
Review Comment:
Related to the comment on `getReservedMaxExecutors`: Spark doesn't actually
keep the initial executors within the maximum, it only validates `min <= max`.
Maybe reword this, and add a case for `spark.executor.instances` >
`maxExecutors` once that's handled.
##########
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
Review Comment:
This bullet is one long sentence that covers the annotation, the required
maximum, the reservation for the whole attempt, the effect on preemption, fair
sharing and `ProvisioningRequest`, the `PodsReady` threshold, and the side
effect of a minimum of 0. Splitting it into two or three sub-bullets would make
it much easier to scan.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -72,6 +72,13 @@ public final class KueueWorkloadFactory {
public static final String PODSET_MASTER =
Constants.LABEL_SPARK_ROLE_MASTER_VALUE;
public static final String PODSET_WORKER =
Constants.LABEL_SPARK_ROLE_WORKER_VALUE;
+ /**
+ * Prefix of the annotation of a Workload, followed by the name of a pod
set, which holds how many
+ * of its pods must be ready for the Kueue `PodsReady` condition instead of
its count, e.g. the
+ * executors which dynamic allocation keeps, see {@link
KueueWorkloadUtils#recordPodsReady}.
+ */
+ public static final String ANNOTATION_PODS_READY_PREFIX =
"spark.operator/kueue-pods-ready-";
Review Comment:
Nit: only the executor pod set ever uses this, so a plain constant
`spark.operator/kueue-pods-ready-executor` would remove the
prefix-plus-pod-set-name indirection here and in `countRequiredReadyPods`. The
annotation key stays the same either way, so I don't feel strongly about this.
##########
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:
To avoid over-reading this: the Kubeflow integration of Kueue matches only
on the `PodsReady` threshold. It reserves `spark.executor.instances` (the
initial count), not `maxExecutors`. Maybe say "Like the integration of Kueue
for the Kubeflow Spark Operator, which also uses `minExecutors` as the
`PodsReady` threshold, ..." so readers don't assume the reservation matches too.
##########
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:
This now hashes every annotation of the desired Workload except the hash
itself, so any annotation added to `buildWorkload` later silently becomes a
trigger for recreating pending Workloads. It would also recreate all pending
Workloads on the upgrade that adds it. Would it be safer to hash only the keys
with `ANNOTATION_PODS_READY_PREFIX`? Alternatively, rename the method (it's no
longer just pod sets) and state the contract in the javadoc. Also, the
`annotations.isEmpty() ? "" : ...` branch is what keeps existing hashes stable.
That deserves a short comment so it doesn't get "simplified" away later.
--
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]
