YGoetschel opened a new pull request, #787:
URL: https://github.com/apache/spark-kubernetes-operator/pull/787
### What changes were proposed in this pull request?
This PR adds a second ingress peer to the worker `NetworkPolicy` generated
for every
`SparkCluster`, so that driver pods in the same namespace are admitted
alongside the cluster's
own pods.
### Why are the changes needed?
We run per-user Spark Connect sessions as SparkApplications attached to a
shared standalone
`SparkCluster`. Most jobs are fine, but every so often one dies like this:
```
Job aborted due to stage failure: Task 11 in stage 2.0 failed 4 times, most
recent failure:
Lost task 11.3 in stage 2.0 (TID 86) (<executor-ip> executor 3):
TaskResultLost (result lost from block manager)
```
What made this confusing is that the tasks themselves succeed. The History
Server lists them as
SUCCESS with TaskResultLost sitting in the Errors column, so the compute
finished and only the
result transfer failed. It also does not hit every task: in the run above, 3
of 25 tasks in the
stage died this way and the other 22 completed normally.
The driver log explains why only some of them:
```
java.io.IOException: Connecting to /<executor-ip>:<ephemeral-port> timed out
(120000 ms)
at
org.apache.spark.network.client.TransportClientFactory.createClient(...)
at
org.apache.spark.network.netty.NettyBlockTransferService$$anon$2.createAndStart(...)
at
org.apache.spark.network.shuffle.RetryingBlockTransferor.transferAllOutstanding(...)
at org.apache.spark.storage.BlockManager.fetchRemoteManagedBuffer(...)
at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(...)
```
The driver cannot reach the executors' block manager. That only matters once
a task result goes
over `spark.task.maxDirectResultSize` (1 MB): below it the result rides
along inline in the RPC,
above it the executor parks the result in its block manager and hands the
driver an
`IndirectTaskResult` to go and fetch. The tasks that stayed under the limit
never noticed
anything. The three that went over timed out, retried three times and gave
up, and the stage
aborted on the fourth attempt.
The culprit is the worker NetworkPolicy generated in
`SparkClusterResourceSpec#buildWorkerNetworkPolicy`:
```yaml
podSelector:
matchLabels: {spark-role: worker, spark.operator/spark-cluster-name:
<cluster>}
ingress:
- from:
- podSelector:
matchLabels: {spark.operator/spark-cluster-name: <cluster>}
policyTypes: [Ingress]
```
Only pods carrying the cluster label may talk to the workers. But a
SparkApplication attached to
the cluster is a separate resource: its driver is labelled `spark-role:
driver` and
`spark.operator/spark-app-name: <app>`, never the cluster label. So the
driver is locked out of
its own executors.
None of this is visible from the outside. Worker egress is unrestricted and
no policy selects
driver pods, so the executor-to-driver direction works perfectly well —
executors register,
tasks get dispatched, tasks run. Only the return path for large results is
broken, so the
cluster looks healthy right up until a job happens to cross the 1 MB
threshold.
There is no way around it from the client side either. Executors read
`spark.task.maxDirectResultSize` from the driver SparkConf when they start,
so a Connect client
cannot raise it partway through a session.
Ports are left unrestricted on the new peer because block manager ports are
ephemeral, and
pinning `spark.blockManager.port` is not workable when several executors
share a worker pod.
### Does this PR introduce _any_ user-facing change?
Yes. The generated worker `NetworkPolicy` now also admits driver pods in the
same namespace.
Since NetworkPolicies are additive this can only widen ingress, so it cannot
deny traffic that
was previously allowed. Clusters running on a CNI that does not enforce
NetworkPolicy are
unaffected either way.
### How was this patch tested?
Extended `SparkClusterResourceSpecTest#testWorkerNetworkPolicy` to assert
both ingress peers.
```
./gradlew :spark-submission-worker:test --tests
'*SparkClusterResourceSpecTest*'
./gradlew :spark-submission-worker:spotlessCheck
```
Both pass. We also applied the equivalent policy by hand against operator
1.0.0, and the job
that reproduced this now completes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
--
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]
