Skip to content

[SPARK-28042][K8S] Support using volume mount as local storage #24879

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
chenjunjiedada wants to merge 9 commits into from
Closed

[SPARK-28042][K8S] Support using volume mount as local storage #24879

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6fca505
[SPARK-28042][K8S] Support using hostpath volume mount as local storage
Jun 14, 2019
ef66c87
SPARK-28042: fix unit test and address comments
chenjunjiedada Jun 16, 2019
5610fe4
SPARK-28042: refactor checks according to comments
chenjunjiedada Jun 16, 2019
f071c2c
Address comments
chenjunjiedada Jul 24, 2019
45c8bc5
Update documents as well.
chenjunjiedada Jul 24, 2019
9392dad
update document
chenjunjiedada Jul 24, 2019
6e5fcf6
Don't set configuration when specific volume is set
chenjunjiedada Jul 25, 2019
2abb8e9
address comments
chenjunjiedada Jul 26, 2019
c29dd7a
fix coding style
chenjunjiedada Jul 28, 2019
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
46 changes: 35 additions & 11 deletions ...netes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,8 @@ package org.apache.spark.deploy.k8s.features

import java.util.UUID

import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, PodBuilder, VolumeBuilder, VolumeMountBuilder}
import collection.JavaConverters._

@vanzin vanzin Jul 25, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

import scala....

import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, VolumeBuilder, VolumeMount, VolumeMountBuilder}

import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
import org.apache.spark.deploy.k8s.Config._
Expand All @@ -43,20 +44,34 @@ private[spark] class LocalDirsFeatureStep(
val localDirVolumes = resolvedLocalDirs
.zipWithIndex
.map { case (localDir, index) =>
new VolumeBuilder()
.withName(s"spark-local-dir-${index + 1}")
.withNewEmptyDir()
.withMedium(if (useLocalDirTmpFs) "Memory" else null)
.endEmptyDir()
.build()
val name = s"spark-local-dir-${index + 1}"
hasVolume(pod, name) match {

@felixcheung felixcheung Jun 15, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

not a super big deal, but hasVolume is basically the same logic pod.pod.getSpec().getVolumes().asScala.exists as pod.pod.getSpec.getVolumes().asScala.find

I think you can refactor this to avoid scanning multiple times.

@chenjunjiedada chenjunjiedada Jun 17, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Updated in the latest commit.

case true =>
pod.pod.getSpec.getVolumes().asScala.find(v => v.getName.equals(name)).get
case false =>
new VolumeBuilder()
.withName(s"spark-local-dir-${index + 1}")

@erikerlandson erikerlandson Jul 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

What is the scope of visibility on volume names "spark-local-dir-1", "spark-local-dir-2" ...
Can those names collide with volumes created by other running jobs?

@chenjunjiedada chenjunjiedada Jul 23, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

It should be visible for global scope. Since the task distributed should run after setting up executor, so it will not collide with volumes which are built later.

@rvesse rvesse Jul 23, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

volume names in this context are scoped to the pod spec

.withNewEmptyDir()
.withMedium(if (useLocalDirTmpFs) "Memory" else null)
.endEmptyDir()
.build()
}
}
val localDirVolumeMounts = localDirVolumes
.zip(resolvedLocalDirs)
.map { case (localDirVolume, localDirPath) =>
new VolumeMountBuilder()
.withName(localDirVolume.getName)
.withMountPath(localDirPath)
.build()
hasVolumeMount(pod, localDirVolume.getName, localDirPath) match {
case true =>
pod.container.getVolumeMounts().asScala
.find(m => m.getName.equals(localDirVolume.getName)

@felixcheung felixcheung Jun 15, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

please see other lines for indentation

@chenjunjiedada chenjunjiedada Jun 16, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

ok, will check and run scalafmt.

&& m.getMountPath.equals(localDirPath))
.get
case false =>
new VolumeMountBuilder()
.withName(localDirVolume.getName)
.withMountPath(localDirPath)
.build()
}
}
val podWithLocalDirVolumes = new PodBuilder(pod.pod)
.editSpec()
Expand All @@ -72,4 +87,13 @@ private[spark] class LocalDirsFeatureStep(
.build()
SparkPod(podWithLocalDirVolumes, containerWithLocalDirVolumeMounts)
}

def hasVolume(pod: SparkPod, name: String): Boolean = {
pod.pod.getSpec().getVolumes().asScala.exists(v => v.getName.equals(name))
}

def hasVolumeMount(pod: SparkPod, name: String, path: String): Boolean = {
pod.container.getVolumeMounts().asScala
.exists(m => m.getName.equals(name) && m.getMountPath.equals(path))
}
}
4 changes: 2 additions & 2 deletions ...etes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,12 @@ private[spark] class KubernetesDriverBuilder {
new DriverServiceFeatureStep(conf),
new MountSecretsFeatureStep(conf),
new EnvSecretsFeatureStep(conf),
new LocalDirsFeatureStep(conf),
new MountVolumesFeatureStep(conf),
new DriverCommandFeatureStep(conf),
new HadoopConfDriverFeatureStep(conf),
new KerberosConfDriverFeatureStep(conf),
new PodTemplateConfigMapStep(conf))
new PodTemplateConfigMapStep(conf),
new LocalDirsFeatureStep(conf))

