dongjoon-hyun commented on code in PR #966:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/966#discussion_r4267727261
##########
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:
Thank you, @peter-toth. You're right.
`ExecutorAllocationManager.addExecutors` checks `maxExecutors` per
`ResourceProfile`, and `ResourceProfileManager.isSupported` allows a
non-default profile on K8s only with dynamic allocation. Since the operator
cannot see the profiles in the spec, I documented that the reservation covers
only the executors of the default resource profile, in both the feature table
and the dynamic allocation item.
##########
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:
Nice catch. Since Kueue never reads the annotation, it is now updated in
place like the queue name and the priority class, and `hashPodSets` covers the
pod sets only again. The test now checks that a `minExecutors` change keeps the
pending `Workload` and its `uid`.
##########
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:
Thank you for the idea. I agree that it fits better in a new JIRA issue with
its own opt-in, since it lowers the user's `maxExecutors` and the admitted
count needs to reach the driver conf.
##########
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:
Thank you for the review, @viirya. Spark doesn't create 5 executor pods in
that case. `KubernetesClusterSchedulerBackend` computes its `initialExecutors`
with `SchedulerBackendUtils.getInitialTargetExecutorNumber`, which requires
`minExecutors <= initial executors <= maxExecutors` when dynamic allocation is
enabled. It runs while `SparkContext` creates the task scheduler, before the
`ExecutorAllocationManager` is created, so such a driver fails at startup
(Spark 3.5.0, 4.0.0 and 4.2.0). Still, rejecting it upfront is better than
waiting for the admission of an attempt which fails right away, so I added the
check.
##########
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:
Reworded, and added the cases where `spark.dynamicAllocation.minExecutors`,
`spark.dynamicAllocation.initialExecutors` or `spark.executor.instances`
exceeds the maximum.
##########
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 is addressed together with @peter-toth's Finding 2. The annotation is
now updated in place, and `hashPodSets` covers the pod sets only again.
--
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]
