From b289bcc95f0b67cda94ddf416fc9a15e5d1855b4 Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Wed, 4 Oct 2017 07:30:31 -0400
Subject: [PATCH 1/8] [SPARK-22131] Mesos driver secrets. The driver launches
 executors that have access to env or file-based secrets.

---
 .../apache/spark/deploy/mesos/config.scala    |  64 ++++----
 .../cluster/mesos/MesosClusterScheduler.scala | 116 +-------------
 .../MesosCoarseGrainedSchedulerBackend.scala  |   5 +-
 .../MesosFineGrainedSchedulerBackend.scala    |   4 +-
 .../mesos/MesosSchedulerBackendUtil.scala     | 123 +++++++++++++-
 .../mesos/MesosClusterSchedulerSuite.scala    | 150 +++---------------
 ...osCoarseGrainedSchedulerBackendSuite.scala |  34 +++-
 .../MesosSchedulerBackendUtilSuite.scala      |   5 +-
 .../spark/scheduler/cluster/mesos/Utils.scala | 107 +++++++++++++
 9 files changed, 325 insertions(+), 283 deletions(-)

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 7e85de91c5d3..1d1ff423f11f 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
@@ -21,6 +21,39 @@ import java.util.concurrent.TimeUnit
 
 import org.apache.spark.internal.config.ConfigBuilder
 
+private[spark] class MesosSecretConfig(taskType: String) {
+  private[spark] val SECRET_NAME =
+    ConfigBuilder(s"spark.mesos.$taskType.secret.names")
+      .doc("A comma-separated list of secret reference names. Consult the Mesos Secret " +
+        "protobuf for more information.")
+      .stringConf
+      .toSequence
+      .createOptional
+
+  private[spark] val SECRET_VALUE =
+    ConfigBuilder(s"spark.mesos.$taskType.secret.values")
+      .doc("A comma-separated list of secret values.")
+      .stringConf
+      .toSequence
+      .createOptional
+
+  private[spark] val SECRET_ENVKEY =
+    ConfigBuilder(s"spark.mesos.$taskType.secret.envkeys")
+      .doc("A comma-separated list of the environment variables to contain the secrets." +
+        "The environment variable will be set on the driver.")
+      .stringConf
+      .toSequence
+      .createOptional
+
+  private[spark] val SECRET_FILENAME =
+    ConfigBuilder(s"spark.mesos.$taskType.secret.filenames")
+      .doc("A comma-separated list of file paths secret will be written to.  Consult the Mesos " +
+        "Secret protobuf for more information.")
+      .stringConf
+      .toSequence
+      .createOptional
+}
+
 package object config {
 
   /* Common app configuration. */
@@ -64,36 +97,9 @@ package object config {
       .stringConf
       .createOptional
 
-  private[spark] val SECRET_NAME =
-    ConfigBuilder("spark.mesos.driver.secret.names")
-      .doc("A comma-separated list of secret reference names. Consult the Mesos Secret protobuf " +
-        "for more information.")
-      .stringConf
-      .toSequence
-      .createOptional
-
-  private[spark] val SECRET_VALUE =
-    ConfigBuilder("spark.mesos.driver.secret.values")
-      .doc("A comma-separated list of secret values.")
-      .stringConf
-      .toSequence
-      .createOptional
+  private[spark] val driverSecretConfig = new MesosSecretConfig("driver")
 
-  private[spark] val SECRET_ENVKEY =
-    ConfigBuilder("spark.mesos.driver.secret.envkeys")
-      .doc("A comma-separated list of the environment variables to contain the secrets." +
-        "The environment variable will be set on the driver.")
-      .stringConf
-      .toSequence
-      .createOptional
-
-  private[spark] val SECRET_FILENAME =
-    ConfigBuilder("spark.mesos.driver.secret.filenames")
-      .doc("A comma-seperated list of file paths secret will be written to.  Consult the Mesos " +
-        "Secret protobuf for more information.")
-      .stringConf
-      .toSequence
-      .createOptional
+  private[spark] val executorSecretConfig = new MesosSecretConfig("executor")
 
   private[spark] val DRIVER_FAILOVER_TIMEOUT =
     ConfigBuilder("spark.mesos.driver.failoverTimeout")
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index ec533f91474f..47b7ca184d51 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -28,7 +28,6 @@ import org.apache.mesos.{Scheduler, SchedulerDriver}
 import org.apache.mesos.Protos.{TaskState => MesosTaskState, _}
 import org.apache.mesos.Protos.Environment.Variable
 import org.apache.mesos.Protos.TaskStatus.Reason
-import org.apache.mesos.protobuf.ByteString
 
 import org.apache.spark.{SecurityManager, SparkConf, SparkException, TaskState}
 import org.apache.spark.deploy.mesos.MesosDriverDescription
@@ -394,39 +393,11 @@ private[spark] class MesosClusterScheduler(
     }
 
     // add secret environment variables
-    getSecretEnvVar(desc).foreach { variable =>
-      if (variable.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName}" +
-          s"on file ${variable.getName}")
-      } else {
-        logInfo(s"Setting secret on environment variable name=${variable.getName}")
-      }
-      envBuilder.addVariables(variable)
-    }
+    MesosSchedulerBackendUtil.addSecretEnvVar(envBuilder, desc.conf, config.driverSecretConfig)
 
     envBuilder.build()
   }
 
