[jira] [Created] (SPARK-58649) Allow SparkApplication drivers ingress to SparkCluster workers

["Yannick Goetschel (Jira)" <ji...@apache.org>]

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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org