@felixcheung felixcheung Jun 15, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

why not move this right after new MountVolumesFeatureStep(conf)?

@chenjunjiedada chenjunjiedada Jun 16, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I think some volume mounts maybe setup in PodTemplateConfigMapStep, so I move to last to ensure that.

@rvesse rvesse Jul 23, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Yep, since the intent is for users to provide custom local dir volumes via either mount volumes or pod templates it needs to appear after both

@vanzin vanzin Jul 25, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Either is fine, just wanted to point out that the pod template is initialized before any step is executed (i.e. PodTemplateConfigMapStep is not where the template is loaded).


val spec = KubernetesDriverSpec(
initialPod,
Expand Down
4 changes: 2 additions & 2 deletions ...ore/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,8 @@ private[spark] class KubernetesExecutorBuilder {
new BasicExecutorFeatureStep(conf, secMgr),
new MountSecretsFeatureStep(conf),
new EnvSecretsFeatureStep(conf),
new LocalDirsFeatureStep(conf),
new MountVolumesFeatureStep(conf))
new MountVolumesFeatureStep(conf),
new LocalDirsFeatureStep(conf))

features.foldLeft(initialPod) { case (pod, feature) => feature.configurePod(pod) }
}
Expand Down
27 changes: 25 additions & 2 deletions .../core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,8 @@ package org.apache.spark.deploy.k8s.features
import io.fabric8.kubernetes.api.model.{EnvVarBuilder, VolumeBuilder, VolumeMountBuilder}

import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.deploy.k8s.{KubernetesTestConf, SparkPod}
import org.apache.spark.deploy.k8s.{KubernetesHostPathVolumeConf, KubernetesTestConf, KubernetesVolumeSpec, SparkPod}

@vanzin vanzin Jul 25, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Use the wildcard when the import line gets too long.

import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.submit.JavaMainAppResource
import org.apache.spark.util.SparkConfWithEnv

class LocalDirsFeatureStepSuite extends SparkFunSuite {
Expand Down Expand Up @@ -116,4 +115,28 @@ class LocalDirsFeatureStepSuite extends SparkFunSuite {
.withValue(defaultLocalDir)
.build())
}

test("local dir on mounted volume") {
Comment thread
dongjoon-hyun marked this conversation as resolved.
val volumeConf = KubernetesVolumeSpec(
"spark-local-dir-1",
"/tmp",
"",
false,
KubernetesHostPathVolumeConf("/hostPath/tmp")
)
val mountVolumeConf = KubernetesTestConf.createDriverConf(volumes = Seq(volumeConf))
val mountVolumeStep = new MountVolumesFeatureStep(mountVolumeConf)
val configuredPod = mountVolumeStep.configurePod(SparkPod.initialPod())

val sparkConf = new SparkConfWithEnv(Map("SPARK_LOCAL_DIRS" -> "/tmp"))
val localDirConf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
val localDirStep = new LocalDirsFeatureStep(localDirConf, defaultLocalDir)
val newConfiguredPod = localDirStep.configurePod(configuredPod)

assert(newConfiguredPod.pod.getSpec.getVolumes.size() === 1)
assert(newConfiguredPod.pod.getSpec.getVolumes.get(0).getHostPath.getPath === "/hostPath/tmp")
assert(newConfiguredPod.container.getVolumeMounts.size() === 1)
assert(newConfiguredPod.container.getVolumeMounts.get(0).getMountPath === "/tmp")
assert(newConfiguredPod.container.getVolumeMounts.get(0).getName === "testVolume")
}
}