diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index ae3855084a650d083f0c1b8646fcdeaa9a4fcc06..0e5a20c578db3ba81f9b3e5c2d9c0e71b54467f5 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -537,6 +537,20 @@ See the [configuration page](configuration.html) for information on Spark config
for more details.
</td>
</tr>
+<tr>
+ <td><code>spark.mesos.network.labels</code></td>
+ <td><code>(none)</code></td>
+ <td>
+ Pass network labels to CNI plugins. This is a comma-separated list
+ of key-value pairs, where each key-value pair has the format key:value.
+ Example:
+
+ <pre>key1:val1,key2:val2</pre>
+ See
+ <a href="http://mesos.apache.org/documentation/latest/cni/#mesos-meta-data-to-cni-plugins">the Mesos CNI docs</a>
+ for more details.
+ </td>
+</tr>
<tr>
<td><code>spark.mesos.fetcherCache.enable</code></td>
<td><code>false</code></td>
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala
index 6c8619e3c3c13fab761d57e5b57e693747cf8869..a5015b924331649991f8337a73aa6c8466b167e0 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala
@@ -56,7 +56,7 @@ package object config {
.stringConf
.createOptional
- private [spark] val DRIVER_LABELS =
+ private[spark] val DRIVER_LABELS =
ConfigBuilder("spark.mesos.driver.labels")
.doc("Mesos labels to add to the driver. Labels are free-form key-value pairs. Key-value " +
"pairs should be separated by a colon, and commas used to list more than one." +
@@ -64,10 +64,25 @@ package object config {
.stringConf
.createOptional
- private [spark] val DRIVER_FAILOVER_TIMEOUT =
+ private[spark] val DRIVER_FAILOVER_TIMEOUT =
ConfigBuilder("spark.mesos.driver.failoverTimeout")
.doc("Amount of time in seconds that the master will wait to hear from the driver, " +
"during a temporary disconnection, before tearing down all the executors.")
.doubleConf
.createWithDefault(0.0)
+
+ private[spark] val NETWORK_NAME =
+ ConfigBuilder("spark.mesos.network.name")
+ .doc("Attach containers to the given named network. If this job is launched " +
+ "in cluster mode, also launch the driver in the given named network.")
+ .stringConf
+ .createOptional
+
+ private[spark] val NETWORK_LABELS =
+ ConfigBuilder("spark.mesos.network.labels")
+ .doc("Network labels to pass to CNI plugins. This is a comma-separated list " +
+ "of key-value pairs, where each key-value pair has the format key:value. " +
+ "Example: key1:val1,key2:val2")
+ .stringConf
+ .createOptional
}
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
index 5ecd466194d8bdd0b259814ca95e9c32bd336bd1..26699873145b4178e29d035e444b86f81656ae35 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
@@ -670,7 +670,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
}
private def executorHostname(offer: Offer): String = {
- if (sc.conf.getOption("spark.mesos.network.name").isDefined) {
+ if (sc.conf.get(NETWORK_NAME).isDefined) {
// The agent's IP is not visible in a CNI container, so we bind to 0.0.0.0
"0.0.0.0"
} else {
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtil.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtil.scala
index fbcbc55099ec55af1d424260e462fa2cbd685134..e5c1e801f2772ae8ed9d06ac45ead5b5cd80c1b0 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtil.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtil.scala
@@ -21,6 +21,7 @@ import org.apache.mesos.Protos.{ContainerInfo, Image, NetworkInfo, Parameter, Vo
import org.apache.mesos.Protos.ContainerInfo.{DockerInfo, MesosInfo}
import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.mesos.config.{NETWORK_LABELS, NETWORK_NAME}
import org.apache.spark.internal.Logging
/**
@@ -161,8 +162,12 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
volumes.foreach(_.foreach(containerInfo.addVolumes(_)))
}
- conf.getOption("spark.mesos.network.name").map { name =>
- val info = NetworkInfo.newBuilder().setName(name).build()
+ conf.get(NETWORK_NAME).map { name =>
+ val networkLabels = MesosProtoUtils.mesosLabels(conf.get(NETWORK_LABELS).getOrElse(""))
+ val info = NetworkInfo.newBuilder()
+ .setName(name)
+ .setLabels(networkLabels)
+ .build()
containerInfo.addNetworkInfos(info)
}
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
index 0bb47906347d5acbbd7dfa36a8da22f61808f427..50bb50107150949be7453e00c7e2418cde153889 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
@@ -222,7 +222,7 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
assert(env.getOrElse("TEST_ENV", null) == "TEST_VAL")
}
- test("supports spark.mesos.network.name") {
+ test("supports spark.mesos.network.name and spark.mesos.network.labels") {
setScheduler()
val mem = 1000
@@ -233,7 +233,8 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
command,
Map("spark.mesos.executor.home" -> "test",
"spark.app.name" -> "test",
- "spark.mesos.network.name" -> "test-network-name"),
+ "spark.mesos.network.name" -> "test-network-name",
+ "spark.mesos.network.labels" -> "key1:val1,key2:val2"),
"s1",
new Date()))
@@ -246,6 +247,10 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
val networkInfos = launchedTasks.head.getContainer.getNetworkInfosList
assert(networkInfos.size == 1)
assert(networkInfos.get(0).getName == "test-network-name")
+ assert(networkInfos.get(0).getLabels.getLabels(0).getKey == "key1")
+ assert(networkInfos.get(0).getLabels.getLabels(0).getValue == "val1")
+ assert(networkInfos.get(0).getLabels.getLabels(1).getKey == "key2")
+ assert(networkInfos.get(0).getLabels.getLabels(1).getValue == "val2")
}
test("supports spark.mesos.driver.labels") {
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala
index a8175e29bc9cf74b79ef092c6e42658e481f7801..ab29c295dd8933fd5007c595ca964f554ca5191f 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala
@@ -568,9 +568,10 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
assert(launchedTasks.head.getLabels.equals(taskLabels))
}
- test("mesos supports spark.mesos.network.name") {
+ test("mesos supports spark.mesos.network.name and spark.mesos.network.labels") {
setBackend(Map(
- "spark.mesos.network.name" -> "test-network-name"
+ "spark.mesos.network.name" -> "test-network-name",
+ "spark.mesos.network.labels" -> "key1:val1,key2:val2"
))
val (mem, cpu) = (backend.executorMemory(sc), 4)
@@ -582,6 +583,10 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
val networkInfos = launchedTasks.head.getContainer.getNetworkInfosList
assert(networkInfos.size == 1)
assert(networkInfos.get(0).getName == "test-network-name")
+ assert(networkInfos.get(0).getLabels.getLabels(0).getKey == "key1")
+ assert(networkInfos.get(0).getLabels.getLabels(0).getValue == "val1")
+ assert(networkInfos.get(0).getLabels.getLabels(1).getKey == "key2")
+ assert(networkInfos.get(0).getLabels.getLabels(1).getValue == "val2")
}
test("supports spark.scheduler.minRegisteredResourcesRatio") {