Guang Yang created SPARK-60778:
----------------------------------

             Summary: [K8S] ExecutorPodsAllocator can reuse a PVC still mounted 
by a live executor when no new snapshot arrives in an allocation round
                 Key: SPARK-60778
                 URL: https://issues.apache.org/jira/browse/SPARK-60778
             Project: Spark
          Issue Type: Bug
          Components: Kubernetes
    Affects Versions: 4.1.1
            Reporter: Guang Yang


h3. Summary                                                                     
                                                                                
                                                                                
 
                                                                                
                                                                                
                                                                                
   
  When {{spark.kubernetes.driver.ownPersistentVolumeClaim=true}} and 
{{spark.kubernetes.driver.reusePersistentVolumeClaim=true}} (both default 
true), ExecutorPodsAllocator can give a new executor pod a PVC that a *running* 
executor is still
  using. With the default ReadWriteOncePod access mode, the new pod can never 
be scheduled. It stays Pending (scheduler reason {{VolumeRestrictions}}) until 
dynamic allocation deletes it or the application ends. Several pending pods can 
end up
  pointing at the same claim.                                                   
                                         
                                                                                
                                         
  h3. Root cause                                                                
                                                                                
                                                                                
   
                                                                                
                                         
  In {{ExecutorPodsAllocator.onNewSnapshots}}, the set of PVCs in use 
({{k8sKnownPVCNames}}) is built only from the snapshots delivered in the 
current batch:
                                                                                
                                         
  {code:scala}                                                                  
                                         
  val aggregatedPods = if (snapshots.size <= 1) {                               
                                         
    snapshots.headOption.map(_.executorPods).getOrElse(Map.empty)               
                                                                                
                                                                                
   
  } else { ... merge snapshots ... }                                            
                                                                                
                                                                                
   
  ...                                                                           
                                                                                
                                                                                
   
  val k8sKnownPVCNames: Set[String] = {                                         
                                                                                
                                                                                
   
    aggregatedPods.valuesIterator.foreach { state => /* collect claimName of 
each PVC volume */ }                                                            
                                                                                
      
  }                                                                             
                                                                                
                                                                                
   
  {code}                                                                        
                                                                                
                                                                                
   
                                                                                
                                         
  This set is passed to {{requestNewExecutors}} -> 
{{getReusablePVCs(applicationId, pvcsInUse)}}. That method treats any 
driver-owned PVC as reusable if it is not in {{pvcsInUse}} and is older than 
{{podCreationTimeout}}.                      
                                                                                
                                         
  There are two gaps.                                                           
                                                                                
                                                                                
   
                                                                                
                                                                                
                                                                                
   
  # *Allocation rounds with no new snapshot.* 
{{ExecutorPodsSnapshotsStoreImpl.SnapshotsSubscriber}} runs every 
{{spark.kubernetes.allocation.batch.delay}} (default 1s). It drains the 
snapshot buffer and calls {{onNewSnapshots}} even when the 
  buffer is empty. Executor counting still uses {{lastSnapshot}}, so the 
allocator can decide to scale up in that round. But {{aggregatedPods}} is 
empty, so {{pvcsInUse}} is empty. Every driver-owned PVC older than 
{{podCreationTimeout}} then 
  looks reusable, including PVCs mounted by Running executors.                  
                                                                                
                                                                                
   
  # *Executors requested but not yet in any snapshot.* A pod created in round N 
that reused PVC X shows up in a snapshot only after the watch or poll delivers 
it. In round N+1, X is again missing from {{pvcsInUse}}, so it can be handed 
out a  
  second time.

  SPARK-41388 and SPARK-49385 (PR #47867) reduced related races by skipping 
PVCs younger than {{podCreationTimeout}}. That only protects recently created 
PVCs. In a long-running application, nearly every PVC is older than the 
timeout, so 
  neither change covers this case.

  h3. Observed in production (Spark 4.1.1, dynamic allocation, on-demand 
executor PVC)

  Configuration:
  {code}
  
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName=OnDemand
  
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit=100Gi
  
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/mnt/spark-local
  spark.dynamicAllocation.enabled=true
  # own/reuse PVC left at default (true); access mode default ReadWriteOncePod
  {code}

  What we saw on one application after it had run for more than 10 minutes and 
scaled up again:
  * {{<app>-exec-1}} had been Running on {{<app>-exec-1-pvc-0}} for over a 
minute. The PVC was 17 minutes old, older than {{podCreationTimeout}}.
  * The allocator created {{<app>-exec-11}} with the same claim 
{{<app>-exec-1-pvc-0}}.
  * Seven pods in total referenced {{exec-1-pvc-0}}. Two other claims had four 
pods each. The application had 18 duplicate Pending executor pods and 636 
driver-owned PVCs.
  * The duplicate pods failed scheduling with {{VolumeRestrictions}} and never 
ran. They were cleaned up only when dynamic allocation scaled down or the 
application finished.

  Effects beyond the stuck pods:
  * The application gets fewer executors than it asked for.
  * The cluster autoscaler keeps provisioning nodes for pods that can never run.
  * A batch scheduler (YuniKorn in our case) re-checks these pods against every 
node on every cycle. Across many applications, this starved other applications' 
pods.



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)

---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to