Skip to content

[SPARK-58748][CORE][K8S] Preserve advertised driver host for wildcard bind addresses #57977

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sunchao wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[SPARK-58748][CORE][K8S] Preserve advertised driver host for wildcard bind addresses #57977

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion core/src/main/scala/org/apache/spark/SparkContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,8 @@ class SparkContext(config: SparkConf) extends Logging {
// Set Spark driver host and port system properties. This explicitly sets the configuration
// instead of relying on the default value of the config constant.
if (SparkMasterRegex.isK8s(master) &&
_conf.getBoolean("spark.kubernetes.executor.useDriverPodIP", true)) {
_conf.getBoolean("spark.kubernetes.executor.useDriverPodIP", true) &&
!Utils.isAnyLocalAddress(_conf.get(DRIVER_BIND_ADDRESS))) {

@sarutak sarutak Aug 13, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: When the wildcard fallback is taken, could we add a log line so users know why DRIVER_BIND_ADDRESS wasn't used as the advertised address? e.g.:

logInfo(log"spark.kubernetes.executor.useDriverPodIP is true but bind address " +
  log"${MDC(LogKeys.BIND_ADDRESS, _conf.get(DRIVER_BIND_ADDRESS))} is a wildcard; " +
  log"preserving advertised driver host ${MDC(LogKeys.HOST, _conf.get(DRIVER_HOST_ADDRESS))}")

This would help users debug cases where they set useDriverPodIP=true but the bind address is intentionally 0.0.0.0 or ::.

logInfo("Use DRIVER_BIND_ADDRESS instead of DRIVER_HOST_ADDRESS as driver address " +
"because spark.kubernetes.executor.useDriverPodIP is true in K8s mode.")
_conf.set(DRIVER_HOST_ADDRESS, Utils.normalizeIpIfNeeded(_conf.get(DRIVER_BIND_ADDRESS)))
Expand Down
6 changes: 6 additions & 0 deletions core/src/main/scala/org/apache/spark/util/Utils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -960,6 +960,12 @@ private[spark] object Utils
}
}

/** Returns whether a literal IPv4 or IPv6 address binds to every local interface. */
private[spark] def isAnyLocalAddress(host: String): Boolean = {
val address = host.stripPrefix("[").stripSuffix("]")
InetAddresses.isInetAddress(address) && InetAddresses.forString(address).isAnyLocalAddress
}

/**
* Checks if the host contains only valid hostname/ip without port
* NOTE: Incase of IPV6 ip it should be enclosed inside []
Expand Down
33 changes: 32 additions & 1 deletion core/src/test/scala/org/apache/spark/SparkContextSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ import org.apache.spark.launcher.SparkLauncher
import org.apache.spark.resource.ResourceAllocation
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.resource.TestResourceIDs._
import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorMetricsUpdate, SparkListenerJobStart, SparkListenerTaskEnd, SparkListenerTaskStart}
import org.apache.spark.scheduler.{LiveListenerBus, SparkListener, SparkListenerExecutorMetricsUpdate, SparkListenerJobStart, SparkListenerTaskEnd, SparkListenerTaskStart}
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.util.{ThreadUtils, Utils}
import org.apache.spark.util.ArrayImplicits._
Expand Down Expand Up @@ -713,6 +713,37 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
}
}

test("SPARK-58748: Kubernetes drivers do not advertise wildcard bind addresses") {
Seq(
("0.0.0.0", "10.129.36.37", "10.129.36.37"),
("::", "10.129.36.37", "10.129.36.37"),
("10.138.148.230", "driver-service", "10.138.148.230"),
("2001:DB8:0:0::BEEF", "driver-service", "[2001:db8::beef]")).foreach {
case (bindAddress, advertisedAddress, expectedAddress) =>
var observedAddress: Option[String] = None
val conf = new SparkConf(false)
.setMaster("k8s://https://localhost:6443")
.setAppName("driver-bind-address")
.set(DRIVER_BIND_ADDRESS, bindAddress)
.set(DRIVER_HOST_ADDRESS, advertisedAddress)

val error = intercept[SparkException] {
new SparkContext(conf) {
override private[spark] def createSparkEnv(
conf: SparkConf,
isLocal: Boolean,
listenerBus: LiveListenerBus): SparkEnv = {
observedAddress = Some(conf.get(DRIVER_HOST_ADDRESS))
throw new SparkException("stop after resolving the driver address")
}
}
}

assert(error.getMessage === "stop after resolving the driver address")
assert(observedAddress.contains(expectedAddress))
}
}

testCancellingTasks("that raise interrupted exception on cancel") {
Thread.sleep(9999999)
}
Expand Down
9 changes: 9 additions & 0 deletions core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1486,6 +1486,15 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
assert(Utils.buildLocationMetadata(paths, 18) == "(5 paths)[path0, path1, ...]")
}

test("SPARK-58748: wildcard IPv4 and IPv6 bind addresses are not advertised driver addresses") {
Seq("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0").foreach { address =>
assert(Utils.isAnyLocalAddress(address), s"$address should be a wildcard address")
}
Seq("10.129.36.37", "::1", "[::1]", "2001:db8::1", "localhost").foreach { address =>
assert(!Utils.isAnyLocalAddress(address), s"$address should not be a wildcard address")
}
}

test("checkHost supports both IPV4 and IPV6") {
// IPV4 ips
Utils.checkHost("0.0.0.0")
Expand Down
3 changes: 2 additions & 1 deletion ...s/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ private[spark] class BasicExecutorFeatureStep(

private val executorPodNamePrefix = kubernetesConf.resourceNamePrefix

private val driverAddress = if (kubernetesConf.get(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP)) {
private val driverAddress = if (kubernetesConf.get(KUBERNETES_EXECUTOR_USE_DRIVER_POD_IP) &&
!Utils.isAnyLocalAddress(kubernetesConf.get(DRIVER_BIND_ADDRESS))) {
kubernetesConf.get(DRIVER_BIND_ADDRESS)
} else {
kubernetesConf.get(DRIVER_HOST_ADDRESS)
Expand Down
11 changes: 8 additions & 3 deletions ...e/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,11 +352,16 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
ENV_EXECUTOR_ATTRIBUTE_EXECUTOR_ID -> KubernetesTestConf.EXECUTOR_ID))
}

test("SPARK-53944: Support spark.kubernetes.executor.useDriverPodIP") {
test("SPARK-53944, SPARK-58748: Support spark.kubernetes.executor.useDriverPodIP") {
Seq(
(false, "bindAddress", "localhost"),
(false, "10.138.148.230", "localhost"),
(true, "10.138.148.230", "10.138.148.230"),
(true, "bindAddress", "bindAddress"),
(true, "2001:DB8:0:0::BEEF", "[2001:db8::beef]")).foreach {
(true, "2001:DB8:0:0::BEEF", "[2001:db8::beef]"),
(true, "0.0.0.0", "localhost"),
(true, "::", "localhost"),
(true, "[::]", "localhost"),
(true, "0:0:0:0:0:0:0:0", "localhost")).foreach {
case (flag, bindAddress, address) =>
val conf = baseConf.clone()
.set(DRIVER_BIND_ADDRESS, bindAddress)
Expand Down