Addressed review comments: (1) pluralized SECRET_xyz config entries, (2) renamed 'containerInfo' -> 'buildContainerInfo', (3) added additional case to illegalSecretInput() check.

Author: susanxhuynh

resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala

@@ -22,30 +22,30 @@ 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.")
       .stringConf
       .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.")
       .stringConf
       .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.")

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)

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

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

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

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

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