sarutak commented on code in PR #59112:
URL: https://github.com/apache/spark/pull/59112#discussion_r4130468434
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala:
##########
@@ -121,23 +121,28 @@ class ExecutorPVCResizeDriverPlugin extends DriverPlugin
with Logging {
}
private[k8s] def checkAndResizePVCs(): Unit = {
- logInfo(s"Latest PVC usage reports: $latestReports")
val appId = sparkContext.applicationId
sparkContext.schedulerBackend match {
case b: KubernetesClusterSchedulerBackend =>
val client = b.kubernetesClient
+ // Skip terminated pods kept by deleteOnTermination=false since their
reports are stale.
val pods = client.pods()
.inNamespace(namespace)
.withLabel(SPARK_APP_ID_LABEL, appId)
.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+ .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
.list()
.getItems.asScala
val podByExecId = pods.flatMap { p =>
Option(p.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)).map(_
-> p)
}.toMap
+ // Drop reports of executors without a pod so that latestReports does
not grow unbounded.
+ latestReports.keySet().retainAll(podByExecId.keySet.asJava)
Review Comment:
[receive()](https://github.com/apache/spark/pull/59112/changes#diff-5fa000711e41475041f5991251a1d462e02a76b1d0c391b102ab84d1dbf723ccL111)
calls `latestReports.put(...)` from the RPC callback thread, while
`checkAndResizePVCs()` runs on the single `pvc-resize-plugin` scheduled thread.
There is a small window where a brand-new executor reports (`put`) after the
`client.pods()...list()` snapshot is taken but before this `retainAll`, so its
very first report can be pruned here. This is benign (the executor re-reports
every interval, so it self-heals with at most a one-interval delay, and the
`ConcurrentHashMap` bulk op stays structurally safe), but a one-line comment
noting that the race is intentionally acceptable would help future readers.
```scala
// A newly reported executor may be pruned here if its pod is not yet in the
// listing; harmless because the executor re-reports every interval.
latestReports.keySet().retainAll(podByExecId.keySet.asJava)
```
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala:
##########
@@ -308,6 +309,27 @@ class ExecutorPVCResizePluginSuite
verify(pvcsWithNamespace,
never()).withName(org.mockito.ArgumentMatchers.anyString())
}
+ test("SPARK-59839: Reports of executors without a pod are removed") {
+ val plugin = createPlugin()
+ val pod = createPodWithPVC(1, "pvc-1", "/data")
+ when(podList.getItems).thenReturn(Collections.singletonList(pod))
+ plugin.receive(PVCDiskUsageReport("1", 0.5))
+ plugin.receive(PVCDiskUsageReport("2", 0.5)) // No pod for executor 2
+
+ plugin.checkAndResizePVCs()
+
+ assert(plugin.latestReports.keySet() === Collections.singleton("1"))
Review Comment:
The post-prune assertion (`=== Collections.singleton("1")`) verifies the
live report survives. It would be a bit more robust to also assert that both
`"1"` and `"2"` exist before `checkAndResizePVCs()`, so that replacing the
`retainAll` with an unconditional `clear()` (or dropping the live entry) is
clearly caught.
Minor.
```scala
assert(plugin.latestReports.keySet() === Set("1", "2").asJava) // add:
pre-prune
plugin.checkAndResizePVCs()
assert(plugin.latestReports.keySet() === Collections.singleton("1"))
```
--
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]
