Yannick Goetschel created SPARK-58649:
-----------------------------------------
Summary: Allow SparkApplication drivers ingress to SparkCluster
workers
Key: SPARK-58649
URL: https://issues.apache.org/jira/browse/SPARK-58649
Project: Spark
Issue Type: Bug
Components: Kubernetes
Affects Versions: kubernetes-operator-1.0.0
Environment: 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:
{code:java}
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)
{code}
The driver log explains why only some of them:
{code:java}
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(...)
{code}
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 the operator generates for every
SparkCluster in
{{SparkClusterResourceSpec#buildWorkerNetworkPolicy}}:
{code:yaml}
podSelector:
matchLabels: {spark-role: worker, spark.operator/spark-cluster-name:
<cluster>}
ingress:
- from:
- podSelector:
matchLabels: {spark.operator/spark-cluster-name: <cluster>}
policyTypes: [Ingress]
{code}
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.
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.
Reporter: Yannick Goetschel
The worker \{{NetworkPolicy}} generated for every \{{SparkCluster}} by
\{{SparkClusterResourceSpec#buildWorkerNetworkPolicy}} admits ingress only
from pods
labelled \{{spark.operator/spark-cluster-name: <cluster>}}:
\{code:yaml}
podSelector:
matchLabels: \{spark-role: worker, spark.operator/spark-cluster-name:
<cluster>}
ingress:
- from:
- podSelector:
matchLabels: \{spark.operator/spark-cluster-name: <cluster>}
policyTypes: [Ingress]
\{code}
A \{{SparkApplication}} that attaches to a standalone \{{SparkCluster}}
(\{{spark.master=spark://<cluster>-master-svc:7077}}) is a separate resource.
Its driver
pod is labelled \{{spark-role: driver}} and \{{spark.operator/spark-app-name:
<app>}},
never the cluster label, so every driver-to-worker connection is dropped.
The gap is invisible under normal load. Worker egress is unrestricted and no
policy
selects driver pods, so executors register with the driver and tasks run
normally. Spark
only uses the driver-to-executor direction when a task result exceeds
\{{spark.task.maxDirectResultSize}} (1 MB): the executor then stores the
result in its
block manager and returns an \{{IndirectTaskResult}} for the driver to fetch.
That fetch
times out, \{{RetryingBlockTransferor}} retries three times, and the task
fails with:
\{code}
TaskResultLost (result lost from block manager)
\{code}
After four attempts the stage aborts. The failure is not reachable any other
way: that
branch in \{{TaskResultGetter#enqueueSuccessfulTask}} only runs for an
\{{IndirectTaskResult}}.
It cannot be worked around from the client side either. Executors read
\{{spark.task.maxDirectResultSize}} from the driver \{{SparkConf}} at launch,
so a Spark
Connect client cannot raise it after the session has started.
Observed on a Spark Connect session attached to a standalone
\{{SparkCluster}}, on a CNI
that enforces NetworkPolicy. Three of 25 tasks in a write stage produced
results above
the threshold and the job aborted; every other task in the same stage
succeeded. Driver
log:
\{code}
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(...)
\{code}
Proposed fix: add a second ingress peer to the generated policy admitting
same-namespace
pods labelled \{{spark-role: driver}}. This only widens ingress and cannot
deny traffic
that was previously allowed.
\{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
