Skip to content

[SPARK-41410][K8S] Support PVC-oriented executor pod allocation #38943

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
dongjoon-hyun wants to merge 2 commits into from
Closed

[SPARK-41410][K8S] Support PVC-oriented executor pod allocation #38943

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
999efc0
[SPARK-41410][K8S] Support PVC-oriented executor pod allocation
dongjoon-hyun Dec 6, 2022
7e861d0
Address comments
dongjoon-hyun Dec 6, 2022
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
14 changes: 12 additions & 2 deletions resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,7 @@ import java.util.concurrent.TimeUnit

import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config.{PYSPARK_DRIVER_PYTHON, PYSPARK_PYTHON}
import org.apache.spark.internal.config.ConfigBuilder
import org.apache.spark.internal.config.{ConfigBuilder, DYN_ALLOCATION_MAX_EXECUTORS, EXECUTOR_INSTANCES, PYSPARK_DRIVER_PYTHON, PYSPARK_PYTHON}

private[spark] object Config extends Logging {

Expand Down Expand Up @@ -101,6 +100,17 @@ private[spark] object Config extends Logging {
.booleanConf
.createWithDefault(true)

val KUBERNETES_DRIVER_WAIT_TO_REUSE_PVC =
ConfigBuilder("spark.kubernetes.driver.waitToReusePersistentVolumeClaims")
.doc("If true, driver pod counts the number of created on-demand persistent volume claims " +
s"and wait if the number is greater than or equal to the maximum which is " +
s"${EXECUTOR_INSTANCES.key} or ${DYN_ALLOCATION_MAX_EXECUTORS.key}. " +
s"This config requires both ${KUBERNETES_DRIVER_OWN_PVC.key}=true and " +
s"${KUBERNETES_DRIVER_REUSE_PVC.key}=true.")
Comment on lines +104 to +109
Copy link
Member

@viirya viirya Dec 6, 2022

Choose a reason for hiding this comment

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

Why not to mention PVC-oriented executor pod allocation in the config description? I think it is more clear on what this feature is.

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Dec 6, 2022

Choose a reason for hiding this comment

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

Yes, initially, I tried to use it as a config name but PVC-oriented executor pod allocation was achieved by three configurations.

  • spark.kubernetes.driver.waitToReusePersistentVolumeClaims
  • spark.kubernetes.driver.ownPersistentVolumeClaims
  • spark.kubernetes.driver.reusePersistentVolumeClaims

I'll add a K8s document section with that name.

.version("3.4.0")
.booleanConf
.createWithDefault(false)

val KUBERNETES_NAMESPACE =
ConfigBuilder("spark.kubernetes.namespace")
.doc("The namespace that will be used for running the driver and executor pods.")
Expand Down
20 changes: 19 additions & 1 deletion ...es/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,9 @@ import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.deploy.k8s.KubernetesConf
import org.apache.spark.deploy.k8s.KubernetesUtils.addOwnerReference
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config.DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT
import org.apache.spark.internal.config.{DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT, DYN_ALLOCATION_MAX_EXECUTORS, EXECUTOR_INSTANCES}
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.scheduler.cluster.SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS
import org.apache.spark.util.{Clock, Utils}

class ExecutorPodsAllocator(
Expand All @@ -47,6 +48,17 @@ class ExecutorPodsAllocator(

private val EXECUTOR_ID_COUNTER = new AtomicInteger(0)

private val PVC_COUNTER = new AtomicInteger(0)

private val maxPVCs = if (Utils.isDynamicAllocationEnabled(conf)) {
conf.get(DYN_ALLOCATION_MAX_EXECUTORS)
} else {
conf.getInt(EXECUTOR_INSTANCES.key, DEFAULT_NUMBER_EXECUTORS)
}

private val reusePVC = conf.get(KUBERNETES_DRIVER_OWN_PVC) &&
conf.get(KUBERNETES_DRIVER_REUSE_PVC) && conf.get(KUBERNETES_DRIVER_WAIT_TO_REUSE_PVC)
Copy link
Member

@viirya viirya Dec 6, 2022

Choose a reason for hiding this comment

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

Hmm, if don't use KUBERNETES_DRIVER_WAIT_TO_REUSE_PVC, cannot we still reuse PVCs?

Copy link
Member

@viirya viirya Dec 6, 2022

Choose a reason for hiding this comment

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

I think this named reusePVC is a bit confused. Isn't it the combination of three configs you mentioned (https://github.com/apache/spark/pull/38943/files#r1041347366)?

Maybe podAllocOnPVC?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Dec 6, 2022

Choose a reason for hiding this comment

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

You are right. Let me rename~


// ResourceProfile id -> total expected executors per profile, currently we don't remove
// any resource profiles - https://issues.apache.org/jira/browse/SPARK-30749
private val totalExpectedExecutorsPerResourceProfileId = new ConcurrentHashMap[Int, Int]()
Expand Down Expand Up @@ -398,6 +410,10 @@ class ExecutorPodsAllocator(
// Check reusable PVCs for this executor allocation batch
val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
for ( _ <- 0 until numExecutorsToAllocate) {
if (reusablePVCs.isEmpty && reusePVC && maxPVCs <= PVC_COUNTER.get()) {
Copy link
Member

@viirya viirya Dec 6, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Is reusablePVCs always less than or equal to maxPVCs?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Dec 6, 2022

Choose a reason for hiding this comment

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

Thank you for review.

Theoretically, reusablePVCs are all driver-owned PVCs whose creation time is bigger than podAllocationDelay + now. So, it can be bigger than maxPVCs is there is other PVC creation logic (For example, Spark driver plugin).

spark/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala

Lines 364 to 382 in 89b2ee2

private def getReusablePVCs(applicationId: String, pvcsInUse: Seq[String]) = {
if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && conf.get(KUBERNETES_DRIVER_REUSE_PVC) &&
driverPod.nonEmpty) {
try {
val createdPVCs = kubernetesClient
.persistentVolumeClaims
.inNamespace(namespace)
.withLabel("spark-app-selector", applicationId)
.list()
.getItems
.asScala
val now = Instant.now().toEpochMilli
val reusablePVCs = createdPVCs
.filterNot(pvc => pvcsInUse.contains(pvc.getMetadata.getName))
.filter(pvc => now - Instant.parse(pvc.getMetadata.getCreationTimestamp).toEpochMilli
> podAllocationDelay)
logInfo(s"Found ${reusablePVCs.size} reusable PVCs from ${createdPVCs.size} PVCs")
reusablePVCs

Also, previously, Spark creates new pod and PVCs when some executors are dead. In that case, PVCs could be created a little more.

Copy link
Member

@viirya viirya Dec 6, 2022

Choose a reason for hiding this comment

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

So even maxPVCs <= PVC_COUNTER.get(), if reusablePVCs is not empty, the driver will continue executor pod allocation?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Dec 6, 2022

Choose a reason for hiding this comment

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

Yes, correct! When we have reusablePVCs, PVC-oriented executor pod allocation doesn't need to be blocked. We halts executor allocation only when there is no available PVCs and reached PVC_COUNTER is greater than or equal to the maximum .

logInfo(s"Wait to reuse one of the existing ${PVC_COUNTER.get()} PVCs.")
return
}
val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
val executorConf = KubernetesConf.createExecutorConf(
conf,
Expand Down Expand Up @@ -429,6 +445,7 @@ class ExecutorPodsAllocator(
logInfo(s"Trying to create PersistentVolumeClaim ${pvc.getMetadata.getName} with " +
s"StorageClass ${pvc.getSpec.getStorageClassName}")
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
PVC_COUNTER.incrementAndGet()
}
newlyCreatedExecutors(newExecutorId) = (resourceProfileId, clock.getTimeMillis())
logDebug(s"Requested executor with id $newExecutorId from Kubernetes.")
Expand All @@ -438,6 +455,7 @@ class ExecutorPodsAllocator(
.inNamespace(namespace)
.resource(createdExecutorPod)
.delete()
PVC_COUNTER.decrementAndGet()
throw e
}
}
Expand Down
75 changes: 74 additions & 1 deletion ...re/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesExecutorSp
import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.deploy.k8s.Fabric8Aliases._
import org.apache.spark.internal.config.DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT
import org.apache.spark.internal.config.{DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT, EXECUTOR_INSTANCES}
import org.apache.spark.resource._
import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils._
import org.apache.spark.util.ManualClock
Expand Down Expand Up @@ -90,6 +90,9 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
@Mock
private var pvcWithNamespace: PVC_WITH_NAMESPACE = _

@Mock
private var pvcResource: io.fabric8.kubernetes.client.dsl.Resource[PersistentVolumeClaim] = _

@Mock
private var labeledPersistentVolumeClaims: LABELED_PERSISTENT_VOLUME_CLAIMS = _

Expand Down Expand Up @@ -139,6 +142,7 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
when(kubernetesClient.persistentVolumeClaims()).thenReturn(persistentVolumeClaims)
when(persistentVolumeClaims.inNamespace("default")).thenReturn(pvcWithNamespace)
when(pvcWithNamespace.withLabel(any(), any())).thenReturn(labeledPersistentVolumeClaims)
when(pvcWithNamespace.resource(any())).thenReturn(pvcResource)
when(labeledPersistentVolumeClaims.list()).thenReturn(persistentVolumeClaimList)
when(persistentVolumeClaimList.getItems).thenReturn(Seq.empty[PersistentVolumeClaim].asJava)
}
Expand Down Expand Up @@ -809,6 +813,75 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
podsAllocatorUnderTest invokePrivate getReusablePVCs("appId", Seq("pvc-1"))
}

test("SPARK-41410: Support waitToReusePersistentVolumeClaims") {
val prefix = "spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1"
val confWithPVC = conf.clone
.set(KUBERNETES_DRIVER_OWN_PVC.key, "true")
.set(KUBERNETES_DRIVER_REUSE_PVC.key, "true")
.set(KUBERNETES_DRIVER_WAIT_TO_REUSE_PVC.key, "true")
.set(EXECUTOR_INSTANCES.key, "1")
.set(s"$prefix.mount.path", "/spark-local-dir")
.set(s"$prefix.mount.readOnly", "false")
.set(s"$prefix.option.claimName", "OnDemand")
.set(s"$prefix.option.sizeLimit", "200Gi")
.set(s"$prefix.option.storageClass", "gp3")

when(executorBuilder.buildFromFeatures(any(classOf[KubernetesExecutorConf]), meq(secMgr),
meq(kubernetesClient), any(classOf[ResourceProfile])))
.thenAnswer((invocation: InvocationOnMock) => {
val k8sConf: KubernetesExecutorConf = invocation.getArgument(0)
KubernetesExecutorSpec(
executorPodWithIdAndVolume(k8sConf.executorId.toInt, k8sConf.resourceProfileId),
Seq(persistentVolumeClaim("pvc-0", "gp3", "200Gi")))
})

podsAllocatorUnderTest = new ExecutorPodsAllocator(
confWithPVC, secMgr, executorBuilder,
kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
podsAllocatorUnderTest.start(TEST_SPARK_APP_ID, schedulerBackend)

when(podsWithNamespace
.withField("status.phase", "Pending"))
.thenReturn(labeledPods)
when(labeledPods
.withLabel(SPARK_APP_ID_LABEL, TEST_SPARK_APP_ID))
.thenReturn(labeledPods)
when(labeledPods
.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE))
.thenReturn(labeledPods)
when(labeledPods
.withLabelIn(meq(SPARK_EXECUTOR_ID_LABEL), any()))
.thenReturn(labeledPods)

val startTime = Instant.now.toEpochMilli
waitForExecutorPodsClock.setTime(startTime)

val counter = PrivateMethod[AtomicInteger](Symbol("PVC_COUNTER"))()
assert(podsAllocatorUnderTest.invokePrivate(counter).get() === 0)

// Target 1 executor, make sure it's requested, even with an empty initial snapshot.
podsAllocatorUnderTest.setTotalExpectedExecutors(Map(defaultProfile -> 1))
assert(podsAllocatorUnderTest.numOutstandingPods.get() == 1)
verify(podsWithNamespace).resource(podWithAttachedContainerForIdAndVolume(1))
assert(podsAllocatorUnderTest.invokePrivate(counter).get() === 1)

// Mark executor as running, verify that subsequent allocation cycle is a no-op.
snapshotsStore.updatePod(runningExecutor(1))
snapshotsStore.notifySubscribers()
assert(podsAllocatorUnderTest.numOutstandingPods.get() == 0)
verify(podResource, times(1)).create()
verify(podResource, never()).delete()
verify(pvcWithNamespace, times(1)).resource(any())
assert(podsAllocatorUnderTest.invokePrivate(counter).get() === 1)

// Request a new executor, make sure that no new pod and pvc are created
podsAllocatorUnderTest.setTotalExpectedExecutors(Map(defaultProfile -> 2))
snapshotsStore.notifySubscribers()
assert(podsAllocatorUnderTest.numOutstandingPods.get() == 0)
verify(podResource, times(1)).create()
assert(podsAllocatorUnderTest.invokePrivate(counter).get() === 1)
}

private def executorPodAnswer(): Answer[KubernetesExecutorSpec] =
(invocation: InvocationOnMock) => {
val k8sConf: KubernetesExecutorConf = invocation.getArgument(0)
Expand Down