-  private def getSecretEnvVar(desc: MesosDriverDescription): List[Variable] = {
-    val secrets = getSecrets(desc)
-    val secretEnvKeys = desc.conf.get(config.SECRET_ENVKEY).getOrElse(Nil)
-    if (illegalSecretInput(secretEnvKeys, secrets)) {
-      throw new SparkException(
-        s"Need to give equal numbers of secrets and environment keys " +
-          s"for environment-based reference secrets got secrets $secrets, " +
-          s"and keys $secretEnvKeys")
-    }
-
-    secrets.zip(secretEnvKeys).map {
-      case (s, k) =>
-        Variable.newBuilder()
-          .setName(k)
-          .setType(Variable.Type.SECRET)
-          .setSecret(s)
-          .build
-    }.toList
-  }
-
   private def getDriverUris(desc: MesosDriverDescription): List[CommandInfo.URI] = {
     val confUris = List(conf.getOption("spark.mesos.uris"),
       desc.conf.getOption("spark.mesos.uris"),
@@ -572,96 +543,13 @@ private[spark] class MesosClusterScheduler(
       .setName(s"Driver for ${appName}")
       .setSlaveId(offer.offer.getSlaveId)
       .setCommand(buildDriverCommand(desc))
-      .setContainer(getContainerInfo(desc))
+      .setContainer(MesosSchedulerBackendUtil.containerInfo(desc.conf, config.driverSecretConfig))
       .addAllResources(cpuResourcesToUse.asJava)
       .addAllResources(memResourcesToUse.asJava)
       .setLabels(driverLabels)
       .build
   }
 
-  private def getContainerInfo(desc: MesosDriverDescription): ContainerInfo.Builder = {
-    val containerInfo = MesosSchedulerBackendUtil.containerInfo(desc.conf)
-
-    getSecretVolume(desc).foreach { volume =>
-      if (volume.getSource.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName}" +
-          s"on file ${volume.getContainerPath}")
-      } else {
-        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
-      }
-      containerInfo.addVolumes(volume)
-    }
-
-    containerInfo
-  }
-
-
-  private def getSecrets(desc: MesosDriverDescription): Seq[Secret] = {
-    def createValueSecret(data: String): Secret = {
-      Secret.newBuilder()
-        .setType(Secret.Type.VALUE)
-        .setValue(Secret.Value.newBuilder().setData(ByteString.copyFrom(data.getBytes)))
-        .build()
-    }
-
-    def createReferenceSecret(name: String): Secret = {
-      Secret.newBuilder()
-        .setReference(Secret.Reference.newBuilder().setName(name))
-        .setType(Secret.Type.REFERENCE)
-        .build()
-    }
-
-    val referenceSecrets: Seq[Secret] =
-      desc.conf.get(config.SECRET_NAME).getOrElse(Nil).map(s => createReferenceSecret(s))
-
-    val valueSecrets: Seq[Secret] = {
-      desc.conf.get(config.SECRET_VALUE).getOrElse(Nil).map(s => createValueSecret(s))
-    }
-
-    if (valueSecrets.nonEmpty && referenceSecrets.nonEmpty) {
-      throw new SparkException("Cannot specify VALUE type secrets and REFERENCE types ones")
-    }
-
-    if (referenceSecrets.nonEmpty) referenceSecrets else valueSecrets
-  }
-
-  private def illegalSecretInput(dest: Seq[String], s: Seq[Secret]): Boolean = {
-    if (dest.isEmpty) {  // no destination set (ie not using secrets of this type
-      return false
-    }
-    if (dest.nonEmpty && s.nonEmpty) {
-      // make sure there is a destination for each secret of this type
-      if (dest.length != s.length) {
-        return true
-      }
-    }
-    false
-  }
-
-  private def getSecretVolume(desc: MesosDriverDescription): List[Volume] = {
-    val secrets = getSecrets(desc)
-    val secretPaths: Seq[String] =
-      desc.conf.get(config.SECRET_FILENAME).getOrElse(Nil)
-
-    if (illegalSecretInput(secretPaths, secrets)) {
-      throw new SparkException(
-        s"Need to give equal numbers of secrets and file paths for file-based " +
-          s"reference secrets got secrets $secrets, and paths $secretPaths")
-    }
-
-    secrets.zip(secretPaths).map {
-      case (s, p) =>
-        val source = Volume.Source.newBuilder()
-          .setType(Volume.Source.Type.SECRET)
-          .setSecret(s)
-        Volume.newBuilder()
-          .setContainerPath(p)
-          .setSource(source)
-          .setMode(Volume.Mode.RO)
-          .build
-    }.toList
-  }
-
   /**
    * This method takes all the possible candidates and attempt to schedule them with Mesos offers.
    * Every time a new task is scheduled, the afterLaunchCallback is called to perform post scheduled
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 80c0a041b732..8982042530dd 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
@@ -232,6 +232,9 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
         .setValue(value)
         .build())
     }
+
+    MesosSchedulerBackendUtil.addSecretEnvVar(environment, conf, executorSecretConfig)
+
     val command = CommandInfo.newBuilder()
       .setEnvironment(environment)
 
@@ -457,7 +460,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
             .setName(s"${sc.appName} $taskId")
             .setLabels(MesosProtoUtils.mesosLabels(taskLabels))
             .addAllResources(resourcesToUse.asJava)
-            .setContainer(MesosSchedulerBackendUtil.containerInfo(sc.conf))
+            .setContainer(MesosSchedulerBackendUtil.containerInfo(sc.conf, executorSecretConfig))
 
           tasks(offer.getId) ::= taskBuilder.build()
           remainingResources(offerId) = resourcesLeft.asJava
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
index 66b8e0a64012..103aad800a15 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
@@ -28,6 +28,7 @@ import org.apache.mesos.SchedulerDriver
 import org.apache.mesos.protobuf.ByteString
 
 import org.apache.spark.{SparkContext, SparkException, TaskState}
+import org.apache.spark.deploy.mesos.config
 import org.apache.spark.executor.MesosExecutorBackend
 import org.apache.spark.scheduler._
 import org.apache.spark.scheduler.cluster.ExecutorInfo
@@ -159,7 +160,8 @@ private[spark] class MesosFineGrainedSchedulerBackend(
       .setCommand(command)
       .setData(ByteString.copyFrom(createExecArg()))
 
-    executorInfo.setContainer(MesosSchedulerBackendUtil.containerInfo(sc.conf))
+    executorInfo.setContainer(
+      MesosSchedulerBackendUtil.containerInfo(sc.conf, config.executorSecretConfig))
     (executorInfo.build(), resourcesAfterMem.asJava)
   }
 
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 f29e541addf2..ec38c25a99f0 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
@@ -17,10 +17,14 @@
 
 package org.apache.spark.scheduler.cluster.mesos
 
-import org.apache.mesos.Protos.{ContainerInfo, Image, NetworkInfo, Parameter, Volume}
+import org.apache.mesos.Protos._
 import org.apache.mesos.Protos.ContainerInfo.{DockerInfo, MesosInfo}
+import org.apache.mesos.Protos.Environment.Variable
+import org.apache.mesos.protobuf.ByteString
 
-import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.SparkConf
+import org.apache.spark.SparkException
+import org.apache.spark.deploy.mesos.MesosSecretConfig
 import org.apache.spark.deploy.mesos.config.{NETWORK_LABELS, NETWORK_NAME}
 import org.apache.spark.internal.Logging
 
@@ -122,7 +126,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     .toList
   }
 
-  def containerInfo(conf: SparkConf): ContainerInfo.Builder = {
+  def containerInfo(conf: SparkConf, secretConfig: MesosSecretConfig): ContainerInfo.Builder = {
     val containerType = if (conf.contains("spark.mesos.executor.docker.image") &&
       conf.get("spark.mesos.containerizer", "docker") == "docker") {
       ContainerInfo.Type.DOCKER
@@ -170,9 +174,122 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
       containerInfo.addNetworkInfos(info)
     }
 
+    getSecretVolume(conf, secretConfig).foreach { volume =>
+      if (volume.getSource.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName}" +
+          s"on file ${volume.getContainerPath}")
+      } else {
+        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
+      }
+      containerInfo.addVolumes(volume)
+    }
+
     containerInfo
   }
 
+  def addSecretEnvVar(
+      envBuilder: Environment.Builder,
+      conf: SparkConf,
+      secretConfig: MesosSecretConfig): Unit = {
+    getSecretEnvVar(conf, secretConfig).foreach { variable =>
+      if (variable.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName}" +
+          s"on file ${variable.getName}")
+      } else {
+        logInfo(s"Setting secret on environment variable name=${variable.getName}")
+      }
+      envBuilder.addVariables(variable)
+    }
+  }
+
+  private def getSecrets(conf: SparkConf, secretConfig: MesosSecretConfig):
+  Seq[Secret] = {
+    def createValueSecret(data: String): Secret = {
+      Secret.newBuilder()
+        .setType(Secret.Type.VALUE)
+        .setValue(Secret.Value.newBuilder().setData(ByteString.copyFrom(data.getBytes)))
+        .build()
+    }
+
+    def createReferenceSecret(name: String): Secret = {
+      Secret.newBuilder()
+        .setReference(Secret.Reference.newBuilder().setName(name))
+        .setType(Secret.Type.REFERENCE)
+        .build()
+    }
+
+    val referenceSecrets: Seq[Secret] =
+      conf.get(secretConfig.SECRET_NAME).getOrElse(Nil).map(s => createReferenceSecret(s))
+
+    val valueSecrets: Seq[Secret] = {
+      conf.get(secretConfig.SECRET_VALUE).getOrElse(Nil).map(s => createValueSecret(s))
+    }
+
+    if (valueSecrets.nonEmpty && referenceSecrets.nonEmpty) {
+      throw new SparkException("Cannot specify VALUE type secrets and REFERENCE types ones")
+    }
+
+    if (referenceSecrets.nonEmpty) referenceSecrets else valueSecrets
+  }
+
+  private def illegalSecretInput(dest: Seq[String], s: Seq[Secret]): Boolean = {
+    if (dest.isEmpty) {  // no destination set (ie not using secrets of this type
+      return false
+    }
+    if (dest.nonEmpty && s.nonEmpty) {
+      // make sure there is a destination for each secret of this type
+      if (dest.length != s.length) {
+        return true
+      }
+    }
+    false
+  }
+
+  private def getSecretVolume(conf: SparkConf, secretConfig: MesosSecretConfig): List[Volume] = {
+    val secrets = getSecrets(conf, secretConfig)
+    val secretPaths: Seq[String] =
+      conf.get(secretConfig.SECRET_FILENAME).getOrElse(Nil)
+
+    if (illegalSecretInput(secretPaths, secrets)) {
+      throw new SparkException(
+        s"Need to give equal numbers of secrets and file paths for file-based " +
+          s"reference secrets got secrets $secrets, and paths $secretPaths")
+    }
+
+    secrets.zip(secretPaths).map {
+      case (s, p) =>
+        val source = Volume.Source.newBuilder()
+          .setType(Volume.Source.Type.SECRET)
+          .setSecret(s)
+        Volume.newBuilder()
+          .setContainerPath(p)
+          .setSource(source)
+          .setMode(Volume.Mode.RO)
+          .build
+    }.toList
+  }
+
+  private def getSecretEnvVar(conf: SparkConf, secretConfig: MesosSecretConfig):
+  List[Variable] = {
+    val secrets = getSecrets(conf, secretConfig)
+    val secretEnvKeys = conf.get(secretConfig.SECRET_ENVKEY).getOrElse(Nil)
+    if (illegalSecretInput(secretEnvKeys, secrets)) {
+      throw new SparkException(
+        s"Need to give equal numbers of secrets and environment keys " +
+          s"for environment-based reference secrets got secrets $secrets, " +
+          s"and keys $secretEnvKeys")
+    }
+
+    secrets.zip(secretEnvKeys).map {
+      case (s, k) =>
+        Variable.newBuilder()
+          .setName(k)
+          .setType(Variable.Type.SECRET)
+          .setSecret(s)
+          .build
+    }.toList
+  }
+
   private def dockerInfo(
       image: String,
       forcePullImage: Boolean,
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 ff63e3f4ccfc..77acee608f25 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
@@ -24,7 +24,6 @@ import scala.collection.JavaConverters._
 import org.apache.mesos.Protos.{Environment, Secret, TaskState => MesosTaskState, _}
 import org.apache.mesos.Protos.Value.{Scalar, Type}
 import org.apache.mesos.SchedulerDriver
-import org.apache.mesos.protobuf.ByteString
 import org.mockito.{ArgumentCaptor, Matchers}
 import org.mockito.Mockito._
 import org.scalatest.mockito.MockitoSugar
@@ -32,6 +31,7 @@ import org.scalatest.mockito.MockitoSugar
 import org.apache.spark.{LocalSparkContext, SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.Command
 import org.apache.spark.deploy.mesos.MesosDriverDescription
+import org.apache.spark.deploy.mesos.config
 
 class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext with MockitoSugar {
 
@@ -341,132 +341,33 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
   }
 
   test("Creates an env-based reference secrets.") {
-    setScheduler()
-
-    val mem = 1000
-    val cpu = 1
-    val secretName = "/path/to/secret,/anothersecret"
-    val envKey = "SECRET_ENV_KEY,PASSWORD"
-    val driverDesc = new MesosDriverDescription(
-      "d1",
-      "jar",
-      mem,
-      cpu,
-      true,
-      command,
-      Map("spark.mesos.executor.home" -> "test",
-        "spark.app.name" -> "test",
-        "spark.mesos.driver.secret.names" -> secretName,
-        "spark.mesos.driver.secret.envkeys" -> envKey),
-      "s1",
-      new Date())
-    val response = scheduler.submitDriver(driverDesc)
-    assert(response.success)
-    val offer = Utils.createOffer("o1", "s1", mem, cpu)
-    scheduler.resourceOffers(driver, Collections.singletonList(offer))
-    val launchedTasks = Utils.verifyTaskLaunched(driver, "o1")
-    assert(launchedTasks.head
-      .getCommand
-      .getEnvironment
-      .getVariablesCount == 3)  // SPARK_SUBMIT_OPS and the secret
-    val variableOne = launchedTasks.head.getCommand.getEnvironment
-      .getVariablesList.asScala.filter(_.getName == "SECRET_ENV_KEY").head
-    assert(variableOne.getSecret.isInitialized)
-    assert(variableOne.getSecret.getType == Secret.Type.REFERENCE)
-    assert(variableOne.getSecret.getReference.getName == "/path/to/secret")
-    assert(variableOne.getType == Environment.Variable.Type.SECRET)
-    val variableTwo = launchedTasks.head.getCommand.getEnvironment
-      .getVariablesList.asScala.filter(_.getName == "PASSWORD").head
-    assert(variableTwo.getSecret.isInitialized)
-    assert(variableTwo.getSecret.getType == Secret.Type.REFERENCE)
-    assert(variableTwo.getSecret.getReference.getName == "/anothersecret")
-    assert(variableTwo.getType == Environment.Variable.Type.SECRET)
+    val launchedTasks = launchDriverTask(
+      Utils.configEnvBasedRefSecrets(config.driverSecretConfig))
+    Utils.verifyEnvBasedRefSecrets(launchedTasks)
   }
 
   test("Creates an env-based value secrets.") {
-    setScheduler()
-    val mem = 1000
-    val cpu = 1
-    val secretValues = "user,password"
-    val envKeys = "USER,PASSWORD"
-    val driverDesc = new MesosDriverDescription(
-      "d1",
-      "jar",
-      mem,
-      cpu,
-      true,
-      command,
-      Map("spark.mesos.executor.home" -> "test",
-        "spark.app.name" -> "test",
-        "spark.mesos.driver.secret.values" -> secretValues,
-        "spark.mesos.driver.secret.envkeys" -> envKeys),
-      "s1",
-      new Date())
-    val response = scheduler.submitDriver(driverDesc)
-    assert(response.success)
-    val offer = Utils.createOffer("o1", "s1", mem, cpu)
-    scheduler.resourceOffers(driver, Collections.singletonList(offer))
-    val launchedTasks = Utils.verifyTaskLaunched(driver, "o1")
-    assert(launchedTasks.head
-      .getCommand
-      .getEnvironment
-      .getVariablesCount == 3)  // SPARK_SUBMIT_OPS and the secret
-    val variableOne = launchedTasks.head.getCommand.getEnvironment
-      .getVariablesList.asScala.filter(_.getName == "USER").head
-    assert(variableOne.getSecret.isInitialized)
-    assert(variableOne.getSecret.getType == Secret.Type.VALUE)
-    assert(variableOne.getSecret.getValue.getData == ByteString.copyFrom("user".getBytes))
-    assert(variableOne.getType == Environment.Variable.Type.SECRET)
-    val variableTwo = launchedTasks.head.getCommand.getEnvironment
-      .getVariablesList.asScala.filter(_.getName == "PASSWORD").head
-    assert(variableTwo.getSecret.isInitialized)
-    assert(variableTwo.getSecret.getType == Secret.Type.VALUE)
-    assert(variableTwo.getSecret.getValue.getData == ByteString.copyFrom("password".getBytes))
-    assert(variableTwo.getType == Environment.Variable.Type.SECRET)
+    val launchedTasks = launchDriverTask(
+      Utils.configEnvBasedValueSecrets(config.driverSecretConfig))
+    Utils.verifyEnvBasedValueSecrets(launchedTasks)
   }
 
   test("Creates file-based reference secrets.") {
-    setScheduler()
-    val mem = 1000
-    val cpu = 1
-    val secretName = "/path/to/secret,/anothersecret"
-    val secretPath = "/topsecret,/mypassword"
-    val driverDesc = new MesosDriverDescription(
-      "d1",
-      "jar",
-      mem,
-      cpu,
-      true,
-      command,
-      Map("spark.mesos.executor.home" -> "test",
-        "spark.app.name" -> "test",
-        "spark.mesos.driver.secret.names" -> secretName,
-        "spark.mesos.driver.secret.filenames" -> secretPath),
-      "s1",
-      new Date())
-    val response = scheduler.submitDriver(driverDesc)
-    assert(response.success)
-    val offer = Utils.createOffer("o1", "s1", mem, cpu)
-    scheduler.resourceOffers(driver, Collections.singletonList(offer))
-    val launchedTasks = Utils.verifyTaskLaunched(driver, "o1")
-    val volumes = launchedTasks.head.getContainer.getVolumesList
-    assert(volumes.size() == 2)
-    val secretVolOne = volumes.get(0)
-    assert(secretVolOne.getContainerPath == "/topsecret")
-    assert(secretVolOne.getSource.getSecret.getType == Secret.Type.REFERENCE)
-    assert(secretVolOne.getSource.getSecret.getReference.getName == "/path/to/secret")
-    val secretVolTwo = volumes.get(1)
-    assert(secretVolTwo.getContainerPath == "/mypassword")
-    assert(secretVolTwo.getSource.getSecret.getType == Secret.Type.REFERENCE)
-    assert(secretVolTwo.getSource.getSecret.getReference.getName == "/anothersecret")
+    val launchedTasks = launchDriverTask(
+      Utils.configFileBasedRefSecrets(config.driverSecretConfig))
+    Utils.verifyFileBasedRefSecrets(launchedTasks)
   }
 
   test("Creates a file-based value secrets.") {
+    val launchedTasks = launchDriverTask(
+      Utils.configFileBasedValueSecrets(config.driverSecretConfig))
+    Utils.verifyFileBasedValueSecrets(launchedTasks)
+  }
+
+  private def launchDriverTask(addlSparkConfVars: Map[String, String]): List[TaskInfo] = {
     setScheduler()
     val mem = 1000
     val cpu = 1
-    val secretValues = "user,password"
-    val secretPath = "/whoami,/mypassword"
     val driverDesc = new MesosDriverDescription(
       "d1",
       "jar",
@@ -475,27 +376,14 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
       true,
       command,
       Map("spark.mesos.executor.home" -> "test",
-        "spark.app.name" -> "test",
-        "spark.mesos.driver.secret.values" -> secretValues,
-        "spark.mesos.driver.secret.filenames" -> secretPath),
+        "spark.app.name" -> "test") ++
+        addlSparkConfVars,
       "s1",
       new Date())
     val response = scheduler.submitDriver(driverDesc)
     assert(response.success)
     val offer = Utils.createOffer("o1", "s1", mem, cpu)
     scheduler.resourceOffers(driver, Collections.singletonList(offer))
-    val launchedTasks = Utils.verifyTaskLaunched(driver, "o1")
-    val volumes = launchedTasks.head.getContainer.getVolumesList
-    assert(volumes.size() == 2)
-    val secretVolOne = volumes.get(0)
-    assert(secretVolOne.getContainerPath == "/whoami")
-    assert(secretVolOne.getSource.getSecret.getType == Secret.Type.VALUE)
-    assert(secretVolOne.getSource.getSecret.getValue.getData ==
-      ByteString.copyFrom("user".getBytes))
-    val secretVolTwo = volumes.get(1)
-    assert(secretVolTwo.getContainerPath == "/mypassword")
-    assert(secretVolTwo.getSource.getSecret.getType == Secret.Type.VALUE)
-    assert(secretVolTwo.getSource.getSecret.getValue.getData ==
-      ByteString.copyFrom("password".getBytes))
+    Utils.verifyTaskLaunched(driver, "o1")
   }
 }
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 6c40792112f4..f4bd1ee9da6f 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
@@ -21,7 +21,6 @@ import java.util.concurrent.TimeUnit
 
 import scala.collection.JavaConverters._
 import scala.concurrent.duration._
-import scala.reflect.ClassTag
 
 import org.apache.mesos.{Protos, Scheduler, SchedulerDriver}
 import org.apache.mesos.Protos._
@@ -38,7 +37,7 @@ import org.apache.spark.internal.config._
 import org.apache.spark.network.shuffle.mesos.MesosExternalShuffleClient
 import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef}
 import org.apache.spark.scheduler.TaskSchedulerImpl
-import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor, RemoveExecutor}
+import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RegisterExecutor}
 import org.apache.spark.scheduler.cluster.mesos.Utils._
 
 class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
@@ -653,6 +652,37 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
     offerResourcesAndVerify(2, true)
   }
 
+  test("Creates an env-based reference secrets.") {
+    val launchedTasks = launchExecutorTasks(configEnvBasedRefSecrets(executorSecretConfig))
+    verifyEnvBasedRefSecrets(launchedTasks)
+  }
+
+  test("Creates an env-based value secrets.") {
+    val launchedTasks = launchExecutorTasks(configEnvBasedValueSecrets(executorSecretConfig))
+    verifyEnvBasedValueSecrets(launchedTasks)
+  }
+
+  test("Creates file-based reference secrets.") {
+    val launchedTasks = launchExecutorTasks(configFileBasedRefSecrets(executorSecretConfig))
+    verifyFileBasedRefSecrets(launchedTasks)
+  }
+
+  test("Creates a file-based value secrets.") {
+    val launchedTasks = launchExecutorTasks(configFileBasedValueSecrets(executorSecretConfig))
+    verifyFileBasedValueSecrets(launchedTasks)
+  }
+
+  private def launchExecutorTasks(sparkConfVars: Map[String, String]): List[TaskInfo] = {
+    setBackend(sparkConfVars)
+
+    val (mem, cpu) = (backend.executorMemory(sc), 4)
+
+    val offer1 = createOffer("o1", "s1", mem, cpu)
+    backend.resourceOffers(driver, List(offer1).asJava)
+
+    verifyTaskLaunched(driver, "o1")
+  }
+
   private case class Resources(mem: Int, cpus: Int, gpus: Int = 0)
 
   private def registerMockExecutor(executorId: String, slaveId: String, cores: Integer) = {
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
index f49d7c29eda4..c4954d99c720 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
@@ -18,6 +18,7 @@
 package org.apache.spark.scheduler.cluster.mesos
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.mesos.config
 
 class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
 
@@ -26,7 +27,7 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.parameters", "a,b")
     conf.set("spark.mesos.executor.docker.image", "test")
 
-    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf)
+    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf, config.executorSecretConfig)
     val params = containerInfo.getDocker.getParametersList
 
     assert(params.size() == 0)
@@ -37,7 +38,7 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.parameters", "a=1,b=2,c=3")
     conf.set("spark.mesos.executor.docker.image", "test")
 
-    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf)
+    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf, config.executorSecretConfig)
     val params = containerInfo.getDocker.getParametersList
     assert(params.size() == 3)
     assert(params.get(0).getKey == "a")
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
index 833db0c1ff33..590a515c557d 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
@@ -24,9 +24,12 @@ import scala.collection.JavaConverters._
 import org.apache.mesos.Protos._
 import org.apache.mesos.Protos.Value.{Range => MesosRange, Ranges, Scalar}
 import org.apache.mesos.SchedulerDriver
+import org.apache.mesos.protobuf.ByteString
 import org.mockito.{ArgumentCaptor, Matchers}
 import org.mockito.Mockito._
 
+import org.apache.spark.deploy.mesos.MesosSecretConfig
+
 object Utils {
 
   val TEST_FRAMEWORK_ID = FrameworkID.newBuilder()
@@ -105,4 +108,108 @@ object Utils {
   def createTaskId(taskId: String): TaskID = {
     TaskID.newBuilder().setValue(taskId).build()
   }
+
+  def configEnvBasedRefSecrets(secretConfig: MesosSecretConfig): Map[String, String] = {
+    val secretName = "/path/to/secret,/anothersecret"
+    val envKey = "SECRET_ENV_KEY,PASSWORD"
+    Map(
+      secretConfig.SECRET_NAME.key -> secretName,
+      secretConfig.SECRET_ENVKEY.key -> envKey
+    )
+  }
+
+  def verifyEnvBasedRefSecrets(launchedTasks: List[TaskInfo]): Unit = {
+    val envVars = launchedTasks.head
+      .getCommand
+      .getEnvironment
+      .getVariablesList
+      .asScala
+    assert(envVars
+      .filter(!_.getName.startsWith("SPARK_")).length == 2)  // user-defined secret env vars
+    val variableOne = envVars.filter(_.getName == "SECRET_ENV_KEY").head
+    assert(variableOne.getSecret.isInitialized)
+    assert(variableOne.getSecret.getType == Secret.Type.REFERENCE)
+    assert(variableOne.getSecret.getReference.getName == "/path/to/secret")
+    assert(variableOne.getType == Environment.Variable.Type.SECRET)
+    val variableTwo = envVars.filter(_.getName == "PASSWORD").head
+    assert(variableTwo.getSecret.isInitialized)
+    assert(variableTwo.getSecret.getType == Secret.Type.REFERENCE)
+    assert(variableTwo.getSecret.getReference.getName == "/anothersecret")
+    assert(variableTwo.getType == Environment.Variable.Type.SECRET)
+  }
+
+  def configEnvBasedValueSecrets(secretConfig: MesosSecretConfig): Map[String, String] = {
+    val secretValues = "user,password"
+    val envKeys = "USER,PASSWORD"
+    Map(
+      secretConfig.SECRET_VALUE.key -> secretValues,
+      secretConfig.SECRET_ENVKEY.key -> envKeys
+    )
+  }
+
+  def verifyEnvBasedValueSecrets(launchedTasks: List[TaskInfo]): Unit = {
+    val envVars = launchedTasks.head
+      .getCommand
+      .getEnvironment
+      .getVariablesList
+      .asScala
+    assert(envVars
+      .filter(!_.getName.startsWith("SPARK_")).length == 2)  // user-defined secret env vars
+    val variableOne = envVars.filter(_.getName == "USER").head
+    assert(variableOne.getSecret.isInitialized)
+    assert(variableOne.getSecret.getType == Secret.Type.VALUE)
+    assert(variableOne.getSecret.getValue.getData == ByteString.copyFrom("user".getBytes))
+    assert(variableOne.getType == Environment.Variable.Type.SECRET)
+    val variableTwo = envVars.filter(_.getName == "PASSWORD").head
+    assert(variableTwo.getSecret.isInitialized)
+    assert(variableTwo.getSecret.getType == Secret.Type.VALUE)
+    assert(variableTwo.getSecret.getValue.getData == ByteString.copyFrom("password".getBytes))
+    assert(variableTwo.getType == Environment.Variable.Type.SECRET)
+  }
+
+  def configFileBasedRefSecrets(secretConfig: MesosSecretConfig): Map[String, String] = {
+    val secretName = "/path/to/secret,/anothersecret"
+    val secretPath = "/topsecret,/mypassword"
+    Map(
+      secretConfig.SECRET_NAME.key -> secretName,
+      secretConfig.SECRET_FILENAME.key -> secretPath
+    )
+  }
+
+  def verifyFileBasedRefSecrets(launchedTasks: List[TaskInfo]): Unit = {
+    val volumes = launchedTasks.head.getContainer.getVolumesList
+    assert(volumes.size() == 2)
+    val secretVolOne = volumes.get(0)
+    assert(secretVolOne.getContainerPath == "/topsecret")
+    assert(secretVolOne.getSource.getSecret.getType == Secret.Type.REFERENCE)
+    assert(secretVolOne.getSource.getSecret.getReference.getName == "/path/to/secret")
+    val secretVolTwo = volumes.get(1)
+    assert(secretVolTwo.getContainerPath == "/mypassword")
+    assert(secretVolTwo.getSource.getSecret.getType == Secret.Type.REFERENCE)
+    assert(secretVolTwo.getSource.getSecret.getReference.getName == "/anothersecret")
+  }
+
+  def configFileBasedValueSecrets(secretConfig: MesosSecretConfig): Map[String, String] = {
+    val secretValues = "user,password"
+    val secretPath = "/whoami,/mypassword"
+    Map(
+      secretConfig.SECRET_VALUE.key -> secretValues,
+      secretConfig.SECRET_FILENAME.key -> secretPath
+    )
+  }
+
+  def verifyFileBasedValueSecrets(launchedTasks: List[TaskInfo]): Unit = {
+    val volumes = launchedTasks.head.getContainer.getVolumesList
+    assert(volumes.size() == 2)
+    val secretVolOne = volumes.get(0)
+    assert(secretVolOne.getContainerPath == "/whoami")
+    assert(secretVolOne.getSource.getSecret.getType == Secret.Type.VALUE)
+    assert(secretVolOne.getSource.getSecret.getValue.getData ==
+      ByteString.copyFrom("user".getBytes))
+    val secretVolTwo = volumes.get(1)
+    assert(secretVolTwo.getContainerPath == "/mypassword")
+    assert(secretVolTwo.getSource.getSecret.getType == Secret.Type.VALUE)
+    assert(secretVolTwo.getSource.getSecret.getValue.getData ==
+      ByteString.copyFrom("password".getBytes))
+  }
 }

From 6f062c00f6382d266619b4a56a753ec27d1db10b Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Thu, 5 Oct 2017 08:07:20 -0400
Subject: [PATCH 2/8] [SPARK-22131] Updated docs

---
 docs/running-on-mesos.md | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index e0944bc9f5f8..0ef59e33c0b5 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -521,6 +521,43 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
 </tr>
 
+<tr>
+  <td><code>spark.mesos.executor.secret.envkeys</code></td>
+  <td><code>(none)</code></td>
+  <td>
+    A comma-separated list that, if set, the contents of the secret referenced
+    by spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
+    set to the provided environment variable in the executor's process.
+  </td>
+  </tr>
+  <tr>
+<td><code>spark.mesos.executor.secret.filenames</code></td>
+  <td><code>(none)</code></td>
+  <td>
+    A comma-separated list that, if set, the contents of the secret referenced by
+    spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
+    written to the provided file. Paths are relative to the container's work
+    directory.  Absolute paths must already exist.  Consult the Mesos Secret
+    protobuf for more information.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.mesos.executor.secret.names</code></td>
+  <td><code>(none)</code></td>
+  <td>
+    A comma-separated list of secret references. Consult the Mesos Secret
+    protobuf for more information.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.mesos.executor.secret.values</code></td>
+  <td><code>(none)</code></td>
+  <td>
+    A comma-separated list of secret values. Consult the Mesos Secret
+    protobuf for more information.
+  </td>
+</tr>
+
 <tr>
   <td><code>spark.mesos.driverEnv.[EnvironmentVariableName]</code></td>
   <td><code>(none)</code></td>

From 73b2cbfeb43a3448a570c055d62c439e267490be Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Mon, 9 Oct 2017 13:48:41 -0700
Subject: [PATCH 3/8] Addressed review comments: (1) pluralized SECRET_xyz
 config entries, (2) renamed 'containerInfo' -> 'buildContainerInfo', (3)
 added additional case to illegalSecretInput() check.

---
 .../apache/spark/deploy/mesos/config.scala    |  8 +++----
 .../cluster/mesos/MesosClusterScheduler.scala |  3 ++-
 .../MesosCoarseGrainedSchedulerBackend.scala  |  3 ++-
 .../MesosFineGrainedSchedulerBackend.scala    |  2 +-
 .../mesos/MesosSchedulerBackendUtil.scala     | 24 +++++++++----------
 .../MesosSchedulerBackendUtilSuite.scala      |  6 +++--
 .../spark/scheduler/cluster/mesos/Utils.scala | 20 ++++++++--------
 7 files changed, 34 insertions(+), 32 deletions(-)

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 1d1ff423f11f..6d42c4888d94 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
@@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit
 import org.apache.spark.internal.config.ConfigBuilder
 
 private[spark] class MesosSecretConfig(taskType: String) {
-  private[spark] val SECRET_NAME =
+  private[spark] val SECRET_NAMES =
     ConfigBuilder(s"spark.mesos.$taskType.secret.names")
       .doc("A comma-separated list of secret reference names. Consult the Mesos Secret " +
         "protobuf for more information.")
@@ -30,14 +30,14 @@ private[spark] class MesosSecretConfig(taskType: String) {
       .toSequence
       .createOptional
 
-  private[spark] val SECRET_VALUE =
+  private[spark] val SECRET_VALUES =
     ConfigBuilder(s"spark.mesos.$taskType.secret.values")
       .doc("A comma-separated list of secret values.")
       .stringConf
       .toSequence
       .createOptional
 
-  private[spark] val SECRET_ENVKEY =
+  private[spark] val SECRET_ENVKEYS =
     ConfigBuilder(s"spark.mesos.$taskType.secret.envkeys")
       .doc("A comma-separated list of the environment variables to contain the secrets." +
         "The environment variable will be set on the driver.")
@@ -45,7 +45,7 @@ private[spark] class MesosSecretConfig(taskType: String) {
       .toSequence
       .createOptional
 
-  private[spark] val SECRET_FILENAME =
+  private[spark] val SECRET_FILENAMES =
     ConfigBuilder(s"spark.mesos.$taskType.secret.filenames")
       .doc("A comma-separated list of file paths secret will be written to.  Consult the Mesos " +
         "Secret protobuf for more information.")
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index 47b7ca184d51..2d1ee682ee0b 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -543,7 +543,8 @@ private[spark] class MesosClusterScheduler(
       .setName(s"Driver for ${appName}")
       .setSlaveId(offer.offer.getSlaveId)
       .setCommand(buildDriverCommand(desc))
-      .setContainer(MesosSchedulerBackendUtil.containerInfo(desc.conf, config.driverSecretConfig))
+      .setContainer(MesosSchedulerBackendUtil.buildContainerInfo(
+        desc.conf, config.driverSecretConfig))
       .addAllResources(cpuResourcesToUse.asJava)
       .addAllResources(memResourcesToUse.asJava)
       .setLabels(driverLabels)
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 8982042530dd..8586686c2fc7 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
@@ -460,7 +460,8 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
             .setName(s"${sc.appName} $taskId")
             .setLabels(MesosProtoUtils.mesosLabels(taskLabels))
             .addAllResources(resourcesToUse.asJava)
-            .setContainer(MesosSchedulerBackendUtil.containerInfo(sc.conf, executorSecretConfig))
+            .setContainer(MesosSchedulerBackendUtil.buildContainerInfo(
+              sc.conf, executorSecretConfig))
 
           tasks(offer.getId) ::= taskBuilder.build()
           remainingResources(offerId) = resourcesLeft.asJava
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
index 103aad800a15..224dc2af2a00 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
@@ -161,7 +161,7 @@ private[spark] class MesosFineGrainedSchedulerBackend(
       .setData(ByteString.copyFrom(createExecArg()))
 
     executorInfo.setContainer(
-      MesosSchedulerBackendUtil.containerInfo(sc.conf, config.executorSecretConfig))
+      MesosSchedulerBackendUtil.buildContainerInfo(sc.conf, config.executorSecretConfig))
     (executorInfo.build(), resourcesAfterMem.asJava)
   }
 
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 ec38c25a99f0..c2c505740f55 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
@@ -17,7 +17,7 @@
 
 package org.apache.spark.scheduler.cluster.mesos
 
-import org.apache.mesos.Protos._
+import org.apache.mesos.Protos.{ContainerInfo, Environment, Image, NetworkInfo, Parameter, Secret, Volume}
 import org.apache.mesos.Protos.ContainerInfo.{DockerInfo, MesosInfo}
 import org.apache.mesos.Protos.Environment.Variable
 import org.apache.mesos.protobuf.ByteString
@@ -126,7 +126,8 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     .toList
   }
 
-  def containerInfo(conf: SparkConf, secretConfig: MesosSecretConfig): ContainerInfo.Builder = {
+  def buildContainerInfo(conf: SparkConf, secretConfig: MesosSecretConfig):
+  ContainerInfo.Builder = {
     val containerType = if (conf.contains("spark.mesos.executor.docker.image") &&
       conf.get("spark.mesos.containerizer", "docker") == "docker") {
       ContainerInfo.Type.DOCKER
@@ -219,10 +220,10 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     }
 
     val referenceSecrets: Seq[Secret] =
-      conf.get(secretConfig.SECRET_NAME).getOrElse(Nil).map(s => createReferenceSecret(s))
+      conf.get(secretConfig.SECRET_NAMES).getOrElse(Nil).map(s => createReferenceSecret(s))
 
     val valueSecrets: Seq[Secret] = {
-      conf.get(secretConfig.SECRET_VALUE).getOrElse(Nil).map(s => createValueSecret(s))
+      conf.get(secretConfig.SECRET_VALUES).getOrElse(Nil).map(s => createValueSecret(s))
     }
 
     if (valueSecrets.nonEmpty && referenceSecrets.nonEmpty) {
@@ -232,13 +233,10 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     if (referenceSecrets.nonEmpty) referenceSecrets else valueSecrets
   }
 
-  private def illegalSecretInput(dest: Seq[String], s: Seq[Secret]): Boolean = {
-    if (dest.isEmpty) {  // no destination set (ie not using secrets of this type
-      return false
-    }
-    if (dest.nonEmpty && s.nonEmpty) {
-      // make sure there is a destination for each secret of this type
-      if (dest.length != s.length) {
+  private def illegalSecretInput(dest: Seq[String], secrets: Seq[Secret]): Boolean = {
+    if (dest.nonEmpty) {
+      // make sure there is a one-to-one correspondence between destinations and secrets
+      if (dest.length != secrets.length) {
         return true
       }
     }
@@ -248,7 +246,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
   private def getSecretVolume(conf: SparkConf, secretConfig: MesosSecretConfig): List[Volume] = {
     val secrets = getSecrets(conf, secretConfig)
     val secretPaths: Seq[String] =
-      conf.get(secretConfig.SECRET_FILENAME).getOrElse(Nil)
+      conf.get(secretConfig.SECRET_FILENAMES).getOrElse(Nil)
 
     if (illegalSecretInput(secretPaths, secrets)) {
       throw new SparkException(
@@ -272,7 +270,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
   private def getSecretEnvVar(conf: SparkConf, secretConfig: MesosSecretConfig):
   List[Variable] = {
     val secrets = getSecrets(conf, secretConfig)
-    val secretEnvKeys = conf.get(secretConfig.SECRET_ENVKEY).getOrElse(Nil)
+    val secretEnvKeys = conf.get(secretConfig.SECRET_ENVKEYS).getOrElse(Nil)
     if (illegalSecretInput(secretEnvKeys, secrets)) {
       throw new SparkException(
         s"Need to give equal numbers of secrets and environment keys " +
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
index c4954d99c720..da24349df5ee 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
@@ -27,7 +27,8 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.parameters", "a,b")
     conf.set("spark.mesos.executor.docker.image", "test")
 
-    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf, config.executorSecretConfig)
+    val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(
+      conf, config.executorSecretConfig)
     val params = containerInfo.getDocker.getParametersList
 
     assert(params.size() == 0)
@@ -38,7 +39,8 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.parameters", "a=1,b=2,c=3")
     conf.set("spark.mesos.executor.docker.image", "test")
 
-    val containerInfo = MesosSchedulerBackendUtil.containerInfo(conf, config.executorSecretConfig)
+    val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(
+      conf, config.executorSecretConfig)
     val params = containerInfo.getDocker.getParametersList
     assert(params.size() == 3)
     assert(params.get(0).getKey == "a")
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
index 590a515c557d..4ad04d271f48 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
@@ -113,8 +113,8 @@ object Utils {
     val secretName = "/path/to/secret,/anothersecret"
     val envKey = "SECRET_ENV_KEY,PASSWORD"
     Map(
-      secretConfig.SECRET_NAME.key -> secretName,
-      secretConfig.SECRET_ENVKEY.key -> envKey
+      secretConfig.SECRET_NAMES.key -> secretName,
+      secretConfig.SECRET_ENVKEYS.key -> envKey
     )
   }
 
@@ -125,7 +125,7 @@ object Utils {
       .getVariablesList
       .asScala
     assert(envVars
-      .filter(!_.getName.startsWith("SPARK_")).length == 2)  // user-defined secret env vars
+      .count(!_.getName.startsWith("SPARK_")) == 2)  // user-defined secret env vars
     val variableOne = envVars.filter(_.getName == "SECRET_ENV_KEY").head
     assert(variableOne.getSecret.isInitialized)
     assert(variableOne.getSecret.getType == Secret.Type.REFERENCE)
@@ -142,8 +142,8 @@ object Utils {
     val secretValues = "user,password"
     val envKeys = "USER,PASSWORD"
     Map(
-      secretConfig.SECRET_VALUE.key -> secretValues,
-      secretConfig.SECRET_ENVKEY.key -> envKeys
+      secretConfig.SECRET_VALUES.key -> secretValues,
+      secretConfig.SECRET_ENVKEYS.key -> envKeys
     )
   }
 
@@ -154,7 +154,7 @@ object Utils {
       .getVariablesList
       .asScala
     assert(envVars
-      .filter(!_.getName.startsWith("SPARK_")).length == 2)  // user-defined secret env vars
+      .count(!_.getName.startsWith("SPARK_")) == 2)  // user-defined secret env vars
     val variableOne = envVars.filter(_.getName == "USER").head
     assert(variableOne.getSecret.isInitialized)
     assert(variableOne.getSecret.getType == Secret.Type.VALUE)
@@ -171,8 +171,8 @@ object Utils {
     val secretName = "/path/to/secret,/anothersecret"
     val secretPath = "/topsecret,/mypassword"
     Map(
-      secretConfig.SECRET_NAME.key -> secretName,
-      secretConfig.SECRET_FILENAME.key -> secretPath
+      secretConfig.SECRET_NAMES.key -> secretName,
+      secretConfig.SECRET_FILENAMES.key -> secretPath
     )
   }
 
@@ -193,8 +193,8 @@ object Utils {
     val secretValues = "user,password"
     val secretPath = "/whoami,/mypassword"
     Map(
-      secretConfig.SECRET_VALUE.key -> secretValues,
-      secretConfig.SECRET_FILENAME.key -> secretPath
+      secretConfig.SECRET_VALUES.key -> secretValues,
+      secretConfig.SECRET_FILENAMES.key -> secretPath
     )
   }
 

From 770d307bd67796b93f20d7dba105905059af7a0e Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Tue, 17 Oct 2017 17:16:10 -0700
Subject: [PATCH 4/8] Addressed review comments: added examples to docs,
 reverted the change in FineGrainedSchedulerBackend, and moved the
 getSecretEnvVar and getSecretVolumes code blocks to be more consistent with
 rest of code.

---
 docs/running-on-mesos.md                      | 32 ++++++++++++++-----
 .../cluster/mesos/MesosClusterScheduler.scala | 31 ++++++++++++++++--
 .../MesosCoarseGrainedSchedulerBackend.scala  | 31 +++++++++++++++---
 .../MesosFineGrainedSchedulerBackend.scala    |  2 +-
 .../mesos/MesosSchedulerBackendUtil.scala     | 31 ++----------------
 .../MesosSchedulerBackendUtilSuite.scala      |  4 +--
 6 files changed, 85 insertions(+), 46 deletions(-)

diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index 0ef59e33c0b5..92c250d1cc32 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -490,7 +490,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td>
     A comma-separated list that, if set, the contents of the secret referenced
     by spark.mesos.driver.secret.names or spark.mesos.driver.secret.values will be
-    set to the provided environment variable in the driver's process.
+    set to the provided environment variable in the driver's process. Example:
+    
+    <pre>ENVKEY1,ENVKEY2</pre>
   </td>
   </tr>
   <tr>
@@ -501,7 +503,9 @@ See the [configuration page](configuration.html) for information on Spark config
     spark.mesos.driver.secret.names or spark.mesos.driver.secret.values will be
     written to the provided file. Paths are relative to the container's work
     directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>filename1,filename2</pre>
   </td>
 </tr>
 <tr>
@@ -509,7 +513,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>(none)</code></td>
   <td>
     A comma-separated list of secret references. Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>secretname1,secretname2</pre>
   </td>
 </tr>
 <tr>
@@ -517,7 +523,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>(none)</code></td>
   <td>
     A comma-separated list of secret values. Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>secretvalue1,secretvalue2</pre>
   </td>
 </tr>
 
@@ -527,7 +535,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td>
     A comma-separated list that, if set, the contents of the secret referenced
     by spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
-    set to the provided environment variable in the executor's process.
+    set to the provided environment variable in the executor's process. Example:
+    
+    <pre>ENVKEY1,ENVKEY2</pre>
   </td>
   </tr>
   <tr>
@@ -538,7 +548,9 @@ See the [configuration page](configuration.html) for information on Spark config
     spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
     written to the provided file. Paths are relative to the container's work
     directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>filename1,filename2</pre>
   </td>
 </tr>
 <tr>
@@ -546,7 +558,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>(none)</code></td>
   <td>
     A comma-separated list of secret references. Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>secretname1,secretname2</pre>
   </td>
 </tr>
 <tr>
@@ -554,7 +568,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>(none)</code></td>
   <td>
     A comma-separated list of secret values. Consult the Mesos Secret
-    protobuf for more information.
+    protobuf for more information. Example:
+    
+    <pre>secretvalue1,secretvalue2</pre>
   </td>
 </tr>
 
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index 2d1ee682ee0b..c84f517d8216 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -393,7 +393,16 @@ private[spark] class MesosClusterScheduler(
     }
 
     // add secret environment variables
-    MesosSchedulerBackendUtil.addSecretEnvVar(envBuilder, desc.conf, config.driverSecretConfig)
+    MesosSchedulerBackendUtil.getSecretEnvVar(desc.conf, config.driverSecretConfig)
+      .foreach { variable =>
+      if (variable.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName} " +
+          s"on file ${variable.getName}")
+      } else {
+        logInfo(s"Setting secret on environment variable name=${variable.getName}")
+      }
+      envBuilder.addVariables(variable)
+    }
 
     envBuilder.build()
   }
@@ -411,6 +420,23 @@ private[spark] class MesosClusterScheduler(
       CommandInfo.URI.newBuilder().setValue(uri.trim()).setCache(useFetchCache).build())
   }
 
+  private def getContainerInfo(desc: MesosDriverDescription): ContainerInfo.Builder = {
+    val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(desc.conf)
+
+    MesosSchedulerBackendUtil.getSecretVolume(desc.conf, config.driverSecretConfig)
+      .foreach { volume =>
+      if (volume.getSource.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName} " +
+          s"on file ${volume.getContainerPath}")
+      } else {
+        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
+      }
+      containerInfo.addVolumes(volume)
+    }
+
+    containerInfo
+  }
+
   private def getDriverCommandValue(desc: MesosDriverDescription): String = {
     val dockerDefined = desc.conf.contains("spark.mesos.executor.docker.image")
     val executorUri = getDriverExecutorURI(desc)
@@ -543,8 +569,7 @@ private[spark] class MesosClusterScheduler(
       .setName(s"Driver for ${appName}")
       .setSlaveId(offer.offer.getSlaveId)
       .setCommand(buildDriverCommand(desc))
-      .setContainer(MesosSchedulerBackendUtil.buildContainerInfo(
-        desc.conf, config.driverSecretConfig))
+      .setContainer(getContainerInfo(desc))
       .addAllResources(cpuResourcesToUse.asJava)
       .addAllResources(memResourcesToUse.asJava)
       .setLabels(driverLabels)
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 8586686c2fc7..ea02aa60788a 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
@@ -28,7 +28,7 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.concurrent.Future
 
-import org.apache.spark.{SecurityManager, SparkContext, SparkException, TaskState}
+import org.apache.spark.{SecurityManager, SparkConf, SparkContext, SparkException, TaskState}
 import org.apache.spark.deploy.mesos.config._
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
 import org.apache.spark.internal.config
@@ -233,7 +233,15 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
         .build())
     }
 
-    MesosSchedulerBackendUtil.addSecretEnvVar(environment, conf, executorSecretConfig)
+    MesosSchedulerBackendUtil.getSecretEnvVar(conf, executorSecretConfig).foreach { variable =>
+      if (variable.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName} " +
+          s"on file ${variable.getName}")
+      } else {
+        logInfo(s"Setting secret on environment variable name=${variable.getName}")
+      }
+      environment.addVariables(variable)
+    }
 
     val command = CommandInfo.newBuilder()
       .setEnvironment(environment)
@@ -409,6 +417,22 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
     }
   }
 
+  private def getContainerInfo(conf: SparkConf): ContainerInfo.Builder = {
+    val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(conf)
+
+    MesosSchedulerBackendUtil.getSecretVolume(conf, executorSecretConfig).foreach { volume =>
+      if (volume.getSource.getSecret.getReference.isInitialized) {
+        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName} " +
+          s"on file ${volume.getContainerPath}")
+      } else {
+        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
+      }
+      containerInfo.addVolumes(volume)
+    }
+
+    containerInfo
+  }
+
   /**
    * Returns a map from OfferIDs to the tasks to launch on those offers.  In order to maximize
    * per-task memory and IO, tasks are round-robin assigned to offers.
@@ -460,8 +484,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
             .setName(s"${sc.appName} $taskId")
             .setLabels(MesosProtoUtils.mesosLabels(taskLabels))
             .addAllResources(resourcesToUse.asJava)
-            .setContainer(MesosSchedulerBackendUtil.buildContainerInfo(
-              sc.conf, executorSecretConfig))
+            .setContainer(getContainerInfo(sc.conf))
 
           tasks(offer.getId) ::= taskBuilder.build()
           remainingResources(offerId) = resourcesLeft.asJava
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
index 224dc2af2a00..d6d939d24610 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosFineGrainedSchedulerBackend.scala
@@ -161,7 +161,7 @@ private[spark] class MesosFineGrainedSchedulerBackend(
       .setData(ByteString.copyFrom(createExecArg()))
 
     executorInfo.setContainer(
-      MesosSchedulerBackendUtil.buildContainerInfo(sc.conf, config.executorSecretConfig))
+      MesosSchedulerBackendUtil.buildContainerInfo(sc.conf))
     (executorInfo.build(), resourcesAfterMem.asJava)
   }
 
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 c2c505740f55..3d1d6e866671 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
@@ -126,7 +126,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     .toList
   }
 
-  def buildContainerInfo(conf: SparkConf, secretConfig: MesosSecretConfig):
+  def buildContainerInfo(conf: SparkConf):
   ContainerInfo.Builder = {
     val containerType = if (conf.contains("spark.mesos.executor.docker.image") &&
       conf.get("spark.mesos.containerizer", "docker") == "docker") {
@@ -175,34 +175,9 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
       containerInfo.addNetworkInfos(info)
     }
 
-    getSecretVolume(conf, secretConfig).foreach { volume =>
-      if (volume.getSource.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName}" +
-          s"on file ${volume.getContainerPath}")
-      } else {
-        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
-      }
-      containerInfo.addVolumes(volume)
-    }
-
     containerInfo
   }
 
-  def addSecretEnvVar(
-      envBuilder: Environment.Builder,
-      conf: SparkConf,
-      secretConfig: MesosSecretConfig): Unit = {
-    getSecretEnvVar(conf, secretConfig).foreach { variable =>
-      if (variable.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName}" +
-          s"on file ${variable.getName}")
-      } else {
-        logInfo(s"Setting secret on environment variable name=${variable.getName}")
-      }
-      envBuilder.addVariables(variable)
-    }
-  }
-
   private def getSecrets(conf: SparkConf, secretConfig: MesosSecretConfig):
   Seq[Secret] = {
     def createValueSecret(data: String): Secret = {
@@ -243,7 +218,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     false
   }
 
-  private def getSecretVolume(conf: SparkConf, secretConfig: MesosSecretConfig): List[Volume] = {
+  def getSecretVolume(conf: SparkConf, secretConfig: MesosSecretConfig): List[Volume] = {
     val secrets = getSecrets(conf, secretConfig)
     val secretPaths: Seq[String] =
       conf.get(secretConfig.SECRET_FILENAMES).getOrElse(Nil)
@@ -267,7 +242,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     }.toList
   }
 
-  private def getSecretEnvVar(conf: SparkConf, secretConfig: MesosSecretConfig):
+  def getSecretEnvVar(conf: SparkConf, secretConfig: MesosSecretConfig):
   List[Variable] = {
     val secrets = getSecrets(conf, secretConfig)
     val secretEnvKeys = conf.get(secretConfig.SECRET_ENVKEYS).getOrElse(Nil)
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
index da24349df5ee..442c43960ec1 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackendUtilSuite.scala
@@ -28,7 +28,7 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.image", "test")
 
     val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(
-      conf, config.executorSecretConfig)
+      conf)
     val params = containerInfo.getDocker.getParametersList
 
     assert(params.size() == 0)
@@ -40,7 +40,7 @@ class MesosSchedulerBackendUtilSuite extends SparkFunSuite {
     conf.set("spark.mesos.executor.docker.image", "test")
 
     val containerInfo = MesosSchedulerBackendUtil.buildContainerInfo(
-      conf, config.executorSecretConfig)
+      conf)
     val params = containerInfo.getDocker.getParametersList
     assert(params.size() == 3)
     assert(params.get(0).getKey == "a")

From b2a36753b41820ec5e2d85a6a29fd8677bc0029a Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Tue, 17 Oct 2017 17:24:05 -0700
Subject: [PATCH 5/8] Fixed formatting

---
 .../scheduler/cluster/mesos/MesosSchedulerBackendUtil.scala    | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

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 3d1d6e866671..05f44b9a0298 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
@@ -178,8 +178,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     containerInfo
   }
 
-  private def getSecrets(conf: SparkConf, secretConfig: MesosSecretConfig):
-  Seq[Secret] = {
+  private def getSecrets(conf: SparkConf, secretConfig: MesosSecretConfig): Seq[Secret] = {
     def createValueSecret(data: String): Secret = {
       Secret.newBuilder()
         .setType(Secret.Type.VALUE)

From a80179953b38ec5e7f015eadc1b48075f01e4b65 Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Wed, 18 Oct 2017 08:08:52 -0700
Subject: [PATCH 6/8] Mentioned need for custom module in file-based secrets
 docs; formatting.

---
 docs/running-on-mesos.md                                  | 8 ++++++--
 .../cluster/mesos/MesosSchedulerBackendUtil.scala         | 4 ++--
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index 92c250d1cc32..d28e8646057f 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -503,7 +503,9 @@ See the [configuration page](configuration.html) for information on Spark config
     spark.mesos.driver.secret.names or spark.mesos.driver.secret.values will be
     written to the provided file. Paths are relative to the container's work
     directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information. Example:
+    protobuf for more information. Note: File-based secrets require a custom
+    <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
+    module</a>. Example:
     
     <pre>filename1,filename2</pre>
   </td>
@@ -548,7 +550,9 @@ See the [configuration page](configuration.html) for information on Spark config
     spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
     written to the provided file. Paths are relative to the container's work
     directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information. Example:
+    protobuf for more information. Note: File-based secrets require a custom
+    <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
+    module</a>. Example:
     
     <pre>filename1,filename2</pre>
   </td>
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 05f44b9a0298..d35ed5c88c53 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
@@ -127,7 +127,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
   }
 
   def buildContainerInfo(conf: SparkConf):
-  ContainerInfo.Builder = {
+    ContainerInfo.Builder = {
     val containerType = if (conf.contains("spark.mesos.executor.docker.image") &&
       conf.get("spark.mesos.containerizer", "docker") == "docker") {
       ContainerInfo.Type.DOCKER
@@ -242,7 +242,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
   }
 
   def getSecretEnvVar(conf: SparkConf, secretConfig: MesosSecretConfig):
-  List[Variable] = {
+    List[Variable] = {
     val secrets = getSecrets(conf, secretConfig)
     val secretEnvKeys = conf.get(secretConfig.SECRET_ENVKEYS).getOrElse(Nil)
     if (illegalSecretInput(secretEnvKeys, secrets)) {

From 3cd1ae82e136b42059091213f21e4649d267cba4 Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Wed, 25 Oct 2017 08:51:16 -0700
Subject: [PATCH 7/8] Addressed review comments: grouped properties together in
 the documentation, added private constructor, fixed formatting.

---
 docs/running-on-mesos.md                      | 133 ++++++++----------
 .../apache/spark/deploy/mesos/config.scala    |  66 ++++-----
 .../cluster/mesos/MesosClusterScheduler.scala |  28 ++--
 .../mesos/MesosSchedulerBackendUtil.scala     |  37 +++--
 .../spark/scheduler/cluster/mesos/Utils.scala |   2 +-
 5 files changed, 127 insertions(+), 139 deletions(-)

diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index d28e8646057f..2dda282ed2c0 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -485,96 +485,87 @@ See the [configuration page](configuration.html) for information on Spark config
 </tr>
 
 <tr>
-  <td><code>spark.mesos.driver.secret.envkeys</code></td>
-  <td><code>(none)</code></td>
   <td>
-    A comma-separated list that, if set, the contents of the secret referenced
-    by spark.mesos.driver.secret.names or spark.mesos.driver.secret.values will be
-    set to the provided environment variable in the driver's process. Example:
-    
-    <pre>ENVKEY1,ENVKEY2</pre>
+    <code>spark.mesos.driver.secret.values</code>,
+    <code>spark.mesos.driver.secret.names</code>,
+    <code>spark.mesos.executor.secret.values</code>,
+    <code>spark.mesos.executor.secret.names</code>,
   </td>
-  </tr>
-  <tr>
-<td><code>spark.mesos.driver.secret.filenames</code></td>
   <td><code>(none)</code></td>
   <td>
-    A comma-separated list that, if set, the contents of the secret referenced by
-    spark.mesos.driver.secret.names or spark.mesos.driver.secret.values will be
-    written to the provided file. Paths are relative to the container's work
-    directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information. Note: File-based secrets require a custom
-    <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
-    module</a>. Example:
+    A secret is specified by its contents and destination. These properties
+    specify a secret's contents. To specify a secret's destination, see the cell below.
+
+    You can specify a secret's contents either (1) by value or (2) by reference.
+    (1) To specify a secret by value, set the
+    <code>spark.mesos.[driver|executor].secret.values</code>
+    property, to make the secret available in the driver or executors.
+    For example, to make a secret password "guessme" available to the driver process, set:
     
-    <pre>filename1,filename2</pre>
-  </td>
-</tr>
-<tr>
-  <td><code>spark.mesos.driver.secret.names</code></td>
-  <td><code>(none)</code></td>
-  <td>
-    A comma-separated list of secret references. Consult the Mesos Secret
-    protobuf for more information. Example:
+    <pre>spark.mesos.driver.secret.values=guessme</pre>
     
-    <pre>secretname1,secretname2</pre>
-  </td>
-</tr>
-<tr>
-  <td><code>spark.mesos.driver.secret.values</code></td>
-  <td><code>(none)</code></td>
-  <td>
-    A comma-separated list of secret values. Consult the Mesos Secret
-    protobuf for more information. Example:
+    (2) To specify a secret that has been placed in a secret store
+    by reference, specify its name within the secret store
+    by setting the <code>spark.mesos.[driver|executor].secret.names</code>
+    property. For example, to make a secret password named "password" in a secret store
+    available to the driver process, set:
+
+    <pre>spark.mesos.driver.secret.names=password</pre>
+
+    To specify multiple secrets, provide a comma-separated list:
+
+    <pre>spark.mesos.driver.secret.values=guessme,passwd123</pre>
+
+    or
+
+    <pre>spark.mesos.driver.secret.names=password1,password2</pre>
     
-    <pre>secretvalue1,secretvalue2</pre>
   </td>
 </tr>
 
 <tr>
-  <td><code>spark.mesos.executor.secret.envkeys</code></td>
-  <td><code>(none)</code></td>
   <td>
-    A comma-separated list that, if set, the contents of the secret referenced
-    by spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
-    set to the provided environment variable in the executor's process. Example:
-    
-    <pre>ENVKEY1,ENVKEY2</pre>
+    <code>spark.mesos.driver.secret.envkeys</code>,
+    <code>spark.mesos.driver.secret.filenames</code>,
+    <code>spark.mesos.executor.secret.envkeys</code>,
+    <code>spark.mesos.executor.secret.filenames</code>,
   </td>
-  </tr>
-  <tr>
-<td><code>spark.mesos.executor.secret.filenames</code></td>
   <td><code>(none)</code></td>
   <td>
-    A comma-separated list that, if set, the contents of the secret referenced by
-    spark.mesos.executor.secret.names or spark.mesos.executor.secret.values will be
-    written to the provided file. Paths are relative to the container's work
-    directory.  Absolute paths must already exist.  Consult the Mesos Secret
-    protobuf for more information. Note: File-based secrets require a custom
+    A secret is specified by its contents and destination. These properties
+    specify a secret's destination. To specify a secret's contents, see the cell above.
+
+    You can specify a secret's destination in the driver or
+    executors as either (1) an environment variable or (2) as a file.
+    (1) To make an environment-based secret, set the
+    <code>spark.mesos.[driver|executor].secret.envkeys</code> property.
+    The secret will appear as an environment variable with the
+    given name in the driver or executors. For example, to make a secret password available
+    to the driver process as $PASSWORD, set:
+
+    <pre>spark.mesos.driver.secret.envkeys=PASSWORD</pre>
+
+    (2) To make a file-based secret, set the
+    <code>spark.mesos.[driver|executor].secret.filenames</code> property.
+    The secret will appear in the contents of a file with the given file name in
+    the driver or executors. For example, to make a secret password available in a
+    file named "pwdfile" in the driver process, set:
+
+    <pre>spark.mesos.driver.secret.filenames=pwdfile</pre>
+
+    Paths are relative to the container's work directory. Absolute paths must
+    already exist. Note: File-based secrets require a custom
     <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
-    module</a>. Example:
+    module</a>.
     
-    <pre>filename1,filename2</pre>
-  </td>
-</tr>
-<tr>
-  <td><code>spark.mesos.executor.secret.names</code></td>
-  <td><code>(none)</code></td>
-  <td>
-    A comma-separated list of secret references. Consult the Mesos Secret
-    protobuf for more information. Example:
+    To specify env vars or file names corresponding to multiple secrets,
+    provide a comma-separated list:
+
+    <pre>spark.mesos.driver.secret.envkeys=PASSWORD1,PASSWORD2</pre>
     
-    <pre>secretname1,secretname2</pre>
-  </td>
-</tr>
-<tr>
-  <td><code>spark.mesos.executor.secret.values</code></td>
-  <td><code>(none)</code></td>
-  <td>
-    A comma-separated list of secret values. Consult the Mesos Secret
-    protobuf for more information. Example:
+    or
     
-    <pre>secretvalue1,secretvalue2</pre>
+    <pre>spark.mesos.driver.secret.filenames=pwdfile1,pwdfile2</pre>
   </td>
 </tr>
 
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 6d42c4888d94..821534eb4fc3 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
@@ -21,41 +21,41 @@ import java.util.concurrent.TimeUnit
 
 import org.apache.spark.internal.config.ConfigBuilder
 
-private[spark] class MesosSecretConfig(taskType: String) {
-  private[spark] val SECRET_NAMES =
-    ConfigBuilder(s"spark.mesos.$taskType.secret.names")
-      .doc("A comma-separated list of secret reference names. Consult the Mesos Secret " +
-        "protobuf for more information.")
-      .stringConf
-      .toSequence
-      .createOptional
-
-  private[spark] val SECRET_VALUES =
-    ConfigBuilder(s"spark.mesos.$taskType.secret.values")
-      .doc("A comma-separated list of secret values.")
-      .stringConf
-      .toSequence
-      .createOptional
-
-  private[spark] val SECRET_ENVKEYS =
-    ConfigBuilder(s"spark.mesos.$taskType.secret.envkeys")
-      .doc("A comma-separated list of the environment variables to contain the secrets." +
-        "The environment variable will be set on the driver.")
-      .stringConf
-      .toSequence
-      .createOptional
-
-  private[spark] val SECRET_FILENAMES =
-    ConfigBuilder(s"spark.mesos.$taskType.secret.filenames")
-      .doc("A comma-separated list of file paths secret will be written to.  Consult the Mesos " +
-        "Secret protobuf for more information.")
-      .stringConf
-      .toSequence
-      .createOptional
-}
-
 package object config {
 
+  private[spark] class MesosSecretConfig private[config](taskType: String) {
+    private[spark] val SECRET_NAMES =
+      ConfigBuilder(s"spark.mesos.$taskType.secret.names")
+        .doc("A comma-separated list of secret reference names. Consult the Mesos Secret " +
+          "protobuf for more information.")
+        .stringConf
+        .toSequence
+        .createOptional
+
+    private[spark] val SECRET_VALUES =
+      ConfigBuilder(s"spark.mesos.$taskType.secret.values")
+        .doc("A comma-separated list of secret values.")
+        .stringConf
+        .toSequence
+        .createOptional
+
+    private[spark] val SECRET_ENVKEYS =
+      ConfigBuilder(s"spark.mesos.$taskType.secret.envkeys")
+        .doc("A comma-separated list of the environment variables to contain the secrets." +
+          "The environment variable will be set on the driver.")
+        .stringConf
+        .toSequence
+        .createOptional
+
+    private[spark] val SECRET_FILENAMES =
+      ConfigBuilder(s"spark.mesos.$taskType.secret.filenames")
+        .doc("A comma-separated list of file paths secret will be written to.  Consult the Mesos " +
+          "Secret protobuf for more information.")
+        .stringConf
+        .toSequence
+        .createOptional
+  }
+
   /* Common app configuration. */
 
   private[spark] val SHUFFLE_CLEANER_INTERVAL_S =
diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index c84f517d8216..82470264f2a4 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -395,13 +395,13 @@ private[spark] class MesosClusterScheduler(
     // add secret environment variables
     MesosSchedulerBackendUtil.getSecretEnvVar(desc.conf, config.driverSecretConfig)
       .foreach { variable =>
-      if (variable.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName} " +
-          s"on file ${variable.getName}")
-      } else {
-        logInfo(s"Setting secret on environment variable name=${variable.getName}")
-      }
-      envBuilder.addVariables(variable)
+        if (variable.getSecret.getReference.isInitialized) {
+          logInfo(s"Setting reference secret ${variable.getSecret.getReference.getName} " +
+            s"on file ${variable.getName}")
+        } else {
+          logInfo(s"Setting secret on environment variable name=${variable.getName}")
+        }
+        envBuilder.addVariables(variable)
     }
 
     envBuilder.build()
@@ -425,13 +425,13 @@ private[spark] class MesosClusterScheduler(
 
     MesosSchedulerBackendUtil.getSecretVolume(desc.conf, config.driverSecretConfig)
       .foreach { volume =>
-      if (volume.getSource.getSecret.getReference.isInitialized) {
-        logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName} " +
-          s"on file ${volume.getContainerPath}")
-      } else {
-        logInfo(s"Setting secret on file name=${volume.getContainerPath}")
-      }
-      containerInfo.addVolumes(volume)
+        if (volume.getSource.getSecret.getReference.isInitialized) {
+          logInfo(s"Setting reference secret ${volume.getSource.getSecret.getReference.getName} " +
+            s"on file ${volume.getContainerPath}")
+        } else {
+          logInfo(s"Setting secret on file name=${volume.getContainerPath}")
+        }
+        containerInfo.addVolumes(volume)
     }
 
     containerInfo
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 d35ed5c88c53..3f703b9cd658 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
@@ -24,8 +24,8 @@ import org.apache.mesos.protobuf.ByteString
 
 import org.apache.spark.SparkConf
 import org.apache.spark.SparkException
