peter-toth commented on code in PR #841:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/841#discussion_r4036382711
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -300,6 +333,23 @@ static long calculateDriverMemoryMiB(
return memMiB + calculateMemoryOverheadMiB(sparkConf, "spark.driver",
memMiB, isNonJvm);
}
+ /**
+ * Calculates total memory in MiB including overhead for Spark standalone
master or worker.
+ *
+ * @param daemonMemory `SPARK_DAEMON_MEMORY`, the JVM heap of the master or
worker daemon.
+ * @param workerMemory `SPARK_WORKER_MEMORY`, the memory which the worker
gives to executors.
+ */
+ static long calculateDaemonMemoryMiB(final String daemonMemory, final String
workerMemory) {
+ // Like Spark, the memory is in bytes unless otherwise specified.
+ String memory = StringUtils.isEmpty(daemonMemory) ? DEFAULT_MEMORY :
daemonMemory;
+ long memMiB = JavaUtils.byteStringAsBytes(memory) / 1024 / 1024;
+ if (StringUtils.isNotEmpty(workerMemory)) {
+ memMiB += JavaUtils.byteStringAsBytes(workerMemory) / 1024 / 1024;
+ }
+ long minOverheadMiB =
JavaUtils.byteStringAsMb(DEFAULT_MIN_MEMORY_OVERHEAD);
+ return memMiB + Math.max((long) (DEFAULT_MEMORY_OVERHEAD_FACTOR * memMiB),
minOverheadMiB);
Review Comment:
**Finding 2.** The overhead rule is borrowed from the `SparkApplication` pod
sets, where `memMiB` is one JVM's heap. Here `memMiB` is `daemonHeap +
SPARK_WORKER_MEMORY`, and `SPARK_WORKER_MEMORY` is a pool the worker hands out
to N separate executor processes, each of which needs its own non-heap headroom.
With `SPARK_WORKER_MEMORY=8g` split into eight 1g executors this returns
`1024 + 8192 + 921`. Spark's own Kubernetes rule for those same eight executors
would add `8 * max(0.1 * 1024, 384) = 3072Mi`, so the pod is short about 2.1
GiB of the headroom Spark would have asked for.
The executor count is not knowable here, so I am not asking for an exact
number. Applying the factor to the daemon heap only and leaving
`SPARK_WORKER_MEMORY` to carry its own executors would at least not imply a
precision the formula does not have:
```java
long memMiB = JavaUtils.byteStringAsBytes(memory) / 1024 / 1024;
long minOverheadMiB =
JavaUtils.byteStringAsMb(DEFAULT_MIN_MEMORY_OVERHEAD);
long total = memMiB + Math.max((long) (DEFAULT_MEMORY_OVERHEAD_FACTOR *
memMiB), minOverheadMiB);
if (StringUtils.isNotEmpty(workerMemory)) {
total += JavaUtils.byteStringAsBytes(workerMemory) / 1024 / 1024;
}
return total;
```
That is a smaller number than what is there now for a large pool, so if you
prefer the current shape a comment saying the factor is a pool-wide
approximation would do.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -188,14 +189,46 @@ private static Workload buildWorkload(
return workload;
}
- private static PodSet buildPodSet(final String name, final StatefulSet
statefulSet) {
+ /**
+ * Builds a PodSet for a Spark standalone role (`master` or `worker`) from
the StatefulSet. Unlike
+ * the SparkApplication pods, Spark does not set the requests of the master
and worker pods, so
+ * the missing CPU and memory requests are calculated from the environment
variables of the
+ * container in the same way as Spark standalone does. Otherwise, Kueue
admits the pods without
+ * accounting them against the quota.
+ */
+ private static PodSet buildPodSet(final String role, final StatefulSet
statefulSet) {
+ PodTemplateSpec templateSpec = statefulSet.getSpec().getTemplate();
+ // The operator always creates the container named after the role.
+ Container container = selectContainer(templateSpec.getSpec(), role, role);
+ Map<String, Quantity> requests = getOrCreateRequests(container);
+ Map<String, Quantity> limits = container.getResources().getLimits();
+ Map<String, String> env = getEnv(container);
+ boolean isWorker = PODSET_WORKER.equals(role);
+ String cpu = isWorker ? env.getOrDefault("SPARK_WORKER_CORES",
DEFAULT_CORES) : DEFAULT_CORES;
Review Comment:
**Finding 1.** Spark standalone does not default the worker to one core.
`WorkerArguments` sets `var cores = inferDefaultCores()` and only overrides it
when the env var is present:
```scala
def inferDefaultCores(): Int = {
Runtime.getRuntime.availableProcessors()
}
def inferDefaultMemory(): Int = {
...
// Leave out 1 GB for the operating system, but don't return a negative
memory size
math.max(totalMb - 1024, Utils.DEFAULT_DRIVER_MEM_MB) //
DEFAULT_DRIVER_MEM_MB == 1024
}
```
`SPARK_WORKER_MEMORY` is the same story. Unset does not mean zero; it means
the node's total memory minus 1 GiB. And with no CPU or memory limit on the
container there is no cgroup bound for the JVM to read, so both inferences see
the node rather than the pod.
Here is what a plain `SparkCluster` produces at this head, with no
`resources` and no env in either template, i.e. the input the description names
as the bug:
```
### podSet=master count=1 container=master requests={cpu=1, memory=1408Mi}
limits={}
### podSet=worker count=2 container=worker requests={cpu=1, memory=1408Mi}
limits={}
```
The master numbers are right: its heap really is `SPARK_DAEMON_MEMORY` and
it reserves no cores. The worker numbers are a floor. A `ClusterQueue` with a
16-CPU quota would admit 16 of these workers, each of which advertises the
whole node to executors, so the quota is still not enforced for the case the PR
targets.
Two ways out that fit in this PR:
1. State the limitation. Correct the description's table, and `log.warn`
here when the worker container has neither `SPARK_WORKER_CORES` /
`SPARK_WORKER_MEMORY` nor a CPU / memory limit, so whoever reads the operator
log knows the Workload is a floor.
2. Refuse the shape, which is what this file already does for the other two
things it cannot account for. `buildWorkload` throws
`UnsupportedOperationException` for a `HorizontalPodAutoscaler` and for dynamic
allocation, both because a count is not knowable. A worker whose size is not
knowable is the same class of problem. Harsher, since it rejects the default
cluster outright.
Finding 4 is the third way: make the size knowable.
On what I checked how: the probe output above I ran at `84df491`. That a
worker pod with no limits really sees the node's CPUs and memory is read off
`WorkerArguments` plus the JVM's container support, not observed on a cluster.
Deploying a default `SparkCluster` on a multi-core node and reading the `Cores`
figure on the worker's web UI would settle it.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -437,6 +478,31 @@ private static void decorateContainerResources(
}
}
+ /** Returns the environment variables with the literal values. `valueFrom`
is ignored. */
Review Comment:
**Finding 3.** `valueFrom` is not the only source this misses. `envFrom` (a
`ConfigMap` or `Secret` env source on the container) is also invisible, and so
is `conf/spark-env.sh`, which is the place Spark's own template documents for
exactly these three variables:
```
# - SPARK_WORKER_CORES, to set the number of cores to use on this machine
# - SPARK_WORKER_MEMORY, to set how much total memory workers have to give
executors (e.g. 1000m, 2g)
# - SPARK_DAEMON_MEMORY, to allocate to the master, worker and history
server themselves (default: 1g).
```
Both `sbin/start-worker.sh` and `sbin/spark-daemon.sh` source
`bin/load-spark-env.sh`, so a `spark-env.sh` baked into a custom image is read
by the worker and cannot be read by the operator. That is not fixable here, but
the comment should say what the accounting actually covers rather than naming
one of three gaps:
```suggestion
/**
* Returns the environment variables with the literal `value`s of the
container. Anything the
* operator cannot read at build time is ignored: `valueFrom`, `envFrom`,
and `conf/spark-env.sh`
* inside the image, which `load-spark-env.sh` sources for the master and
worker daemons.
*/
```
The description's matching bullet says only "Only literal `value`s of the
container `env` are used. `valueFrom` is ignored" and would want the same
widening.
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadFactory.java:
##########
@@ -188,14 +189,46 @@ private static Workload buildWorkload(
return workload;
}
- private static PodSet buildPodSet(final String name, final StatefulSet
statefulSet) {
+ /**
+ * Builds a PodSet for a Spark standalone role (`master` or `worker`) from
the StatefulSet. Unlike
+ * the SparkApplication pods, Spark does not set the requests of the master
and worker pods, so
+ * the missing CPU and memory requests are calculated from the environment
variables of the
+ * container in the same way as Spark standalone does. Otherwise, Kueue
admits the pods without
+ * accounting them against the quota.
+ */
+ private static PodSet buildPodSet(final String role, final StatefulSet
statefulSet) {
Review Comment:
**Finding 4.** An alternative worth considering, as a follow-up rather than
in this PR: instead of reading the env to predict what Spark will infer, have
the operator *write* it.
`SparkClusterResourceSpec.buildWorkerStatefulSet` already owns the worker
container. If it set the two variables from the container's own resources
whenever they are absent, roughly:
```java
// in buildWorkerStatefulSet, from the worker container's limits,
falling back to requests
Quantity cpu = limitOrRequest(container, "cpu");
Quantity mem = limitOrRequest(container, "memory");
if (cpu != null) addEnvIfAbsent("SPARK_WORKER_CORES", cores(cpu));
if (mem != null) addEnvIfAbsent("SPARK_WORKER_MEMORY", (mib(mem) -
daemonMiB - overheadMiB) + "m");
```
then three things follow. The worker's advertised capacity stops depending
on which node it lands on, which is worth having on its own. Spark's view and
Kueue's accounting agree by construction, so finding 1 goes away. And
`buildPodSet` no longer needs `getEnv` at all: the requests are always present,
and it reduces to the `fillMissingRequest` limits fallback.
The honest counter-arguments:
- It changes the applied `StatefulSet`, so an existing cluster's worker
capacity would go from node-sized to limit-sized on upgrade. That is arguably a
fix, but it is a behavior change and belongs in its own ticket.
- It spans `spark-submission-worker`, well outside this PR.
- A worker with no limits at all still has nothing to derive from, so option
1 or 2 from finding 1 is still needed for that case.
- Deliberate oversubscription stays possible only if the operator fills the
variables when absent rather than overwriting them, which is why the sketch
guards on absence.
--
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]
