dongjoon-hyun commented on code in PR #59756:
URL: https://github.com/apache/spark/pull/59756#discussion_r4426960796
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
+ addOwnerReference(createdExecutorPod, resources)
+ resources.iterator.filter(_.getKind ==
"PersistentVolumeClaim").takeWhile(_ => !stopped)
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
+
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
+ PVC_COUNTER.incrementAndGet()
+ }
+ succeeded = !stopped
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId,
- Option(secMgr.getSecretKey()))
- val resolvedExecutorSpec =
executorBuilder.buildFromFeatures(executorConf,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ failed = false
+ succeeded
+ } finally {
+ // Clean up failed creations and, unless pods should be retained,
requests finishing at stop.
+ if (failed || (!succeeded && shouldDeleteExecutors)) {
+ Utils.tryLogNonFatalError {
+
kubernetesClient.pods().inNamespace(namespace).resource(createdExecutorPod).delete()
+ }
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ }
+
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Set[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var reservedPVCs = PVC_COUNTER.get().toLong
+ var canAllocate = true
+ // Build the complete batch before submitting it so a feature-step failure
cannot leave
+ // already-submitted pods untracked. Reserve PVC capacity without counting
it as created.
+ while (podSpecs.size < numExecutorsToAllocate && canAllocate && !stopped) {
Review Comment:
**6. [Low] Building the batch up front misses a recovery-mode switch during
the batch**
`setRecoveryMode()` updates `conf`, and `KubernetesConf.createExecutorConf`
clones it at build time (L622). Serial builds each pod right before its POST
(L556-L557), so pods after the switch start as single-task executors
(`BasicExecutorFeatureStep.scala:170`). Here the build takes milliseconds and
the POSTs come later: with a 3s webhook and concurrency 5, a batch of 10 takes
about 6s, and none of its pods see a switch made during that time.
A task at risk of OOM can then run again on a full-core executor and use up
more of `spark.task.maxFailures`. The effect is limited to one round. This
depends on timing and I did not reproduce it. If this is an accepted tradeoff,
a note next to the comment at L688 would help.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
+ addOwnerReference(createdExecutorPod, resources)
+ resources.iterator.filter(_.getKind ==
"PersistentVolumeClaim").takeWhile(_ => !stopped)
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
+
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
+ PVC_COUNTER.incrementAndGet()
+ }
+ succeeded = !stopped
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId,
- Option(secMgr.getSecretKey()))
- val resolvedExecutorSpec =
executorBuilder.buildFromFeatures(executorConf,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ failed = false
+ succeeded
+ } finally {
+ // Clean up failed creations and, unless pods should be retained,
requests finishing at stop.
+ if (failed || (!succeeded && shouldDeleteExecutors)) {
+ Utils.tryLogNonFatalError {
+
kubernetesClient.pods().inNamespace(namespace).resource(createdExecutorPod).delete()
+ }
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ }
+
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Set[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var reservedPVCs = PVC_COUNTER.get().toLong
+ var canAllocate = true
+ // Build the complete batch before submitting it so a feature-step failure
cannot leave
+ // already-submitted pods untracked. Reserve PVC capacity without counting
it as created.
+ while (podSpecs.size < numExecutorsToAllocate && canAllocate && !stopped) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= reservedPVCs) {
+ canAllocate = false
+ } else {
+ val spec = buildExecutorPod(applicationId, resourceProfileId,
reusablePVCs)
Review Comment:
**1. [High] A batch that reaches `spark.kubernetes.allocation.maximum` is
discarded, including the pods below the limit**
`buildExecutorPod` throws at the limit (L619), and this loop runs before
anything is submitted, so the specs built so far in this batch are lost. Their
IDs stay consumed, so every later round throws on its first build.
With `spark.executor.instances=10`, `spark.kubernetes.allocation.maximum=10`
and the default batch size of 20:
- Serial creates 9 pods and then hits the limit.
- Parallel builds IDs 1 to 9, throws on ID 10 and creates nothing.
`registerPodCreationFailure` is never called, so the application does not fail
either. It waits with no executors, and the only trace is the WARN at
`ExecutorPodsSnapshotsStoreImpl.scala:148`.
A limit above the batch size loses the last batch in the same way:
`maximum=100` with `instances=100` gives 80 pods instead of 99.
`Parallel allocation respects the maximum executor pod ID`
(`ExecutorPodsAllocatorSuite.scala:1434`) pins this with `verify(podResource,
never()).create()`, and neither the docs nor the PR description mention the
difference. Could the limit end this loop instead, so the specs already built
are submitted and awaited, and the exception is raised after the batch? A
feature-step failure can still discard the batch, as the comment at L688
intends.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
Review Comment:
**4. [Low] With pod retention, a stop during the Pod POST can leave a pod
whose PVC was never created**
The conditions are `spark.kubernetes.executor.deleteOnTermination=false`,
`spark.kubernetes.driver.ownPersistentVolumeClaim=false` and on-demand PVCs,
with `stopped` set while L646 is in flight. This check then skips the PVC
creation, and `failed` and `succeeded` are both false, so L670 keeps the pod.
The backend does not delete it either
(`KubernetesClusterSchedulerBackend.scala:187`).
The pod already references the claim
(`MountVolumesFeatureStep.scala:118-122`), so it stays Pending and holds quota
until its owner reference to the driver pod collects it. Without
`spark.kubernetes.driver.pod.name`, someone has to delete it. With several
PVCs, the `takeWhile` at L652 leaves the same state with only some of the
claims.
A PVC creation failure in this method deletes the pod regardless of
retention. Could a pod whose PVCs were not all created get the same treatment?
`Stopping drains API responses` (`ExecutorPodsAllocatorSuite.scala:1535`)
covers a spec without a PVC and a PVC POST in flight, but not this case. With
the default `ownPersistentVolumeClaim=true` the backend deletes the PVCs by
label anyway, so this is limited to the non-default setting.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala:
##########
@@ -645,6 +645,41 @@ private[spark] object Config extends Logging {
.checkValue(value => value > 0, "Allocation maximum should be a positive
integer")
.createWithDefault(Int.MaxValue)
+ val KUBERNETES_ALLOCATION_PARALLEL_ENABLED =
+ ConfigBuilder("spark.kubernetes.allocation.batch.parallel.enabled")
+ .doc("When true, executor pods are created in parallel using a thread
pool sized by " +
+ "spark.kubernetes.allocation.batch.concurrency. This can speed up
executor allocation " +
+ "when Kubernetes API latency is high. When false, pods are created
serially. " +
+ "Only applies when spark.kubernetes.allocation.pods.allocator is
direct.")
+ .version("4.4.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val KUBERNETES_ALLOCATION_BATCH_CONCURRENCY =
+ ConfigBuilder("spark.kubernetes.allocation.batch.concurrency")
+ .doc("Number of concurrent threads used to create executor pods in each
allocation batch. " +
+ "Only takes effect when spark.kubernetes.allocation.pods.allocator is
direct and " +
+ "spark.kubernetes.allocation.batch.parallel.enabled is true. " +
+ "The number of pods requested is still limited by
spark.kubernetes.allocation.batch.size " +
+ "and spark.kubernetes.allocation.maxPendingPods.")
+ .version("4.4.0")
+ .intConf
+ .checkValue(value => value > 0,
+ "Allocation batch concurrency should be a positive integer")
+ .createWithDefault(5)
+
+ val KUBERNETES_ALLOCATION_SHUTDOWN_TIMEOUT =
+ ConfigBuilder("spark.kubernetes.allocation.shutdownTimeout")
+ .doc("Maximum time to wait for in-flight parallel executor pod creation
during shutdown. " +
+ "Queued requests are cancelled. After this timeout, application
cleanup proceeds and " +
+ "late API responses may not be cleaned up by the allocator. This wait
is separate from " +
+ "the snapshot subscriber shutdown timeout and only applies to parallel
allocation " +
+ "when spark.kubernetes.allocation.pods.allocator is direct.")
+ .version("4.4.0")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .checkValue(value => value >= 0, "Allocation shutdown timeout must not
be negative")
+ .createWithDefaultString("20s")
Review Comment:
**2. [Medium] The default timeouts add up to more than the 30s shutdown hook
budget**
Spark's shutdown hooks share `hadoop.service.shutdown.timeout` (30s by
default), and the doc of
`spark.kubernetes.executorSnapshotsSubscribersShutdownGracePeriod` (L974-L979)
asks for its 20s to stay shorter than that, so the rest of the shutdown is not
blocked. This new 20s wait runs in front of it. The docs in this PR say the two
"should fit within the overall application shutdown budget", but the defaults
do not fit on this path:
1. `ExecutorPodsLifecycleManager` calls `sys.exit(11)` on a subscriber
thread (`ExecutorPodsLifecycleManager.scala:95`). That thread never returns, so
`snapshotsStore.stop()` always uses its full 20s.
2. A create is still in flight, and `stopAllocation` waits for more than 10s
first. For example, a webhook with `failurePolicy: Fail` that keeps timing out
holds a POST that long, because the client retries on 5xx.
3. At 30s Hadoop cancels the hook with an interrupt. `ThreadUtils.shutdown`
rethrows the `InterruptedException`, `Utils.tryLogNonFatalError` does not catch
it, and the whole stop unwinds from
`KubernetesClusterSchedulerBackend.scala:156`.
That skips deleting the Service, PVCs, pods and ConfigMap, and closing the
client. The rest of `SparkContext.stop` is skipped too, including
`listenerBus.stop` and `_eventLogger.stop`, so the event log stays
`.inprogress`. Before this PR the same path had about 10s left after the grace
period.
A normal stop does not reach 40s, because the snapshot store wait is short
there. Lowering the default alone would give up the late-create cleanup this
wait exists for, so the two waits probably need a shared deadline or a single
budget (see item 12).
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala:
##########
@@ -235,6 +235,48 @@ class KubernetesClusterSchedulerBackendSuite extends
SparkFunSuite with BeforeAn
verify(kubernetesClient).close()
}
+ Seq(false, true).foreach { deleteExecutors =>
+ test(s"Stop allocation before closing the client:
deleteExecutors=$deleteExecutors") {
+ val conf = sparkConf.clone.set(KUBERNETES_DELETE_EXECUTORS,
deleteExecutors)
+ when(sc.conf).thenReturn(conf)
+ val backend = new KubernetesClusterSchedulerBackend(taskScheduler, sc,
kubernetesClient,
+ schedulerExecutorService, eventQueue, podAllocator, lifecycleManager,
+ watchEvents, pollEvents)
+ backend.stop()
+ val order = inOrder(podAllocator, eventQueue, kubernetesClient)
+ order.verify(podAllocator).stopAllocation()
+ order.verify(eventQueue).stop()
+ if (deleteExecutors) {
+ order.verify(podAllocator).stop(TEST_SPARK_APP_ID)
+ } else {
+ verify(podAllocator, never()).stop(any())
+ }
+ order.verify(kubernetesClient).close()
+ }
+ }
+
+ test("Interrupted allocation shutdown still cleans resources and restores
interruption") {
+ doAnswer((_: org.mockito.invocation.InvocationOnMock) => {
+ Thread.currentThread().interrupt()
+ // Model an interrupted awaitTermination, which clears the flag before
throwing.
+ assert(Thread.interrupted())
+ throw new InterruptedException("allocation shutdown interrupted")
+ }).when(podAllocator).stopAllocation()
+ doAnswer((_: org.mockito.invocation.InvocationOnMock) => {
+ assert(!Thread.currentThread().isInterrupted)
Review Comment:
**9. [Low] This assertion cannot fail the test**
`eventQueue.stop()` runs inside `Utils.tryLogNonFatalError`
(`KubernetesClusterSchedulerBackend.scala:155-157`), so a `TestFailedException`
from this Answer is only logged. If the catch block restored the flag right
away, `verify(eventQueue).stop()` would still pass because Mockito records the
call before running the Answer, and so would the `assert` at L274. That is the
regression this test is meant to catch. Recording the value and asserting after
`stop()` would make it effective:
```scala
var interruptedDuringCleanup = false
doAnswer((_: org.mockito.invocation.InvocationOnMock) => {
interruptedDuringCleanup = Thread.currentThread().isInterrupted
null
}).when(eventQueue).stop()
...
schedulerBackendUnderTest.stop()
assert(!interruptedDuringCleanup)
```
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -654,7 +810,34 @@ class ExecutorPodsAllocator(
}
}
+ override def stopAllocation(): Unit = synchronized {
+ if (!podAllocationParallelEnabled || stopped) return
+ stopped = true
+ podCreationThreadPool.foreach { pool =>
+ // Do not interrupt an in-flight create: Fabric8 stops waiting for the
response on interrupt,
+ // but the API server may still create the resource. Let workers observe
the result and clean
+ // up before the backend deletes PVCs and closes the client. Queued
tasks need not run.
+ pool.shutdown()
+ val queuedTasks = new ArrayList[Runnable]()
+ pool.getQueue.drainTo(queuedTasks)
+ queuedTasks.asScala.foreach {
+ case future: Future[_] => future.cancel(false)
+ case _ =>
+ }
+ // A worker may observe a nonempty queue after shutdown, then block in
take() after
+ // drainTo empties it. Signal idle workers again without interrupting
active API calls.
+ pool.shutdown()
+ val timeout = podAllocationShutdownTimeout
+ if (!pool.awaitTermination(timeout, TimeUnit.MILLISECONDS)) {
Review Comment:
**8. [Nit] `shutdownTimeout=0` can warn about late responses when nothing is
in flight**
The docs say zero skips the wait. `pool.shutdown()` only signals idle
workers, which exit asynchronously, so `awaitTermination(0)` returns false
unless the pool has already terminated. After any batch the fixed pool has idle
workers, so a stop with no active request can still log "late API responses may
leave resources behind". Adding `pool.getActiveCount > 0` to this condition
would keep the warning for the real case. The `0s` case of `Allocation shutdown
uses its own timeout and is idempotent`
(`ExecutorPodsAllocatorSuite.scala:1669`) does not look at this log.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/AbstractPodsAllocator.scala:
##########
@@ -87,6 +87,16 @@ abstract class AbstractPodsAllocator {
@Since("3.3.0")
def start(applicationId: String, schedulerBackend:
KubernetesClusterSchedulerBackend): Unit
+ /**
+ * Stop allocating executors before application resource cleanup, including
when pods are
Review Comment:
**10. [Nit] The new Scaladoc does not hold for the default mode**
`ExecutorPodsAllocator.stopAllocation` returns immediately unless parallel
allocation is enabled (`ExecutorPodsAllocator.scala:814`), and `Disabled
parallel allocation ignores concurrency and the new stop hook`
(`ExecutorPodsAllocatorSuite.scala:1135`) pins that a serial allocator still
creates pods after this call. The default implementation here is also empty, so
the Deployment and StatefulSet allocators do nothing. Could this describe it as
an optional hook called before cleanup, and say that the direct allocator acts
on it only in parallel mode?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
+ addOwnerReference(createdExecutorPod, resources)
+ resources.iterator.filter(_.getKind ==
"PersistentVolumeClaim").takeWhile(_ => !stopped)
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
+
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
+ PVC_COUNTER.incrementAndGet()
+ }
+ succeeded = !stopped
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId,
- Option(secMgr.getSecretKey()))
- val resolvedExecutorSpec =
executorBuilder.buildFromFeatures(executorConf,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ failed = false
+ succeeded
+ } finally {
+ // Clean up failed creations and, unless pods should be retained,
requests finishing at stop.
+ if (failed || (!succeeded && shouldDeleteExecutors)) {
+ Utils.tryLogNonFatalError {
+
kubernetesClient.pods().inNamespace(namespace).resource(createdExecutorPod).delete()
+ }
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ }
+
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Set[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var reservedPVCs = PVC_COUNTER.get().toLong
+ var canAllocate = true
+ // Build the complete batch before submitting it so a feature-step failure
cannot leave
+ // already-submitted pods untracked. Reserve PVC capacity without counting
it as created.
+ while (podSpecs.size < numExecutorsToAllocate && canAllocate && !stopped) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= reservedPVCs) {
+ canAllocate = false
+ } else {
+ val spec = buildExecutorPod(applicationId, resourceProfileId,
reusablePVCs)
+ // As in serial allocation, admit the whole executor before counting
its PVCs. One
+ // executor may need multiple PVCs, so a strict per-PVC cap could
prevent any progress.
+ podSpecs += spec
+ reservedPVCs += spec._3.count(_.getKind == "PersistentVolumeClaim")
+ }
+ }
+ if (!canAllocate) {
+ logInfo(log"Wait to reuse PVCs: ${MDC(LogKeys.COUNT, PVC_COUNTER.get())}
created, " +
+ log"${MDC(LogKeys.TOTAL, reservedPVCs)} including this batch.")
+ }
+
+ val podBuildTotalMs = clock.getTimeMillis() - batchStartTime
+ val futures = podSpecs.iterator.takeWhile(_ => !stopped).flatMap {
+ case (executorId, pod, resources) =>
try {
- addOwnerReference(createdExecutorPod, resources)
- resources
- .filter(_.getKind == "PersistentVolumeClaim")
- .foreach { resource =>
- if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
- addOwnerReference(driverPod.get, Seq(resource))
- }
- val pvc = resource.asInstanceOf[PersistentVolumeClaim]
- logInfo(log"Trying to create PersistentVolumeClaim " +
- log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
- log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
- kubernetesClient
- .persistentVolumeClaims()
- .inNamespace(namespace)
- .resource(pvc)
- .create()
- PVC_COUNTER.incrementAndGet()
+ val future = podCreationThreadPool.get.submit(new
Callable[Option[Long]] {
+ override def call(): Option[Long] = {
+ if (createExecutorPod(pod, resources))
Some(clock.getTimeMillis()) else None
+ }
+ })
+ Some((executorId, future))
+ } catch {
+ case _: RejectedExecutionException if stopped => None
+ }
+ }.toVector
+
+ var succeeded = 0
+ futures.foreach { case (executorId, future) =>
+ if (!stopped) {
+ try {
+ awaitPodCreation(future).foreach { timestamp =>
+ if (!stopped) {
+ newlyCreatedExecutors(executorId) = (resourceProfileId,
timestamp)
+ succeeded += 1
}
- newlyCreatedExecutors(newExecutorId) = (resourceProfileId,
clock.getTimeMillis())
- logDebug(s"Requested executor with id $newExecutorId from
Kubernetes.")
+ }
} catch {
case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
val failureCount = registerPodCreationFailure()
Review Comment:
**3. [Medium] A persistent PVC failure now costs a whole batch of API calls
and failures per round**
The serial path ends the round on a PVC failure (`throw e` at L606). Here
each future records its own failure and the batch continues, so a PVC POST that
keeps failing (for example, on an exhausted quota) is multiplied by the batch
size. Each attempt is a Pod POST, a PVC POST and a Pod DELETE. With the default
batch size of 20:
| | API writes per round | Failures per round | Rounds until `sys.exit(11)`
with `instances=50` |
|---|---|---|---|
| Serial | 3 | 1 | 101 |
| Parallel | 60 | 20 | 6 |
The limit in that example is `max(3, 2 * 50)` and rounds are about 1s apart,
so the application exits after about 6s instead of about 105s. The extra writes
arrive while the API server is already rejecting requests.
Pod POST failures do not multiply this way, because serial also continues
the batch on them (L571). The `failPVC` case of `Parallel allocation reserves
PVC capacity and retries failures` (`ExecutorPodsAllocatorSuite.scala:1377`)
runs with `maxPVCs=1`, so it tries one pod per round and does not show this.
Could a PVC failure stop the rest of the batch, the way `stopped` does, and
rethrow after the batch as serial does?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala:
##########
@@ -132,6 +132,26 @@ private[spark] class KubernetesClusterSchedulerBackend(
super.stop()
}
+ var interrupted = false
+ try {
+ try {
+ Utils.tryLogNonFatalError {
+ podAllocator.stopAllocation()
Review Comment:
**5. [Low] Subscribers keep running after the driver endpoint is gone**
`super.stop()` has already removed the driver endpoint, and polling stops
only in `snapshotsStore.stop()`, which now runs after this wait. While a create
is in flight, the lifecycle subscriber keeps ticking for up to 20s more:
- Each new terminal executor goes to
`driverEndpoint.send(RemoveExecutor(...))`, which throws `Could not find
CoarseGrainedScheduler` and logs a WARN through
`ExecutorPodsSnapshotsStoreImpl.scala:148`.
- If a `PodFailed` in this window passes the failure limit, `sys.exit(11)`
runs. An application that finished its work then exits with code 11. The hook's
`SparkContext.stop` returns at once because a stop is already in progress, so
the JVM ends with `stop()` still waiting here and the Kubernetes cleanup is
skipped.
The second case should be rare, since executors stopped on request end as
`PodSucceeded`. Signaling first and waiting after `snapshotsStore.stop()` would
close this window (see item 12).
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -654,7 +810,34 @@ class ExecutorPodsAllocator(
}
}
+ override def stopAllocation(): Unit = synchronized {
+ if (!podAllocationParallelEnabled || stopped) return
+ stopped = true
+ podCreationThreadPool.foreach { pool =>
+ // Do not interrupt an in-flight create: Fabric8 stops waiting for the
response on interrupt,
+ // but the API server may still create the resource. Let workers observe
the result and clean
+ // up before the backend deletes PVCs and closes the client. Queued
tasks need not run.
+ pool.shutdown()
+ val queuedTasks = new ArrayList[Runnable]()
+ pool.getQueue.drainTo(queuedTasks)
+ queuedTasks.asScala.foreach {
+ case future: Future[_] => future.cancel(false)
+ case _ =>
+ }
+ // A worker may observe a nonempty queue after shutdown, then block in
take() after
+ // drainTo empties it. Signal idle workers again without interrupting
active API calls.
+ pool.shutdown()
Review Comment:
**11. [Low] `drainTo`, `cancel` and the second `shutdown()` are not needed**
`stopped` is set at L815, before `pool.shutdown()`, and a queued task
returns at `if (stopped) return false` (L644) without reaching the API. With
`shutdown()` alone, workers take the queued tasks through the normal `getTask`
path, finish them at once, and the pool terminates by itself. Queued requests
still do not reach the API, active requests are still not interrupted, and the
timeout and idempotence stay the same. I did not find a case that needs the
drain.
The race in [this
thread](https://github.com/apache/spark/pull/59756#discussion_r4396982374)
exists only because `drainTo` goes around `getTask`. Removing the drain also
removes these 9 lines, the 60-line test that writes a private final field
through reflection (`ExecutorPodsAllocatorSuite.scala:1607-1666`), and the path
in item 7. The docs and `Config.scala:674` would then say that queued requests
are skipped instead of cancelled.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
+ addOwnerReference(createdExecutorPod, resources)
+ resources.iterator.filter(_.getKind ==
"PersistentVolumeClaim").takeWhile(_ => !stopped)
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
+
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
+ PVC_COUNTER.incrementAndGet()
+ }
+ succeeded = !stopped
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId,
- Option(secMgr.getSecretKey()))
- val resolvedExecutorSpec =
executorBuilder.buildFromFeatures(executorConf,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ failed = false
+ succeeded
+ } finally {
+ // Clean up failed creations and, unless pods should be retained,
requests finishing at stop.
+ if (failed || (!succeeded && shouldDeleteExecutors)) {
+ Utils.tryLogNonFatalError {
+
kubernetesClient.pods().inNamespace(namespace).resource(createdExecutorPod).delete()
+ }
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ }
+
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Set[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var reservedPVCs = PVC_COUNTER.get().toLong
+ var canAllocate = true
+ // Build the complete batch before submitting it so a feature-step failure
cannot leave
+ // already-submitted pods untracked. Reserve PVC capacity without counting
it as created.
+ while (podSpecs.size < numExecutorsToAllocate && canAllocate && !stopped) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= reservedPVCs) {
+ canAllocate = false
+ } else {
+ val spec = buildExecutorPod(applicationId, resourceProfileId,
reusablePVCs)
+ // As in serial allocation, admit the whole executor before counting
its PVCs. One
+ // executor may need multiple PVCs, so a strict per-PVC cap could
prevent any progress.
+ podSpecs += spec
+ reservedPVCs += spec._3.count(_.getKind == "PersistentVolumeClaim")
+ }
+ }
+ if (!canAllocate) {
+ logInfo(log"Wait to reuse PVCs: ${MDC(LogKeys.COUNT, PVC_COUNTER.get())}
created, " +
+ log"${MDC(LogKeys.TOTAL, reservedPVCs)} including this batch.")
+ }
+
+ val podBuildTotalMs = clock.getTimeMillis() - batchStartTime
+ val futures = podSpecs.iterator.takeWhile(_ => !stopped).flatMap {
+ case (executorId, pod, resources) =>
try {
- addOwnerReference(createdExecutorPod, resources)
- resources
- .filter(_.getKind == "PersistentVolumeClaim")
- .foreach { resource =>
- if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
- addOwnerReference(driverPod.get, Seq(resource))
- }
- val pvc = resource.asInstanceOf[PersistentVolumeClaim]
- logInfo(log"Trying to create PersistentVolumeClaim " +
- log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
- log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
- kubernetesClient
- .persistentVolumeClaims()
- .inNamespace(namespace)
- .resource(pvc)
- .create()
- PVC_COUNTER.incrementAndGet()
+ val future = podCreationThreadPool.get.submit(new
Callable[Option[Long]] {
+ override def call(): Option[Long] = {
+ if (createExecutorPod(pod, resources))
Some(clock.getTimeMillis()) else None
+ }
+ })
+ Some((executorId, future))
+ } catch {
+ case _: RejectedExecutionException if stopped => None
+ }
+ }.toVector
+
+ var succeeded = 0
+ futures.foreach { case (executorId, future) =>
+ if (!stopped) {
+ try {
+ awaitPodCreation(future).foreach { timestamp =>
+ if (!stopped) {
+ newlyCreatedExecutors(executorId) = (resourceProfileId,
timestamp)
+ succeeded += 1
}
- newlyCreatedExecutors(newExecutorId) = (resourceProfileId,
clock.getTimeMillis())
- logDebug(s"Requested executor with id $newExecutorId from
Kubernetes.")
+ }
} catch {
case NonFatal(e) =>
Review Comment:
**7. [Low] A cancelled queued request can be counted as a pod creation
failure**
If the caller is waiting on a future that `stopAllocation` cancels (L824),
the `CancellationException` comes back wrapped in a `SparkException`
(`ThreadUtils.scala:399-401`). L753 does not match it and this catch has no
`stopped` check, so a pod whose request was never sent gets an ERROR log and
one registered failure.
It can happen at most once per stop and the window is narrow (the future
must still be queued while the caller waits on it), so the real cost is a
misleading ERROR. A `stopped` check here would fix it, and removing the drain
(item 11) removes this path.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -518,86 +540,220 @@ class ExecutorPodsAllocator(
applicationId: String,
resourceProfileId: Int,
pvcsInUse: Set[String]): Unit = {
- // Check reusable PVCs for this executor allocation batch
- val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
- for ( _ <- 0 until numExecutorsToAllocate) {
- if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
- logInfo(
- log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
- return
+ if (podAllocationParallelEnabled) {
+ if (stopped) return
+ requestNewExecutorsParallel(numExecutorsToAllocate, applicationId,
+ resourceProfileId, pvcsInUse)
+ } else {
+ // Check reusable PVCs for this executor allocation batch
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ for ( _ <- 0 until numExecutorsToAllocate) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <=
PVC_COUNTER.get()) {
+ logInfo(
+ log"Wait to reuse one of the existing ${MDC(LogKeys.COUNT,
PVC_COUNTER.get())} PVCs.")
+ return
+ }
+ val (newExecutorId, podWithAttachedContainer, resources) =
+ buildExecutorPod(applicationId, resourceProfileId, reusablePVCs)
+ val optCreatedExecutorPod = try {
+ Some(kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .resource(podWithAttachedContainer)
+ .create())
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to create executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ None
+ }
+ optCreatedExecutorPod.foreach { createdExecutorPod =>
+ try {
+ addOwnerReference(createdExecutorPod, resources)
+ resources
+ .filter(_.getKind == "PersistentVolumeClaim")
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty)
{
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME,
pvc.getMetadata.getName)} with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
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.")
+ } catch {
+ case NonFatal(e) =>
+ // Register failure with global tracker if lifecycle manager is
available
+ val failureCount = registerPodCreationFailure()
+ logError(log"Failed to add owner reference or create PVC for
executor pod " +
+ log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
+ kubernetesClient.pods()
+ .inNamespace(namespace)
+ .resource(createdExecutorPod)
+ .delete()
+ throw e
+ }
+ }
}
- val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
- if (newExecutorId >= podAllocationMaximum) {
- throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ }
+
+ // Build on the allocation thread: feature steps and PVC reuse mutate shared
state.
+ private def buildExecutorPod(
+ applicationId: String,
+ resourceProfileId: Int,
+ reusablePVCs: mutable.Buffer[PersistentVolumeClaim]): (Long, Pod,
Seq[HasMetadata]) = {
+ val newExecutorId = EXECUTOR_ID_COUNTER.incrementAndGet()
+ if (newExecutorId >= podAllocationMaximum) {
+ throw new SparkException(s"Exceed the pod creation limit:
$podAllocationMaximum")
+ }
+ val executorConf = KubernetesConf.createExecutorConf(
+ conf,
+ newExecutorId.toString,
+ applicationId,
+ driverPod,
+ resourceProfileId,
+ Option(secMgr.getSecretKey()))
+ val resolvedExecutorSpec = executorBuilder.buildFromFeatures(executorConf,
+ kubernetesClient, rpIdToResourceProfile(resourceProfileId))
+ val executorPod = resolvedExecutorSpec.pod
+ val podWithAttachedContainer = new PodBuilder(executorPod.pod)
+ .editOrNewSpec()
+ .addToContainers(executorPod.container)
+ .endSpec()
+ .build()
+ val resources = replacePVCsIfNeeded(
+ podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
+ (newExecutorId.toLong, podWithAttachedContainer, resources)
+ }
+
+ // Only the Kubernetes calls run on worker threads; allocator maps are
updated by the caller.
+ private def createExecutorPod(pod: Pod, resources: Seq[HasMetadata]):
Boolean = {
+ if (stopped) return false
+ val createdExecutorPod =
+ kubernetesClient.pods().inNamespace(namespace).resource(pod).create()
+ var succeeded = false
+ var failed = true
+ try {
+ if (!stopped) {
+ addOwnerReference(createdExecutorPod, resources)
+ resources.iterator.filter(_.getKind ==
"PersistentVolumeClaim").takeWhile(_ => !stopped)
+ .foreach { resource =>
+ if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
+ addOwnerReference(driverPod.get, Seq(resource))
+ }
+ val pvc = resource.asInstanceOf[PersistentVolumeClaim]
+ logInfo(log"Trying to create PersistentVolumeClaim " +
+ log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
+ log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
+
kubernetesClient.persistentVolumeClaims().inNamespace(namespace).resource(pvc).create()
+ PVC_COUNTER.incrementAndGet()
+ }
+ succeeded = !stopped
}
- val executorConf = KubernetesConf.createExecutorConf(
- conf,
- newExecutorId.toString,
- applicationId,
- driverPod,
- resourceProfileId,
- Option(secMgr.getSecretKey()))
- val resolvedExecutorSpec =
executorBuilder.buildFromFeatures(executorConf,
- kubernetesClient, rpIdToResourceProfile(resourceProfileId))
- val executorPod = resolvedExecutorSpec.pod
- val podWithAttachedContainer = new PodBuilder(executorPod.pod)
- .editOrNewSpec()
- .addToContainers(executorPod.container)
- .endSpec()
- .build()
- val resources = replacePVCsIfNeeded(
- podWithAttachedContainer,
resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
- val optCreatedExecutorPod = try {
- Some(kubernetesClient
- .pods()
- .inNamespace(namespace)
- .resource(podWithAttachedContainer)
- .create())
- } catch {
- case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
- val failureCount = registerPodCreationFailure()
- logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
- log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- None
+ failed = false
+ succeeded
+ } finally {
+ // Clean up failed creations and, unless pods should be retained,
requests finishing at stop.
+ if (failed || (!succeeded && shouldDeleteExecutors)) {
+ Utils.tryLogNonFatalError {
+
kubernetesClient.pods().inNamespace(namespace).resource(createdExecutorPod).delete()
+ }
}
- optCreatedExecutorPod.foreach { createdExecutorPod =>
+ }
+ }
+
+ private def requestNewExecutorsParallel(
+ numExecutorsToAllocate: Int,
+ applicationId: String,
+ resourceProfileId: Int,
+ pvcsInUse: Set[String]): Unit = {
+ val batchStartTime = clock.getTimeMillis()
+ val reusablePVCs = getReusablePVCs(applicationId, pvcsInUse)
+ val podSpecs = mutable.ArrayBuffer.empty[(Long, Pod, Seq[HasMetadata])]
+ var reservedPVCs = PVC_COUNTER.get().toLong
+ var canAllocate = true
+ // Build the complete batch before submitting it so a feature-step failure
cannot leave
+ // already-submitted pods untracked. Reserve PVC capacity without counting
it as created.
+ while (podSpecs.size < numExecutorsToAllocate && canAllocate && !stopped) {
+ if (reusablePVCs.isEmpty && podAllocOnPVC && maxPVCs <= reservedPVCs) {
+ canAllocate = false
+ } else {
+ val spec = buildExecutorPod(applicationId, resourceProfileId,
reusablePVCs)
+ // As in serial allocation, admit the whole executor before counting
its PVCs. One
+ // executor may need multiple PVCs, so a strict per-PVC cap could
prevent any progress.
+ podSpecs += spec
+ reservedPVCs += spec._3.count(_.getKind == "PersistentVolumeClaim")
+ }
+ }
+ if (!canAllocate) {
+ logInfo(log"Wait to reuse PVCs: ${MDC(LogKeys.COUNT, PVC_COUNTER.get())}
created, " +
+ log"${MDC(LogKeys.TOTAL, reservedPVCs)} including this batch.")
+ }
+
+ val podBuildTotalMs = clock.getTimeMillis() - batchStartTime
+ val futures = podSpecs.iterator.takeWhile(_ => !stopped).flatMap {
+ case (executorId, pod, resources) =>
try {
- addOwnerReference(createdExecutorPod, resources)
- resources
- .filter(_.getKind == "PersistentVolumeClaim")
- .foreach { resource =>
- if (conf.get(KUBERNETES_DRIVER_OWN_PVC) && driverPod.nonEmpty) {
- addOwnerReference(driverPod.get, Seq(resource))
- }
- val pvc = resource.asInstanceOf[PersistentVolumeClaim]
- logInfo(log"Trying to create PersistentVolumeClaim " +
- log"${MDC(LogKeys.PVC_METADATA_NAME, pvc.getMetadata.getName)}
with " +
- log"StorageClass ${MDC(LogKeys.CLASS_NAME,
pvc.getSpec.getStorageClassName)}")
- kubernetesClient
- .persistentVolumeClaims()
- .inNamespace(namespace)
- .resource(pvc)
- .create()
- PVC_COUNTER.incrementAndGet()
+ val future = podCreationThreadPool.get.submit(new
Callable[Option[Long]] {
+ override def call(): Option[Long] = {
+ if (createExecutorPod(pod, resources))
Some(clock.getTimeMillis()) else None
+ }
+ })
+ Some((executorId, future))
+ } catch {
+ case _: RejectedExecutionException if stopped => None
+ }
+ }.toVector
+
+ var succeeded = 0
+ futures.foreach { case (executorId, future) =>
+ if (!stopped) {
+ try {
+ awaitPodCreation(future).foreach { timestamp =>
+ if (!stopped) {
+ newlyCreatedExecutors(executorId) = (resourceProfileId,
timestamp)
+ succeeded += 1
}
- newlyCreatedExecutors(newExecutorId) = (resourceProfileId,
clock.getTimeMillis())
- logDebug(s"Requested executor with id $newExecutorId from
Kubernetes.")
+ }
} catch {
case NonFatal(e) =>
- // Register failure with global tracker if lifecycle manager is
available
val failureCount = registerPodCreationFailure()
- logError(log"Failed to add owner reference or create PVC for
executor pod " +
- log"${MDC(LogKeys.EXECUTOR_ID, newExecutorId)}. " +
+ logError(log"Failed to create executor pod
${MDC(LogKeys.EXECUTOR_ID, executorId)}. " +
log"Total failures: ${MDC(LogKeys.TOTAL, failureCount)}", e)
- kubernetesClient.pods()
- .inNamespace(namespace)
- .resource(createdExecutorPod)
- .delete()
- throw e
}
}
}
+ logDebug(log"Parallel pod allocation: submitted=${MDC(LogKeys.COUNT,
futures.size)}, " +
+ log"succeeded=${MDC(LogKeys.TOTAL, succeeded)}, " +
+ log"podBuildTotalMs=${MDC(LogKeys.DURATION, podBuildTotalMs)}, " +
+ log"batchTotalMs=${MDC(LogKeys.TOTAL_TIME, clock.getTimeMillis() -
batchStartTime)}")
+ }
+
+ private def awaitPodCreation(future: Future[Option[Long]]): Option[Long] = {
+ // Keep waiting for live requests during normal allocation, so late
successes stay tracked.
+ // At shutdown the worker owns cleanup; let the snapshot callback exit
even if the API request
+ // outlives the grace period, rather than blocking snapshot-store shutdown
for another period.
+ while (!stopped) {
Review Comment:
**12. [Low] Could this be one wait instead of two?**
If the hook only signals (`stopped = true` and `pool.shutdown()`) and the
snapshot callback waits for its futures without this polling loop, the existing
subscriber grace period already covers in-flight requests: the callback is the
only submitter, and `snapshotsStore.stop()` waits for the callback. The
`shutdownNow()` there interrupts only the subscriber threads, not the pool
workers, and the risk after the grace period expires is the same as after this
timeout.
That would remove `spark.kubernetes.allocation.shutdownTimeout`, this loop,
the `if (!stopped)` at L723, the `synchronized` on `stopAllocation`, and the
interrupt handling in the backend with its test. It also removes the 20s + 20s
stacking in item 2. Since the hook would no longer block, the window in item 5
closes as well.
The costs are that the wait could no longer be tuned separately, the
callback needs its own warning because the grace period expires silently, L732
needs a `stopped` check, and the shutdown tests need rewriting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
