sunchao commented on code in PR #58054:
URL: https://github.com/apache/spark/pull/58054#discussion_r3808236156
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -321,6 +326,27 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
decommissionExecutors(Array((executorId, v._1)), v._2, v._3)
unknownExecutorsPendingDecommission.invalidate(executorId)
})
+ if (executorsHeld) {
+ // The executors are held; drain this late-registered executor
immediately. Note
+ // that ExecutorMonitor cannot be told here: it builds its state
asynchronously
+ // from the SparkListenerExecutorAdded event posted above, so this
executor is not
+ // tracked there yet and its decommissioning is under-reported in
the metrics.
+ // The monitor's own executor-removal handling covers the eventual
exit.
+ decommissionExecutors(
+ Array((executorId, ExecutorDecommissionInfo("Executors are
held"))),
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ // The cluster manager granted this executor against a stale
requirement (e.g. a
+ // restarted AM using its own initial target, or a lost response
to an earlier
+ // request), so re-assert the zero requirement, or it would keep
granting
+ // replacements that churn through this drain. Do not await the
response: this
+ // runs inside the RPC handler.
+ CoarseGrainedSchedulerBackend.this.synchronized {
+ val defaultProf =
scheduler.sc.resourceProfileManager.defaultResourceProfile
+ requestedTotalExecutorsPerResourceProfile(defaultProf) = 0
+
doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
Review Comment:
[P2] Coordinate the late-registration zero target with resume
A late `RegisterExecutor` can observe `executorsHeld = true`, finish
decommissioning, and pause before this block. Meanwhile, `resumeExecutors()`
can clear the guard, restore a nonzero target, receive an acknowledgment, and
clear `_executorsHeld`. This block then overwrites the restored requirement
with zero without checking whether that hold is still current.
With static Standalone allocation, the last request wins and no allocation
loop repairs it; another `resumeExecutors()` is a no-op because the context
already considers itself resumed. An idle dynamic-allocation application can
also retain its positive EAM target without sending another request. A
deterministic probe using the current method bodies reproduced target requests
`2 -> 0 -> 2 -> 0`.
Could the zero-target reassertion and resume transition share one
ordering/lock or generation check? A regression test that pauses registration
after the held check, completes an acknowledged resume, then releases
registration would cover the failure.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, decommission support,
and shuffle data
+ * kept outside the executors: either an external shuffle service or a
`ShuffleDataIO` with
+ * reliable storage.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
+ conf.get(DECOMMISSION_ENABLED)
+ }
+
+ /** Whether the executors are currently held by `holdExecutors()`. */
+ private[spark] def executorsHeld: Boolean = _executorsHeld
+
+ /**
+ * :: DeveloperApi ::
+ * Hold the whole application by declining to allocate new executors and
gracefully
+ * decommissioning all existing ones. Each executor finishes its running
tasks and then exits,
+ * while the shuffle data already written remains available outside the
executors, so the
+ * application can later pick up where it left off via `resumeExecutors()`.
+ * Cached blocks are not preserved and are recomputed after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * @return whether the lowered executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background; the
+ * executors are drained in either case.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ synchronized {
+ if (_executorsHeld) {
+ true
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume: prefer what the
application actually
+ // requested (e.g. through requestExecutors), which the backend
tracks. When it
+ // never pushed a requirement, Standalone has no explicit one by
default (and
+ // ignores spark.executor.instances, even a leftover value), so
restore an
+ // unbounded one; elsewhere follow the conf, or fall back to the
cluster
+ // manager's default when no executor has registered yet.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+ // Floor with the live executor count: killing an executor of
a static
+ // application seeds the tracked requirement with a zero,
which is a
+ // bookkeeping artifact rather than a requirement to restore.
+ math.max(
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0),
+ b.getExecutorIds().size)
Review Comment:
[P2] Preserve an explicitly reduced static executor target
The existing [`requestTotalExecutors`
contract](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/core/src/main/scala/org/apache/spark/SparkContext.scala#L1962-L1968)
permits lowering the requested total without killing running executors. For
example, a static application can have 10 live executors but explicitly request
a future total of 2. This `math.max` saves 10, so a hold followed by resume
requests 10 instead of the application's explicit target of 2. An explicit zero
target is lost in the same way.
The bookkeeping-zero workaround cannot distinguish that legitimate request
from an inferred target. Could the backend preserve whether the target was
explicitly requested and use the live-count fallback only when it was not? A
focused probe of the current methods reproduced the 2-to-10 restoration.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
+
+ /**
+ * Whether `holdExecutors()` is supported in the current deployment. It
requires a scheduler
+ * backend that can adjust the number of executors, decommission support,
and shuffle data
+ * kept outside the executors: either an external shuffle service or a
`ShuffleDataIO` with
+ * reliable storage.
+ */
+ private[spark] def executorHoldSupported: Boolean = {
+ schedulerBackend.isInstanceOf[ExecutorAllocationClient] &&
+ (conf.get(SHUFFLE_SERVICE_ENABLED) ||
shuffleDriverComponents.supportsReliableStorage()) &&
+ conf.get(DECOMMISSION_ENABLED)
+ }
+
+ /** Whether the executors are currently held by `holdExecutors()`. */
+ private[spark] def executorsHeld: Boolean = _executorsHeld
+
+ /**
+ * :: DeveloperApi ::
+ * Hold the whole application by declining to allocate new executors and
gracefully
+ * decommissioning all existing ones. Each executor finishes its running
tasks and then exits,
+ * while the shuffle data already written remains available outside the
executors, so the
+ * application can later pick up where it left off via `resumeExecutors()`.
+ * Cached blocks are not preserved and are recomputed after resuming.
+ *
+ * This requires decommission support (`spark.decommission.enabled`) and
shuffle data kept
+ * outside the executors: either an external shuffle service
+ * (`spark.shuffle.service.enabled`) or a `ShuffleDataIO` with reliable
storage.
+ *
+ * @return whether the lowered executor requirement was acknowledged by the
cluster manager.
+ * With dynamic allocation a rejected request is retried in the
background; the
+ * executors are drained in either case.
+ */
+ @DeveloperApi
+ def holdExecutors(): Boolean = {
+ schedulerBackend match {
+ case b: ExecutorAllocationClient =>
+ require(executorHoldSupported,
+ s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
+ s"${SHUFFLE_SERVICE_ENABLED.key} or a ShuffleDataIO with reliable
storage")
+ synchronized {
+ if (_executorsHeld) {
+ true
+ } else {
+ if (executorAllocationManager.isEmpty) {
+ // The requirement to restore on resume: prefer what the
application actually
+ // requested (e.g. through requestExecutors), which the backend
tracks. When it
+ // never pushed a requirement, Standalone has no explicit one by
default (and
+ // ignores spark.executor.instances, even a leftover value), so
restore an
+ // unbounded one; elsewhere follow the conf, or fall back to the
cluster
+ // manager's default when no executor has registered yet.
+ val requested = b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.requestedTotalExecutors()
+ case _ => immutable.Map.empty[ResourceProfile, Int]
+ }
+ heldNumExecutors = if (requested.nonEmpty) {
+ // Floor with the live executor count: killing an executor of
a static
+ // application seeds the tracked requirement with a zero,
which is a
+ // bookkeeping artifact rather than a requirement to restore.
+ math.max(
+
requested.getOrElse(resourceProfileManager.defaultResourceProfile, 0),
+ b.getExecutorIds().size)
+ } else {
+ schedulerBackend match {
+ case _: StandaloneSchedulerBackend => Int.MaxValue
+ case _ =>
+
conf.get(EXECUTOR_INSTANCES).getOrElse(math.max(b.getExecutorIds().size,
+ SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS))
+ }
+ }
+ }
+ // Mark the hold before talking to the cluster manager, so that a
partial failure
+ // below leaves the executors held, and thus resumable, instead of
half-held.
+ _executorsHeld = true
+ b match {
+ case cg: CoarseGrainedSchedulerBackend =>
cg.setExecutorsHeld(true)
+ case _ =>
+ }
+ executorAllocationManager match {
+ case Some(manager) =>
+ val acknowledged = manager.suspend()
+ drainHeldExecutors(b)
+ acknowledged
+ case None => zeroExecutorRequirementAndDrain(b)
+ }
+ }
+ }
+ case _ =>
+ logWarning("Holding executors is not supported by current scheduler.")
+ false
+ }
+ }
+
+ // Gracefully decommission all the current executors of a held application
and let the
+ // executor monitor know, so that it does not try to remove the draining
executors again and
+ // reports them in the decommissioning metrics.
+ private def drainHeldExecutors(b: ExecutorAllocationClient): Unit = {
+ val executors = b.getExecutorIds()
+ if (executors.nonEmpty) {
+ val decommissioned = b.decommissionExecutors(
+ executors.map(id => (id, ExecutorDecommissionInfo("Executors are
held"))).toArray,
+ adjustTargetNumExecutors = false,
+ triggeredByExecutor = false)
+ executorAllocationManager.foreach(
+ _.executorMonitor.executorsDecommissioned(decommissioned))
+ }
+ }
+
+ // Restore the hold invariant without dynamic allocation: push the zero
executor requirement
+ // and drain the current executors. The requirement push must not abort the
drain, which has
+ // to run even when the cluster manager is temporarily unreachable; the
registration guard
+ // in CoarseGrainedSchedulerBackend remains the backstop when the push fails.
+ private[spark] def zeroExecutorRequirementAndDrain(b:
ExecutorAllocationClient): Boolean = {
+ val acknowledged = try {
+ b.requestTotalExecutors(
+ immutable.Map(resourceProfileManager.defaultResourceProfile.id -> 0),
+ immutable.Map.empty, immutable.Map.empty)
Review Comment:
[P2] Gate graceful hold for Kubernetes controller allocators
This zero-target request runs before `drainHeldExecutors` (the dynamic path
has the same ordering through `manager.suspend()`). With
`spark.kubernetes.allocation.pods.allocator=deployment` or `statefulset`, the
Kubernetes backend forwards the target to an allocator that immediately scales
its controller to zero:
[Deployment](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/DeploymentPodsAllocator.scala#L113-L123),
[StatefulSet](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/StatefulSetPodsAllocator.scala#L105-L112).
Those applications pass `executorHoldSupported` when decommissioning and a
reliable `ShuffleDataIO` are enabled.
Scaling to zero starts deleting running executor pods. The decommission
`preStop` hook still runs within the pod's termination grace, which Spark
[defaults to 30
seconds](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala#L832-L838);
Kubernetes [forcibly terminates remaining processes after the grace
period](https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-termination-flow).
A sufficiently long-running task is therefore interrupted rather than allowed
to finish, contrary to the new hold API's guarantee.
Could these allocators be excluded by a capability check until there is
controller-aware draining? The source path and termination behavior are
verified; I have not run a Kubernetes integration reproduction.
##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -2075,6 +2075,203 @@ class SparkContext(config: SparkConf) extends Logging {
}
}
+ // Whether the executors are held via `holdExecutors()`, and, when dynamic
allocation is
+ // disabled, the number of executors to restore on `resumeExecutors()`.
+ @volatile private var _executorsHeld: Boolean = false
+ private var heldNumExecutors: Int = 0
Review Comment:
[P2] Initialize hold state before exposing the startup UI
These field initializers execute after the main constructor initialization
block. That block already [attaches the live UI
handlers](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/core/src/main/scala/org/apache/spark/SparkContext.scala#L707-L732)
and then calls `postStartHook()`, which can wait for cluster resources. A
supported application can therefore receive a successful hold during that wait,
only for these later `false`/`0` assignments to erase the context's held flag
and saved target.
The backend (and EAM, when enabled) remains held, but `resumeExecutors()`
sees `_executorsHeld == false` and returns without restoring resources. I
reproduced the reset with Scala 2.13.18 and the PR's exact hold/resume methods.
Please initialize these fields with the other [early context-state
fields](https://github.com/apache/spark/blob/7dcc4702335ef25b918f054645304317a7279c2a/core/src/main/scala/org/apache/spark/SparkContext.scala#L209-L246),
before publishing the context to the UI.
--
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]
