Skip to content

[SPARK-22953][K8S] Avoids adding duplicated secret volumes when init-container is used #20148

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
wants to merge 4 commits into from
Closed

[SPARK-22953][K8S] Avoids adding duplicated secret volumes when init-container is used #20148

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9be26a8
[SPARK-22953][K8S] Avoids adding duplicated secret volumes when init-…
liyinan926 Jan 4, 2018
2e6f7db
Addressed comments
liyinan926 Jan 4, 2018
93e1d64
Addressed more comments
liyinan926 Jan 4, 2018
5278ea7
Addressed some more comments
liyinan926 Jan 4, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions ...rs/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/MountSecretsBootstrap.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,26 +24,36 @@ import io.fabric8.kubernetes.api.model.{Container, ContainerBuilder, Pod, PodBui
private[spark] class MountSecretsBootstrap(secretNamesToMountPaths: Map[String, String]) {

/**
* Mounts Kubernetes secrets as secret volumes into the given container in the given pod.
* Add new secret volumes for the secrets specified in secretNamesToMountPaths into the given pod.
*
* @param pod the pod into which the secret volumes are being added.
* @param container the container into which the secret volumes are being mounted.
* @return the updated pod and container with the secrets mounted.
* @return the updated pod with the secret volumes added.
*/
def mountSecrets(pod: Pod, container: Container): (Pod, Container) = {
def addSecretVolumes(pod: Pod): Pod = {
var podBuilder = new PodBuilder(pod)
secretNamesToMountPaths.keys.foreach { name =>
podBuilder = podBuilder
.editOrNewSpec()
.addNewVolume()
.withName(secretVolumeName(name))
.withNewSecret()
.withSecretName(name)
.endSecret()
.endVolume()
.withName(secretVolumeName(name))
.withNewSecret()
.withSecretName(name)
.endSecret()
.endVolume()
.endSpec()
}

podBuilder.build()
}

/**
* Mounts Kubernetes secret volumes of the secrets specified in secretNamesToMountPaths into the
* given container.
*
* @param container the container into which the secret volumes are being mounted.
* @return the updated container with the secrets mounted.
*/
def mountSecrets(container: Container): Container = {
var containerBuilder = new ContainerBuilder(container)
secretNamesToMountPaths.foreach { case (name, path) =>
containerBuilder = containerBuilder
Expand All @@ -53,7 +63,7 @@ private[spark] class MountSecretsBootstrap(secretNamesToMountPaths: Map[String,
.endVolumeMount()
}

(podBuilder.build(), containerBuilder.build())
containerBuilder.build()
}

private def secretVolumeName(secretName: String): String = {
Expand Down
16 changes: 8 additions & 8 deletions ...tes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/DriverConfigOrchestrator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,12 @@ private[spark] class DriverConfigOrchestrator(
Nil
}

val mountSecretsStep = if (secretNamesToMountPaths.nonEmpty) {
Seq(new DriverMountSecretsStep(new MountSecretsBootstrap(secretNamesToMountPaths)))
} else {
Nil
}

val initContainerBootstrapStep = if (existNonContainerLocalFiles(sparkJars ++ sparkFiles)) {
val orchestrator = new InitContainerConfigOrchestrator(
sparkJars,
Expand All @@ -147,19 +153,13 @@ private[spark] class DriverConfigOrchestrator(
Nil
}

val mountSecretsStep = if (secretNamesToMountPaths.nonEmpty) {
Seq(new DriverMountSecretsStep(new MountSecretsBootstrap(secretNamesToMountPaths)))
} else {
Nil
}

Seq(
initialSubmissionStep,
serviceBootstrapStep,
kubernetesCredentialsStep) ++
dependencyResolutionStep ++
initContainerBootstrapStep ++
mountSecretsStep
mountSecretsStep ++
initContainerBootstrapStep
}

private def existNonContainerLocalFiles(files: Seq[String]): Boolean = {
Expand Down
2 changes: 1 addition & 1 deletion ...rc/main/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStep.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ private[spark] class BasicDriverConfigurationStep(
.endEnv()
.addNewEnv()
.withName(ENV_DRIVER_ARGS)
.withValue(appArgs.map(arg => "\"" + arg + "\"").mkString(" "))
.withValue(appArgs.mkString(" "))
Copy link
Contributor Author

@liyinan926 liyinan926 Jan 4, 2018

Choose a reason for hiding this comment

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

A note on this change: it seems double quotes get literally added. With the double quotes appended explicitly here, when running SparkPi with an argument 5, an NumberFormatException will be thrown complaining that ""5"" is not parsable. So the explicit double quotes are not really needed.

Copy link
Contributor

@foxish foxish Jan 4, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

Thanks for calling this out - let's try and test this using an integration test as well in https://github.com/apache-spark-on-k8s/spark-integration/. Would be good to put in manual integration testing results here until we have the PRB automation figured out.

Copy link
Contributor Author

@liyinan926 liyinan926 Jan 4, 2018

Choose a reason for hiding this comment

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

Sure, I will add one to the PR under review.

.endEnv()
.addNewEnv()
.withName(ENV_DRIVER_BIND_ADDRESS)
Expand Down
4 changes: 2 additions & 2 deletions ...core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/DriverMountSecretsStep.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ private[spark] class DriverMountSecretsStep(
bootstrap: MountSecretsBootstrap) extends DriverConfigurationStep {

override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
val (pod, container) = bootstrap.mountSecrets(
driverSpec.driverPod, driverSpec.driverContainer)
val pod = bootstrap.addSecretVolumes(driverSpec.driverPod)
val container = bootstrap.mountSecrets(driverSpec.driverContainer)
driverSpec.copy(
driverPod = pod,
driverContainer = container
Expand Down
11 changes: 4 additions & 7 deletions ...rg/apache/spark/deploy/k8s/submit/steps/initcontainer/InitContainerMountSecretsStep.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,9 @@ private[spark] class InitContainerMountSecretsStep(
bootstrap: MountSecretsBootstrap) extends InitContainerConfigurationStep {

override def configureInitContainer(spec: InitContainerSpec) : InitContainerSpec = {
val (driverPod, initContainer) = bootstrap.mountSecrets(
spec.driverPod,
spec.initContainer)
spec.copy(
driverPod = driverPod,
initContainer = initContainer
)
// Mount the secret volumes given that the volumes have already been added to the driver pod
// when mounting the secrets into the main driver container.
val initContainer = bootstrap.mountSecrets(spec.initContainer)
spec.copy(initContainer = initContainer)
}
}
6 changes: 4 additions & 2 deletions ...netes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,7 @@ private[spark] class ExecutorPodFactory(

val (maybeSecretsMountedPod, maybeSecretsMountedContainer) =
mountSecretsBootstrap.map { bootstrap =>
bootstrap.mountSecrets(executorPod, containerWithLimitCores)
(bootstrap.addSecretVolumes(executorPod), bootstrap.mountSecrets(containerWithLimitCores))
}.getOrElse((executorPod, containerWithLimitCores))

val (bootstrappedPod, bootstrappedContainer) =
Expand All @@ -227,7 +227,9 @@ private[spark] class ExecutorPodFactory(

val (pod, mayBeSecretsMountedInitContainer) =
initContainerMountSecretsBootstrap.map { bootstrap =>
bootstrap.mountSecrets(podWithInitContainer.pod, podWithInitContainer.initContainer)
// Mount the secret volumes given that the volumes have already been added to the
// executor pod when mounting the secrets into the main executor container.
(podWithInitContainer.pod, bootstrap.mountSecrets(podWithInitContainer.initContainer))
}.getOrElse((podWithInitContainer.pod, podWithInitContainer.initContainer))

val bootstrappedPod = KubernetesUtils.appendInitContainer(
Expand Down
18 changes: 9 additions & 9 deletions ...deploy/k8s/submit/SecretVolumeUtils.scala → .../spark/deploy/k8s/SecretVolumeUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,23 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.spark.deploy.k8s.submit
package org.apache.spark.deploy.k8s

import scala.collection.JavaConverters._

import io.fabric8.kubernetes.api.model.{Container, Pod}

private[spark] object SecretVolumeUtils {

def podHasVolume(driverPod: Pod, volumeName: String): Boolean = {
driverPod.getSpec.getVolumes.asScala.exists(volume => volume.getName == volumeName)
def podHasVolume(pod: Pod, volumeName: String): Boolean = {
pod.getSpec.getVolumes.asScala.exists { volume =>
volume.getName == volumeName
}
}

def containerHasVolume(
driverContainer: Container,
volumeName: String,
mountPath: String): Boolean = {
driverContainer.getVolumeMounts.asScala.exists(volumeMount =>
volumeMount.getName == volumeName && volumeMount.getMountPath == mountPath)
def containerHasVolume(container: Container, volumeName: String, mountPath: String): Boolean = {
container.getVolumeMounts.asScala.exists { volumeMount =>
volumeMount.getName == volumeName && volumeMount.getMountPath == mountPath
}
}
}
4 changes: 2 additions & 2 deletions ...st/scala/org/apache/spark/deploy/k8s/submit/steps/BasicDriverConfigurationStepSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ class BasicDriverConfigurationStepSuite extends SparkFunSuite {
private val CONTAINER_IMAGE_PULL_POLICY = "IfNotPresent"
private val APP_NAME = "spark-test"
private val MAIN_CLASS = "org.apache.spark.examples.SparkPi"
private val APP_ARGS = Array("arg1", "arg2", "arg 3")
private val APP_ARGS = Array("arg1", "arg2", "\"arg 3\"")
private val CUSTOM_ANNOTATION_KEY = "customAnnotation"
private val CUSTOM_ANNOTATION_VALUE = "customAnnotationValue"
private val DRIVER_CUSTOM_ENV_KEY1 = "customDriverEnv1"
Expand Down Expand Up @@ -82,7 +82,7 @@ class BasicDriverConfigurationStepSuite extends SparkFunSuite {
assert(envs(ENV_SUBMIT_EXTRA_CLASSPATH) === "/opt/spark/spark-examples.jar")
assert(envs(ENV_DRIVER_MEMORY) === "256M")
assert(envs(ENV_DRIVER_MAIN_CLASS) === MAIN_CLASS)
assert(envs(ENV_DRIVER_ARGS) === "\"arg1\" \"arg2\" \"arg 3\"")
assert(envs(ENV_DRIVER_ARGS) === "arg1 arg2 \"arg 3\"")
assert(envs(DRIVER_CUSTOM_ENV_KEY1) === "customDriverEnv1")
assert(envs(DRIVER_CUSTOM_ENV_KEY2) === "customDriverEnv2")

Expand Down
4 changes: 2 additions & 2 deletions ...src/test/scala/org/apache/spark/deploy/k8s/submit/steps/DriverMountSecretsStepSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,8 @@
package org.apache.spark.deploy.k8s.submit.steps

import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.deploy.k8s.MountSecretsBootstrap
import org.apache.spark.deploy.k8s.submit.{KubernetesDriverSpec, SecretVolumeUtils}
import org.apache.spark.deploy.k8s.{MountSecretsBootstrap, SecretVolumeUtils}
import org.apache.spark.deploy.k8s.submit.KubernetesDriverSpec

class DriverMountSecretsStepSuite extends SparkFunSuite {

Expand Down
7 changes: 1 addition & 6 deletions ...ache/spark/deploy/k8s/submit/steps/initcontainer/InitContainerMountSecretsStepSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,7 @@ package org.apache.spark.deploy.k8s.submit.steps.initcontainer
import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder}

import org.apache.spark.SparkFunSuite
import org.apache.spark.deploy.k8s.MountSecretsBootstrap
import org.apache.spark.deploy.k8s.submit.SecretVolumeUtils
import org.apache.spark.deploy.k8s.{MountSecretsBootstrap, SecretVolumeUtils}

class InitContainerMountSecretsStepSuite extends SparkFunSuite {

Expand All @@ -44,12 +43,8 @@ class InitContainerMountSecretsStepSuite extends SparkFunSuite {
val initContainerMountSecretsStep = new InitContainerMountSecretsStep(mountSecretsBootstrap)
val configuredInitContainerSpec = initContainerMountSecretsStep.configureInitContainer(
baseInitContainerSpec)

val podWithSecretsMounted = configuredInitContainerSpec.driverPod
val initContainerWithSecretsMounted = configuredInitContainerSpec.initContainer

Seq(s"$SECRET_FOO-volume", s"$SECRET_BAR-volume").foreach(volumeName =>
assert(SecretVolumeUtils.podHasVolume(podWithSecretsMounted, volumeName)))
Seq(s"$SECRET_FOO-volume", s"$SECRET_BAR-volume").foreach(volumeName =>
assert(SecretVolumeUtils.containerHasVolume(
initContainerWithSecretsMounted, volumeName, SECRET_MOUNT_PATH)))
Expand Down
14 changes: 8 additions & 6 deletions .../core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ import org.mockito.Mockito._
import org.scalatest.{BeforeAndAfter, BeforeAndAfterEach}

import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.deploy.k8s.{InitContainerBootstrap, MountSecretsBootstrap, PodWithDetachedInitContainer}
import org.apache.spark.deploy.k8s.{InitContainerBootstrap, MountSecretsBootstrap, PodWithDetachedInitContainer, SecretVolumeUtils}
import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._

Expand Down Expand Up @@ -165,17 +165,19 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef

val factory = new ExecutorPodFactory(
conf,
None,
Some(secretsBootstrap),
Some(initContainerBootstrap),
Some(secretsBootstrap))
val executor = factory.createExecutorPod(
"1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())

assert(executor.getSpec.getVolumes.size() === 1)
assert(SecretVolumeUtils.podHasVolume(executor, "secret1-volume"))
assert(SecretVolumeUtils.containerHasVolume(
executor.getSpec.getContainers.get(0), "secret1-volume", "/var/secret1"))
assert(executor.getSpec.getInitContainers.size() === 1)
assert(executor.getSpec.getInitContainers.get(0).getVolumeMounts.get(0).getName
=== "secret1-volume")
assert(executor.getSpec.getInitContainers.get(0).getVolumeMounts.get(0)
.getMountPath === "/var/secret1")
assert(SecretVolumeUtils.containerHasVolume(
Copy link

@hex108 hex108 Jan 4, 2018

Choose a reason for hiding this comment

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

It might be better to change None at line 168 to Some(secretBootstrap) and check volumes' number to avoid regression.

Copy link
Contributor Author

@liyinan926 liyinan926 Jan 4, 2018

Choose a reason for hiding this comment

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

Done.

Copy link

@hex108 hex108 Jan 4, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

Thanks! We also need check volumes' num in pod spec.

// check volume mounted.
assert(executor.getSpec.getVolumes.size() === 1)
assert(executor.getSpec.getVolumes.get(0).getSecret.getSecretName === "secret1")

Copy link
Contributor Author

@liyinan926 liyinan926 Jan 4, 2018

Choose a reason for hiding this comment

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

Done. 93e1d64.

executor.getSpec.getInitContainers.get(0), "secret1-volume", "/var/secret1"))

checkOwnerReferences(executor, driverPodUid)
}
Expand Down