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.

Author: susanxhuynh

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,23 +503,29 @@ 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>
   <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.
+    protobuf for more information. Example:
+    
+    <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.
+    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,23 +548,29 @@ 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>
   <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.
+    protobuf for more information. Example:
+    
+    <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.
+    protobuf for more information. Example:
+    
+    <pre>secretvalue1,secretvalue2</pre>
   </td>
 </tr>
 

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)

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

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)
   }
 

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)

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")