-import org.apache.spark.deploy.mesos.MesosSecretConfig
 import org.apache.spark.deploy.mesos.config.{NETWORK_LABELS, NETWORK_NAME}
+import org.apache.spark.deploy.mesos.config.MesosSecretConfig
 import org.apache.spark.internal.Logging
 
 /**
@@ -126,8 +126,7 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     .toList
   }
 
-  def buildContainerInfo(conf: SparkConf):
-    ContainerInfo.Builder = {
+  def buildContainerInfo(conf: SparkConf): ContainerInfo.Builder = {
     val containerType = if (conf.contains("spark.mesos.executor.docker.image") &&
       conf.get("spark.mesos.containerizer", "docker") == "docker") {
       ContainerInfo.Type.DOCKER
@@ -228,16 +227,15 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
           s"reference secrets got secrets $secrets, and paths $secretPaths")
     }
 
-    secrets.zip(secretPaths).map {
-      case (s, p) =>
-        val source = Volume.Source.newBuilder()
-          .setType(Volume.Source.Type.SECRET)
-          .setSecret(s)
-        Volume.newBuilder()
-          .setContainerPath(p)
-          .setSource(source)
-          .setMode(Volume.Mode.RO)
-          .build
+    secrets.zip(secretPaths).map { case (s, p) =>
+      val source = Volume.Source.newBuilder()
+        .setType(Volume.Source.Type.SECRET)
+        .setSecret(s)
+      Volume.newBuilder()
+        .setContainerPath(p)
+        .setSource(source)
+        .setMode(Volume.Mode.RO)
+        .build
     }.toList
   }
 
@@ -252,13 +250,12 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
           s"and keys $secretEnvKeys")
     }
 
-    secrets.zip(secretEnvKeys).map {
-      case (s, k) =>
-        Variable.newBuilder()
-          .setName(k)
-          .setType(Variable.Type.SECRET)
-          .setSecret(s)
-          .build
+    secrets.zip(secretEnvKeys).map { case (s, k) =>
+      Variable.newBuilder()
+        .setName(k)
+        .setType(Variable.Type.SECRET)
+        .setSecret(s)
+        .build
     }.toList
   }
 
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
index 4ad04d271f48..5636ac52bd4a 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/Utils.scala
@@ -28,7 +28,7 @@ import org.apache.mesos.protobuf.ByteString
 import org.mockito.{ArgumentCaptor, Matchers}
 import org.mockito.Mockito._
 
-import org.apache.spark.deploy.mesos.MesosSecretConfig
+import org.apache.spark.deploy.mesos.config.MesosSecretConfig
 
 object Utils {
 

From 88dfb42bfd794ab1eea969c3b51970e68e4ca407 Mon Sep 17 00:00:00 2001
From: "Susan X. Huynh" <xhuynh@mesosphere.com>
Date: Thu, 26 Oct 2017 08:42:51 -0700
Subject: [PATCH 8/8] Fixed formatting in docs table cell, style.

---
 docs/running-on-mesos.md                      | 135 ++++++++++--------
 .../mesos/MesosSchedulerBackendUtil.scala     |   6 +-
 2 files changed, 80 insertions(+), 61 deletions(-)

diff --git a/docs/running-on-mesos.md b/docs/running-on-mesos.md
index 2dda282ed2c0..b7e3e6473c33 100644
--- a/docs/running-on-mesos.md
+++ b/docs/running-on-mesos.md
@@ -493,33 +493,44 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
   <td><code>(none)</code></td>
   <td>
-    A secret is specified by its contents and destination. These properties
-    specify a secret's contents. To specify a secret's destination, see the cell below.
-
-    You can specify a secret's contents either (1) by value or (2) by reference.
-    (1) To specify a secret by value, set the
-    <code>spark.mesos.[driver|executor].secret.values</code>
-    property, to make the secret available in the driver or executors.
-    For example, to make a secret password "guessme" available to the driver process, set:
-    
-    <pre>spark.mesos.driver.secret.values=guessme</pre>
-    
-    (2) To specify a secret that has been placed in a secret store
-    by reference, specify its name within the secret store
-    by setting the <code>spark.mesos.[driver|executor].secret.names</code>
-    property. For example, to make a secret password named "password" in a secret store
-    available to the driver process, set:
-
-    <pre>spark.mesos.driver.secret.names=password</pre>
-
-    To specify multiple secrets, provide a comma-separated list:
-
-    <pre>spark.mesos.driver.secret.values=guessme,passwd123</pre>
-
-    or
-
-    <pre>spark.mesos.driver.secret.names=password1,password2</pre>
+    <p>
+      A secret is specified by its contents and destination. These properties
+      specify a secret's contents. To specify a secret's destination, see the cell below.
+    </p>
+    <p>
+      You can specify a secret's contents either (1) by value or (2) by reference.
+    </p>
+    <p>
+      (1) To specify a secret by value, set the
+      <code>spark.mesos.[driver|executor].secret.values</code>
+      property, to make the secret available in the driver or executors.
+      For example, to make a secret password "guessme" available to the driver process, set:
     
+      <pre>spark.mesos.driver.secret.values=guessme</pre>
+    </p>
+    <p>
+      (2) To specify a secret that has been placed in a secret store
+      by reference, specify its name within the secret store
+      by setting the <code>spark.mesos.[driver|executor].secret.names</code>
+      property. For example, to make a secret password named "password" in a secret store
+      available to the driver process, set:
+
+      <pre>spark.mesos.driver.secret.names=password</pre>
+    </p>
+    <p>
+      Note: To use a secret store, make sure one has been integrated with Mesos via a custom
+      <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
+      module</a>.
+    </p>
+    <p>
+      To specify multiple secrets, provide a comma-separated list:
+
+      <pre>spark.mesos.driver.secret.values=guessme,passwd123</pre>
+
+      or
+
+      <pre>spark.mesos.driver.secret.names=password1,password2</pre>
+    </p>
   </td>
 </tr>
 
@@ -532,40 +543,48 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
   <td><code>(none)</code></td>
   <td>
-    A secret is specified by its contents and destination. These properties
-    specify a secret's destination. To specify a secret's contents, see the cell above.
-
-    You can specify a secret's destination in the driver or
-    executors as either (1) an environment variable or (2) as a file.
-    (1) To make an environment-based secret, set the
-    <code>spark.mesos.[driver|executor].secret.envkeys</code> property.
-    The secret will appear as an environment variable with the
-    given name in the driver or executors. For example, to make a secret password available
-    to the driver process as $PASSWORD, set:
-
-    <pre>spark.mesos.driver.secret.envkeys=PASSWORD</pre>
-
-    (2) To make a file-based secret, set the
-    <code>spark.mesos.[driver|executor].secret.filenames</code> property.
-    The secret will appear in the contents of a file with the given file name in
-    the driver or executors. For example, to make a secret password available in a
-    file named "pwdfile" in the driver process, set:
-
-    <pre>spark.mesos.driver.secret.filenames=pwdfile</pre>
-
-    Paths are relative to the container's work directory. Absolute paths must
-    already exist. Note: File-based secrets require a custom
-    <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
-    module</a>.
-    
-    To specify env vars or file names corresponding to multiple secrets,
-    provide a comma-separated list:
-
-    <pre>spark.mesos.driver.secret.envkeys=PASSWORD1,PASSWORD2</pre>
+    <p>
+      A secret is specified by its contents and destination. These properties
+      specify a secret's destination. To specify a secret's contents, see the cell above.
+    </p>
+    <p>
+      You can specify a secret's destination in the driver or
+      executors as either (1) an environment variable or (2) as a file.
+    </p>
+    <p>
+      (1) To make an environment-based secret, set the
+      <code>spark.mesos.[driver|executor].secret.envkeys</code> property.
+      The secret will appear as an environment variable with the
+      given name in the driver or executors. For example, to make a secret password available
+      to the driver process as $PASSWORD, set:
+
+      <pre>spark.mesos.driver.secret.envkeys=PASSWORD</pre>
+    </p>
+    <p>
+      (2) To make a file-based secret, set the
+      <code>spark.mesos.[driver|executor].secret.filenames</code> property.
+      The secret will appear in the contents of a file with the given file name in
+      the driver or executors. For example, to make a secret password available in a
+      file named "pwdfile" in the driver process, set:
+
+      <pre>spark.mesos.driver.secret.filenames=pwdfile</pre>
+    </p>
+    <p>
+      Paths are relative to the container's work directory. Absolute paths must
+      already exist. Note: File-based secrets require a custom
+      <a href="http://mesos.apache.org/documentation/latest/secrets/">SecretResolver
+      module</a>.
+    </p>
+    <p>
+      To specify env vars or file names corresponding to multiple secrets,
+      provide a comma-separated list:
+
+      <pre>spark.mesos.driver.secret.envkeys=PASSWORD1,PASSWORD2</pre>
     
-    or
+      or
     
-    <pre>spark.mesos.driver.secret.filenames=pwdfile1,pwdfile2</pre>
+      <pre>spark.mesos.driver.secret.filenames=pwdfile1,pwdfile2</pre>
+    </p>
   </td>
 </tr>
 
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 3f703b9cd658..bfb73611f053 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
@@ -193,14 +193,14 @@ private[mesos] object MesosSchedulerBackendUtil extends Logging {
     }
 
     val referenceSecrets: Seq[Secret] =
-      conf.get(secretConfig.SECRET_NAMES).getOrElse(Nil).map(s => createReferenceSecret(s))
+      conf.get(secretConfig.SECRET_NAMES).getOrElse(Nil).map { s => createReferenceSecret(s) }
 
     val valueSecrets: Seq[Secret] = {
-      conf.get(secretConfig.SECRET_VALUES).getOrElse(Nil).map(s => createValueSecret(s))
+      conf.get(secretConfig.SECRET_VALUES).getOrElse(Nil).map { s => createValueSecret(s) }
     }
 
     if (valueSecrets.nonEmpty && referenceSecrets.nonEmpty) {
-      throw new SparkException("Cannot specify VALUE type secrets and REFERENCE types ones")
+      throw new SparkException("Cannot specify both value-type and reference-type secrets.")
     }
 
     if (referenceSecrets.nonEmpty) referenceSecrets else valueSecrets