dongjoon-hyun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4300454490
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -121,8 +121,24 @@ private[spark] class Client(
.withMountPath(SPARK_CONF_DIR_INTERNAL)
.endVolumeMount()
.build()
+ // SPARK-38079: a pod template (spark.kubernetes.driver.podTemplateFile)
may already pin
+ // spec.nodeName to a specific node. The Kubernetes API rejects creating a
pod that has
+ // both a non-empty spec.nodeName and a non-empty spec.schedulingGates
(added below) --
+ // "nodeName cannot be set until all schedulingGates have been cleared" --
since nodeName
+ // is meant to be set only once nothing (including scheduling) stands
between the pod and
+ // that node. Translate it to an equivalent kubernetes.io/hostname
nodeSelector entry
+ // instead, which (unlike nodeName) is created and read normally on a
gated pod and has
+ // the same effect once the scheduler runs -- unless
spark.kubernetes.driver.node.selector.
+ // kubernetes.io/hostname (via BasicDriverFeatureStep, applied earlier in
+ // builder.buildFromFeatures() above) already set that same key: that
explicit user
+ // configuration always wins, so it is never overwritten by this fallback.
+ val pinnedNodeName = Option(resolvedDriverSpec.pod.pod.getSpec.getNodeName)
Review Comment:
**Rewriting `spec.nodeName` into a `kubernetes.io/hostname` node selector
isn't equivalent, although the comment at L129-L131 says it "has the same
effect once the scheduler runs".** The scheduler never sees a pod with
`nodeName`; the kubelet admits it and checks only resources, node affinity and
selectors, ports and `NoExecute` taints. After this change the driver goes
through the scheduler named by `spec.schedulerName`, so cordoned nodes,
`NoSchedule` taints (already in the Known limitations), the template's pod
(anti-)affinity and topology spread constraints, and the custom scheduler's
queues all apply. For example, a template that pins the driver to a cordoned or
tainted node used to run there, and now stays `Pending`, with `spark-submit`
waiting forever when `spark.kubernetes.submission.waitAppCompletion` is on.
Two smaller points:
- When a `kubernetes.io/hostname` selector is already set, `nodeName` is
dropped (L141) without any log.
- The "Pod template properties" table in `docs/running-on-kubernetes.md`,
"the full list of pod specifications that will be overwritten by spark",
doesn't mention `nodeName` or `schedulingGates`.
Since such a pod bypasses the scheduler anyway, could we skip the gate when
the resolved pod has `nodeName`, and keep master's flow for that case?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
.endSpec()
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
- // setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod is schedulable, to avoid a
"configmap ...
+ // not found" mount race between the driver pod and the config map it
depends on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ var watch: Watch = null
+ val createdDriverPod: Pod =
+ try {
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
Review Comment:
**This creates the driver pod before every pre-resource, not only the moved
ones, so a pre-resource that pod admission needs doesn't exist yet at the pod
CREATE.** A scheduling gate only blocks binding; admission still runs on the
CREATE. On master, `driverPreKubernetesResources` were applied first, which is
what SPARK-37331 added the hook for ("they should be ready before pod creation
... such as pod group in Volcano"), and what the `@Stable` Scaladoc of
`KubernetesFeatureConfigStep.getAdditionalPreKubernetesResources()` still says
("Resources would be setup/refresh before Pod creation.").
For example, with the documented Volcano setup
(`spark.kubernetes.scheduler.name=volcano` and `VolcanoFeatureStep`) on Volcano
v1.7.0 to v1.11.x:
- The driver pod carries `scheduling.k8s.io/group-name=<appId>-podgroup`,
and Volcano's `validatepod.volcano.sh` webhook (enabled by default, CREATE,
`failurePolicy: Fail`) calls `checkPG(pod, pgName, true)` ([admit_pod.go
L111-L117](https://github.com/volcano-sh/volcano/blob/v1.11.0/pkg/webhooks/admission/pods/validate/admit_pod.go#L111-L117)).
- Since the PodGroup is created only at L190, `checkPG` fails with `failed
to get PodGroup for pod ...`
([L148-L153](https://github.com/volcano-sh/volcano/blob/v1.11.0/pkg/webhooks/admission/pods/validate/admit_pod.go#L148-L153)),
and every Volcano submission fails here with the "kubectl auth can-i create
pod" hint.
- The docs say Volcano is supported "since Spark v3.3.0 and Volcano v1.7.0".
CI doesn't catch this because it installs Volcano v1.14.2, and the check was
removed in v1.12.0.
The same happens to a custom feature step that creates the driver's
ServiceAccount as a pre-resource (`error looking up service account ...`).
Could we keep applying `driverPreKubernetesResources` before the pod, as on
master? The comment at L223 describes an order that does this and still closes
the mount race.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
.endSpec()
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
- // setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod is schedulable, to avoid a
"configmap ...
+ // not found" mount race between the driver pod and the config map it
depends on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ var watch: Watch = null
+ val createdDriverPod: Pod =
+ try {
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ throw e
+ }
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
Now that the
+ // driver pod exists, its UID is known, so each pre-resource's owner
reference can be set
+ // before -- and included in -- the single call that creates it. The pod
itself is still
+ // scheduling-gated, so nothing can be scheduled against it (and so
nothing can attempt to
+ // mount these resources) until they exist -- but they are never ownerless
at any point
+ // after this call: either it succeeds and every pre-resource already has
an owner
+ // reference, or it fails and none of them (that made it to the server)
are left
+ // referencing anything, since the driver pod that would have owned them
is deleted in the
+ // catch block below.
try {
+ addOwnerReference(createdDriverPod, preKubernetesResources)
kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
} catch {
case NonFatal(e) =>
logError("Please check \"kubectl auth can-i create [resource]\"
first." +
" It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
-
- var watch: Watch = null
- var createdDriverPod: Pod = null
- try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ deletePodAndPreResources(createdDriverPod, preKubernetesResources)
throw e
}
- // Refresh all pre-resources' owner references
+ // SPARK-38079: remove the pre-resources scheduling gate now that its
pre-resources exist,
+ // letting the scheduler proceed with this pod. If this process is
terminated abruptly
+ // before this point (e.g. Ctrl-C, SIGTERM, or a fatal JVM error), the
driver pod is left
+ // behind still gated -- inert (kubelet cannot schedule/mount anything on
a gated pod) and
+ // visible as `Pending`/`SchedulingGated` via any standard pod listing.
That is a low-
+ // severity leak, in the same class Spark already accepts elsewhere for a
process killed
+ // right after pod creation (e.g. Ctrl-C before the watch loop below
starts), so no
+ // shutdown hook is registered to guard this window.
try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+
kubernetesClient.pods().inNamespace(conf.namespace).withName(driverPodName).edit(
+ (currentPod: Pod) => new PodBuilder(currentPod)
+ .editSpec()
+ .removeMatchingFromSchedulingGates(_.getName ==
PRE_RESOURCES_SCHEDULING_GATE)
Review Comment:
**`edit()` sends a JSON patch that pins the `resourceVersion` it just read,
so any other write to the pod in between turns into a 409 and then a full
teardown.** fabric8 7.9.0 clears the version only on the base copy
([OperationSupport.java
L380-L389](https://github.com/fabric8io/kubernetes-client/blob/v7.9.0/kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/OperationSupport.java#L380-L389)),
so the patch carries the GET's `metadata.resourceVersion`. A 409 isn't
retried, and after a 5xx or an I/O error fabric8 resends the same body, which
fails once the first one was applied. Either way the catch block logs the RBAC
hint, deletes the pod and all pre-resources, and fails the submission; in the
resend case the pod may already be bound.
A vanilla cluster doesn't write to a gated pod in that window, but Kueue's
pod integration does: its webhook adds `kueue.x-k8s.io/admission` at CREATE,
and on admission Kueue patches the pod to remove that gate and merge node
selectors.
A strategic merge patch keyed by the gate name has no precondition and is
idempotent, since `schedulingGates` merges by `name`:
```json
{"spec":{"schedulingGates":[{"$patch":"delete","name":"spark.apache.org/pre-resources"}]}}
```
This module already patches pods with
`PatchContext.of(PatchType.STRATEGIC_MERGE)` (e.g.
`KubernetesClusterSchedulerBackend`, `ExecutorPodsLifecycleManager`), and
Volcano removes its own gate the same way. Sending it through
`resource(createdDriverPod)` also avoids the extra GET.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
.endSpec()
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
- // setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod is schedulable, to avoid a
"configmap ...
+ // not found" mount race between the driver pod and the config map it
depends on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ var watch: Watch = null
+ val createdDriverPod: Pod =
+ try {
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ throw e
+ }
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
Now that the
+ // driver pod exists, its UID is known, so each pre-resource's owner
reference can be set
+ // before -- and included in -- the single call that creates it. The pod
itself is still
+ // scheduling-gated, so nothing can be scheduled against it (and so
nothing can attempt to
+ // mount these resources) until they exist -- but they are never ownerless
at any point
+ // after this call: either it succeeds and every pre-resource already has
an owner
+ // reference, or it fails and none of them (that made it to the server)
are left
+ // referencing anything, since the driver pod that would have owned them
is deleted in the
+ // catch block below.
try {
+ addOwnerReference(createdDriverPod, preKubernetesResources)
kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
} catch {
case NonFatal(e) =>
logError("Please check \"kubectl auth can-i create [resource]\"
first." +
" It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
-
- var watch: Watch = null
- var createdDriverPod: Pod = null
- try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ deletePodAndPreResources(createdDriverPod, preKubernetesResources)
throw e
}
- // Refresh all pre-resources' owner references
+ // SPARK-38079: remove the pre-resources scheduling gate now that its
pre-resources exist,
+ // letting the scheduler proceed with this pod. If this process is
terminated abruptly
+ // before this point (e.g. Ctrl-C, SIGTERM, or a fatal JVM error), the
driver pod is left
+ // behind still gated -- inert (kubelet cannot schedule/mount anything on
a gated pod) and
+ // visible as `Pending`/`SchedulingGated` via any standard pod listing.
That is a low-
+ // severity leak, in the same class Spark already accepts elsewhere for a
process killed
+ // right after pod creation (e.g. Ctrl-C before the watch loop below
starts), so no
+ // shutdown hook is registered to guard this window.
try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+
kubernetesClient.pods().inNamespace(conf.namespace).withName(driverPodName).edit(
Review Comment:
**This makes `patch` on pods a new requirement for every submission.**
fabric8's `edit()` is a GET followed by an HTTP PATCH
([HasMetadataOperation.java
L58-L61](https://github.com/fabric8io/kubernetes-client/blob/v7.9.0/kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/HasMetadataOperation.java#L58-L61)),
while master's `Client.run()` never wrote to the pod after `create()`;
server-side apply touched only the other resources.
For example, kubeflow/spark-operator runs `spark-submit` inside its
controller with the controller's service account, and its default rules grant
pods only `get`, `list`, `watch`, `create`, `update` and `delete`
([`_helpers.tpl`
L110-L121](https://github.com/kubeflow/spark-operator/blob/v2.5.2/charts/spark-operator-chart/templates/controller/_helpers.tpl#L110-L121)
in v2.5.2, the same on master). Once its image bundles a Spark with this
change, every submission there would create the gated pod and the
pre-resources, get a 403 here, delete the driver pod in the catch block, and
fail. This is the same kind of RBAC gap as the
`OwnerReferencesPermissionEnforcement` case in [round
1](https://github.com/apache/spark/pull/58187#discussion_r4129910475), and I
missed it when I suggested `spec.schedulingGates`.
`docs/running-on-kubernetes.md` only says "list, create, edit and delete
pods", and `edit` isn't an RBAC verb. If we keep the gate, could we document
`patch` on pods explicitly and mention it in the migration guide? `update`
(PUT) would fit kubeflow's role, but it is also a new verb.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -221,14 +306,216 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
assert(secrets.nonEmpty)
assert(configMaps.nonEmpty)
val configMap = configMaps.head
- assert(configMap.getMetadata.getName ===
- KubernetesClientUtils.configMapNameDriver)
+ // SPARK-38079: configMapNameDriver is a `def` -- see its own comment --
so this cannot
+ // compare against a freshly-generated name (that call would return a
different one); it
+ // can only check the name has the shape configMapNameDriver is documented
to produce.
+ assert(configMap.getMetadata.getName.matches("spark-drv-\\S+-conf-map"))
Review Comment:
**Now that `configMapNameDriver` returns a new name on every call, no test
checks that the created ConfigMap is the one the pod mounts.** L271 and L329
compare the captured pod with a pod built from its own volume name, and this
regex (also at L561 and L621) accepts any `spark-drv-*-conf-map`. So if `run()`
read `configMapNameDriver` again for `buildConfigMap` (L109) or for the volume
(L147), the pod would mount a ConfigMap that is never created, which is the
"not found" failure this PR targets, and every `ClientSuite` test would still
pass. `assert(configMap.getMetadata.getName === gatedConfigMapName)` works in
all three places, so the comments saying only the shape can be checked aren't
accurate.
Similarly, the `SPARK_CONF_DIR` test lost its only check of the conf
volume's items for several files. The removed stub,
`when(podsWithNamespace.resource(fullExpectedPod(expectedKeyToPaths)))`, made
`run()` fail unless the volume listed all the expected files, and the test now
checks only the ConfigMap data. A `configMap` volume with explicit `items`
projects only those keys, so dropping one in `buildKeyToPathObjects` would go
unnoticed. Keeping `expectedKeyToPaths` and asserting `gatedPod ===
fullExpectedPod(gatedConfigMapName, expectedKeyToPaths)` would restore it; the
`keyToPaths` parameter of the three helpers is currently unused.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
.endSpec()
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
- // setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod is schedulable, to avoid a
"configmap ...
+ // not found" mount race between the driver pod and the config map it
depends on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
+
+ var watch: Watch = null
+ val createdDriverPod: Pod =
+ try {
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ throw e
+ }
+
+ // SPARK-38079: some of the pre-resources above (e.g. the Kerberos
keytab/delegation token
+ // secrets, the driver Kubernetes credentials secret) carry credentials.
Now that the
+ // driver pod exists, its UID is known, so each pre-resource's owner
reference can be set
+ // before -- and included in -- the single call that creates it. The pod
itself is still
+ // scheduling-gated, so nothing can be scheduled against it (and so
nothing can attempt to
+ // mount these resources) until they exist -- but they are never ownerless
at any point
+ // after this call: either it succeeds and every pre-resource already has
an owner
+ // reference, or it fails and none of them (that made it to the server)
are left
+ // referencing anything, since the driver pod that would have owned them
is deleted in the
+ // catch block below.
try {
+ addOwnerReference(createdDriverPod, preKubernetesResources)
kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
} catch {
case NonFatal(e) =>
logError("Please check \"kubectl auth can-i create [resource]\"
first." +
" It should be yes. And please also check your feature step
implementation.")
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- throw e
- }
-
- var watch: Watch = null
- var createdDriverPod: Pod = null
- try {
- createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
- } catch {
- case NonFatal(e) =>
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
- logError("Please check \"kubectl auth can-i create pod\" first. It
should be yes.")
+ deletePodAndPreResources(createdDriverPod, preKubernetesResources)
throw e
}
- // Refresh all pre-resources' owner references
+ // SPARK-38079: remove the pre-resources scheduling gate now that its
pre-resources exist,
+ // letting the scheduler proceed with this pod. If this process is
terminated abruptly
+ // before this point (e.g. Ctrl-C, SIGTERM, or a fatal JVM error), the
driver pod is left
+ // behind still gated -- inert (kubelet cannot schedule/mount anything on
a gated pod) and
+ // visible as `Pending`/`SchedulingGated` via any standard pod listing.
That is a low-
+ // severity leak, in the same class Spark already accepts elsewhere for a
process killed
+ // right after pod creation (e.g. Ctrl-C before the watch loop below
starts), so no
+ // shutdown hook is registered to guard this window.
try {
- addOwnerReference(createdDriverPod, preKubernetesResources)
- kubernetesClient.resourceList(preKubernetesResources:
_*).forceConflicts().serverSideApply()
+
kubernetesClient.pods().inNamespace(conf.namespace).withName(driverPodName).edit(
+ (currentPod: Pod) => new PodBuilder(currentPod)
+ .editSpec()
+ .removeMatchingFromSchedulingGates(_.getName ==
PRE_RESOURCES_SCHEDULING_GATE)
+ .endSpec()
+ .build())
} catch {
case NonFatal(e) =>
- kubernetesClient.pods().resource(createdDriverPod).delete()
- kubernetesClient.resourceList(preKubernetesResources: _*).delete()
+ logError("Please check \"kubectl auth can-i patch pod\" first. It
should be yes.")
+ deletePodAndPreResources(createdDriverPod, preKubernetesResources)
throw e
}
// setup resources after pod creation, and refresh all resources' owner
references
try {
- val otherKubernetesResources =
resolvedDriverSpec.driverKubernetesResources ++ Seq(configMap)
+ val otherKubernetesResources =
resolvedDriverSpec.driverKubernetesResources
Review Comment:
**Because the gate is removed before this apply (L207-L219), the driver
post-resources are still created after the pod can be scheduled, which is why
the four feature steps' resources and the conf ConfigMap had to move to
pre-resources.** If the gate is removed last instead, that move isn't needed:
1. Apply `driverPreKubernetesResources` before the pod, as on master (see
the comment at L171).
2. Create the gated pod.
3. Refresh the pre-resources' owner references, as on master.
4. Apply `driverKubernetesResources ++ Seq(configMap)` with owner
references. This is a CREATE, so there is no
`OwnerReferencesPermissionEnforcement` issue.
5. Remove the gate.
This covers every driver resource, including the ones that stay
post-resources here, such as the `OnDemand` driver PVC from
`MountVolumesFeatureStep` or a ConfigMap or Secret that a user's feature step
mounts. It also means:
- The four `getAdditionalPreKubernetesResources()` overrides, L166 and the
`HADOOP_CONF_DIR` test exclusions can be reverted.
- The output of `KubernetesDriverBuilder` (a `@DeveloperApi`) stays the
same, so the first Known limitation, about `apache/spark-kubernetes-operator`,
goes away.
- A failure after the pod exists only needs to delete the pod (except in
step 3, as on master), so `deletePodAndPreResources` and the
`configMapNameDriver` change aren't needed.
The trade-off is that pre-resources, which are empty by default, keep
master's ownerless window until step 3. This keeps the part of my [round-1
suggestion](https://github.com/apache/spark/pull/58187#discussion_r4129910563)
that matters (create the pod gated, apply the resources with owner references,
then remove the gate) without moving the pre-resources after the pod, which I
didn't think through then.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientUtils.scala:
##########
@@ -57,8 +57,17 @@ object KubernetesClientUtils extends Logging {
@Since("3.1.0")
val configMapNameExecutor: String =
configMapName(s"spark-exec-${KubernetesUtils.uniqueID()}")
+ // SPARK-38079: a `def`, not a `val` -- it is called exactly once per
submission
+ // (Client.run(), into a local val reused for that submission), so unlike
+ // configMapNameExecutor (set up once by KubernetesClusterSchedulerBackend,
then read
+ // independently by BasicExecutorFeatureStep for every executor, so every
call must agree on
+ // the same name), there is no cross-call consistency requirement to
preserve here. A `val`
+ // would instead share this name across every submission from the same JVM
(e.g. via
+ // InProcessLauncher) for the life of the JVM, which, now that this config
map is a
+ // pre-resource, would let a failed submission's cleanup delete-by-name a
different,
+ // still-running submission's live config map of that same name.
@Since("3.1.0")
- val configMapNameDriver: String =
configMapName(s"spark-drv-${KubernetesUtils.uniqueID()}")
+ def configMapNameDriver: String =
configMapName(s"spark-drv-${KubernetesUtils.uniqueID()}")
Review Comment:
**This turns a `@Since("3.1.0")` member of a `@Stable` `@DeveloperApi`
object from one value per JVM into a new random name on every read.** Nothing
flags it, since the accessor signature is the same, and a parameterless `def`
with a side effect (each call advances the shared RNG) reads like a stable
field. A caller that reads it twice, once for the ConfigMap and once for the
volume as the old `ClientSuite` did, would now get two names and a driver stuck
in `ContainerCreating`. No caller in Spark or
`apache/spark-kubernetes-operator` does that today; the operator has its own
`SparkAppDriverConf.configMapNameDriver()`.
The name sharing ([round
1](https://github.com/apache/spark/pull/58187#discussion_r4129910528), also
SPARK-41060) can be fixed in `Client.run()` without touching this member, for
example with
`KubernetesClientUtils.configMapName(s"${conf.resourceNamePrefix}-driver")` as
the other per-submission resources and the operator do, or with a local
`spark-drv-<uniqueID>` name. A name from `resourceNamePrefix` would also let
`ClientSuite` assert exact names again. With the order in the comment at L223,
the cross-submission delete from round 1 can't happen anyway, so this could
move to SPARK-41060.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -221,14 +306,216 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
assert(secrets.nonEmpty)
assert(configMaps.nonEmpty)
val configMap = configMaps.head
- assert(configMap.getMetadata.getName ===
- KubernetesClientUtils.configMapNameDriver)
+ // SPARK-38079: configMapNameDriver is a `def` -- see its own comment --
so this cannot
+ // compare against a freshly-generated name (that call would return a
different one); it
+ // can only check the name has the shape configMapNameDriver is documented
to produce.
+ assert(configMap.getMetadata.getName.matches("spark-drv-\\S+-conf-map"))
assert(configMap.getImmutable())
assert(configMap.getData.containsKey(SPARK_CONF_FILE_NAME))
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf1key=conf1value"))
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf2key=conf2value"))
}
+ test("SPARK-38079: driver pod is created (still scheduling-gated) before its
own config " +
+ "map, and the config map's single create call already carries the owner
reference") {
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+ submissionClient.run()
+ // The pod itself must be created before any pre-resource resourceList()
call, since a
+ // pre-resource's owner reference (set below) needs the pod's UID.
+ assert(gatedPod === fullExpectedPod(gatedConfigMapName))
+ verify(namedPods).create()
+
+ // The (single) resourceList(...) call creating the driver's own config
map must already
+ // carry the owner reference -- there is no separate, later "refresh"
call. The pod stays
+ // scheduling-gated (so kubelet cannot attempt to mount it) for as long as
this has not yet
+ // happened, avoiding the "configmap ... not found" mount race
(SPARK-38079) without ever
+ // creating the config map ownerless.
+ val resourceListCall = createdResourcesArgumentCaptor.getAllValues.get(0)
+ val configMaps = resourceListCall
+ .filter(_.isInstanceOf[ConfigMap]).map(_.asInstanceOf[ConfigMap])
+ assert(configMaps.nonEmpty,
+ "the driver's own config map must be created as a pre-resource")
+ val ownerReferences = configMaps.head.getMetadata.getOwnerReferences
+ assert(ownerReferences.size() === 1)
+ assert(ownerReferences.get(0).getName === POD_NAME)
+ assert(ownerReferences.get(0).getUid === DRIVER_POD_UID)
+
+ // The scheduling gate must be removed only after the config map has been
created.
+
verify(namedPods).edit(ArgumentMatchers.any[java.util.function.UnaryOperator[Pod]]())
+ }
+
+ // SPARK-38079: the three failure branches in run() between pod creation and
gate removal --
+ // covering (a) the pod creation call itself failing, (b) the pre-resource
creation call
+ // failing after the pod exists, and (c) the gate-removal call failing after
pre-resources
+ // exist -- each verified against real-cluster behavior
(KubernetesClientException on
+ // AlreadyExists/Invalid/NotFound) before being written as these mock-based
tests.
+ test("SPARK-38079: pod creation failure propagates without attempting any
pre-resource " +
+ "creation") {
+ val podCreationFailure = new RuntimeException("simulated pod creation
failure")
+ doThrow(podCreationFailure).when(namedPods).create()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+
+ val thrown = intercept[RuntimeException] {
+ submissionClient.run()
+ }
+
+ assert(thrown eq podCreationFailure)
+ verify(kubernetesClient,
never()).resourceList(ArgumentMatchers.any[Array[HasMetadata]](): _*)
+ }
+
+ test("SPARK-38079: pre-resource creation failure (after the pod exists)
deletes both the " +
+ "pod and the pre-resources, then propagates") {
+ val preResourceFailure = new RuntimeException("simulated pre-resource
creation failure")
+ doThrow(preResourceFailure).when(resourceList).serverSideApply()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+
+ val thrown = intercept[RuntimeException] {
+ submissionClient.run()
+ }
+
+ assert(thrown eq preResourceFailure)
+ // Not just "delete() was called on some mock" -- pods().resource(...)
must have been
+ // called with the actual created (gated) pod, i.e. the same pod object
this cleanup path
+ // is documented to delete, and each pre-resource must be deleted
individually (not as a
+ // single resourceList(...).delete() call, which stops at the first
failing item).
+ verify(namedPods).create()
+ verify(podOperations).resource(podWithOwnerReference(gatedConfigMapName))
+ verify(namedPods).delete()
+ verify(namedResource).delete()
Review Comment:
**Since every `kubernetesClient.resource(...)` returns the same
`namedResource` and each failure test has a single pre-resource, these checks
can't tell which objects were deleted.** For example, these changes still pass:
- passing `resolvedDriverSpec.driverKubernetesResources` at L195 and L217,
- one `Utils.tryLogNonFatalError` around the whole loop, so a 403 on an
earlier item skips the conf ConfigMap (the [round-1
case](https://github.com/apache/spark/pull/58187#discussion_r4129910518)),
- `preResources.take(1)`.
The pod-creation failure test also has no `never()` check on deletes,
although a delete by name after a 409 `AlreadyExists` (with a fixed
`spark.kubernetes.driver.pod.name`) would remove another submission's driver.
The gate checks have the same issue. `withName(POD_NAME)` and
`resource(any)` return the same `namedPods`, and the fixture has only Spark's
gate. So calling `edit()` on `resource(resolvedDriverPod)`, which patches from
the local copy without a GET, or a lambda that clears every gate, would also
pass. In production, both would remove a gate that a webhook added, such as
Kueue's, and Kueue then deletes the pod it didn't admit.
Could we give the fixture a second gate that must survive, add `never()`
delete checks to the creation failure test, and, if the per-item deletes stay
(see the comment at L276), capture the `resource(...)` arguments and add a
failure test with two or more pre-resources where the first delete throws?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -209,6 +257,26 @@ private[spark] class Client(
log"submission ID ${MDC(SUBMISSION_ID, sId)} into Kubernetes")
}
}
+
+ // SPARK-38079: best-effort cleanup for the two failure catch blocks between
pod creation and
+ // gate removal. The pod and each pre-resource are deleted independently --
every delete call
+ // wrapped in its own Utils.tryLogNonFatalError -- so that one of them
failing (e.g. a delete
+ // call itself hitting a permission or network error) neither masks the
original exception
+ // the caller is about to (re)throw, nor skips the rest. In particular,
fabric8's
+ // resourceList(...).delete() deletes the given items sequentially and stops
at the first
+ // non-404 exception, so pre-resources are deleted one at a time here rather
than as a single
+ // resourceList(...).delete() call, which could otherwise leave every
pre-resource after the
+ // one that failed undeleted.
+ private def deletePodAndPreResources(pod: Pod, preResources:
Seq[HasMetadata]): Unit = {
+ Utils.tryLogNonFatalError {
+ kubernetesClient.pods().resource(pod).delete()
+ }
+ preResources.foreach { resource =>
+ Utils.tryLogNonFatalError {
+ kubernetesClient.resource(resource).delete()
Review Comment:
**Now that every pre-resource gets the pod's controller owner reference
before its first write (L189), these per-item deletes duplicate garbage
collection.** Deleting the gated pod (background propagation, and it is removed
at once since it was never scheduled) lets GC remove every namespaced
pre-resource that reached the server, including one whose apply response was
lost, which an explicit DELETE can miss. The loop adds costs on the failure
path:
- an ERROR log with a stack trace per item for submitters without `delete`
on ConfigMaps, Secrets or PodGroups (the docs only require pod permissions),
- deletes by name of objects this submission may not own, e.g. a fixed-name
pre-resource from a custom step whose apply failed with 422, or items that were
never sent after the first failure,
- one more request timeout per item when the API server is unreachable.
I asked for per-item deletes in [round
1](https://github.com/apache/spark/pull/58187#discussion_r4129910518), when
pre-resources were created without an owner. With the owner reference set up
front, could this be a single guarded pod delete, keeping explicit deletes only
for cluster-scoped pre-resources, if any?
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -221,14 +306,216 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
assert(secrets.nonEmpty)
assert(configMaps.nonEmpty)
val configMap = configMaps.head
- assert(configMap.getMetadata.getName ===
- KubernetesClientUtils.configMapNameDriver)
+ // SPARK-38079: configMapNameDriver is a `def` -- see its own comment --
so this cannot
+ // compare against a freshly-generated name (that call would return a
different one); it
+ // can only check the name has the shape configMapNameDriver is documented
to produce.
+ assert(configMap.getMetadata.getName.matches("spark-drv-\\S+-conf-map"))
assert(configMap.getImmutable())
assert(configMap.getData.containsKey(SPARK_CONF_FILE_NAME))
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf1key=conf1value"))
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf2key=conf2value"))
}
+ test("SPARK-38079: driver pod is created (still scheduling-gated) before its
own config " +
+ "map, and the config map's single create call already carries the owner
reference") {
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+ submissionClient.run()
+ // The pod itself must be created before any pre-resource resourceList()
call, since a
+ // pre-resource's owner reference (set below) needs the pod's UID.
+ assert(gatedPod === fullExpectedPod(gatedConfigMapName))
+ verify(namedPods).create()
+
+ // The (single) resourceList(...) call creating the driver's own config
map must already
+ // carry the owner reference -- there is no separate, later "refresh"
call. The pod stays
+ // scheduling-gated (so kubelet cannot attempt to mount it) for as long as
this has not yet
+ // happened, avoiding the "configmap ... not found" mount race
(SPARK-38079) without ever
+ // creating the config map ownerless.
+ val resourceListCall = createdResourcesArgumentCaptor.getAllValues.get(0)
+ val configMaps = resourceListCall
+ .filter(_.isInstanceOf[ConfigMap]).map(_.asInstanceOf[ConfigMap])
+ assert(configMaps.nonEmpty,
+ "the driver's own config map must be created as a pre-resource")
+ val ownerReferences = configMaps.head.getMetadata.getOwnerReferences
Review Comment:
**Two points from [round
1](https://github.com/apache/spark/pull/58187#discussion_r4129910551) are still
open here.**
- `addOwnerReference` mutates the resources in place, and the captor keeps
references to the same objects, so this reads the owner reference after `run()`
returns. If L189 and L190 were swapped, production would create the conf
ConfigMap and the Kerberos and credentials Secrets without owner references,
because fabric8 copies the items in `resourceList(...)` and sends them right
away. Yet this check and the one at L549 would still pass; only `checkPodGroup`
in the Volcano integration test would notice, for the PodGroup alone. Recording
`getOwnerReferences` inside a `doAnswer` on `resourceList(...)` would check the
state at call time.
- `ArgumentMatchers.any[Array[HasMetadata]]()` is the no-arg `any()`, which
"no longer matches varargs" since Mockito 5.0.0 ([ArgumentMatchers.java
L129](https://github.com/mockito/mockito/blob/v5.23.0/mockito-core/src/main/java/org/mockito/ArgumentMatchers.java#L129)):
it matches only calls with exactly one resource. So `never()` at L371 doesn't
see a call with zero or several resources; master's order, which calls
`resourceList(Nil: _*)` before `create()` with this fixture, would pass it. And
with `BUILT_KUBERNETES_SPEC_WITH_PRERES`, the `InOrder` at L286 would bind to
the post-resource call and fail.
`ArgumentMatchers.any(classOf[Array[HasMetadata]]): _*` matches the whole array.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
Review Comment:
**A few of the new comments describe behavior the code doesn't have, or the
earlier shutdown-hook design.**
- L155 says the gate is removed "right before the watch loop", but the
post-resources are applied in between (L221-L230). L155-L157 also point to the
gate-removal step for the ownerless-window reasoning, which is at L178-L187,
and L157 and L205-L206 still mention the shutdown hook.
- L184-L187 say that on failure none of the pre-resources that reached the
server "are left referencing anything", but L189 already set the pod's owner
reference on them; that is why GC removes them once the pod is deleted.
- L129-L131 say the `nodeName` fallback "has the same effect" (see the
comment at L135).
- The comments in the four feature steps (`HadoopConfDriverFeatureStep`
L115-L116, `KerberosConfDriverFeatureStep` L228-L230,
`PodTemplateConfigMapStep` L77-L78, `DriverKubernetesCredentialsFeatureStep`
L165-L169) and in `HadoopConfDriverFeatureStepSuite` say "before the pod
itself", but the pod is now created first (L171). "Before the pod is
schedulable" (L163-L165) is accurate.
The diff still adds 46 comment lines and 40 code lines to this file, and 71
and 46 across the main files. Could these be trimmed to a short comment at the
gate?
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilderSuite.scala:
##########
@@ -66,6 +66,12 @@ class KubernetesDriverBuilderSuite extends PodBuilderSuite {
test("SPARK-37331: check driver pre kubernetes resource, empty by default") {
val sparkConf = new SparkConf(false)
.set(Config.CONTAINER_IMAGE, "spark-driver:latest")
+ // SPARK-38079: HadoopConfDriverFeatureStep now also contributes a
pre-resource
+ // whenever HADOOP_CONF_DIR is set in the environment the test JVM
inherits, which
+ // would otherwise make this assertion depend on that environment
variable. Exclude
+ // it so this test only exercises the "empty by default" case it is
meant to check.
+ .set(Config.KUBERNETES_DRIVER_POD_EXCLUDED_FEATURE_STEPS.key,
+ "org.apache.spark.deploy.k8s.features.HadoopConfDriverFeatureStep")
Review Comment:
**Excluding only `HadoopConfDriverFeatureStep` still leaves these size
checks dependent on the environment.** `KerberosConfDriverFeatureStep` is
always included, and with this PR it returns its delegation-token Secret from
`getAdditionalPreKubernetesResources()` whenever the current UGI has tokens.
UGI loads them from `HADOOP_TOKEN_FILE_LOCATION` (or `HADOOP_TOKEN` and
`hadoop.token.files`), so with a valid token file in the environment, e.g.
inside a YARN container, both tests get an extra `<prefix>-delegation-tokens`
pre-resource and fail. Could we also exclude
`org.apache.spark.deploy.k8s.features.KerberosConfDriverFeatureStep`? The
config takes a comma-separated list.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
.endSpec()
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
- // setup resources before pod creation
- val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources
+ // SPARK-38079: the driver's own base config map (mounted as
SPARK_CONF_VOLUME_DRIVER
+ // above) must also be created before the pod is schedulable, to avoid a
"configmap ...
+ // not found" mount race between the driver pod and the config map it
depends on.
+ val preKubernetesResources =
resolvedDriverSpec.driverPreKubernetesResources ++ Seq(configMap)
Review Comment:
**These changes apply to every cluster-mode submission, with no switch back
to the previous flow and no migration guide entry.** The scheduling gate, the
`nodeName` rewrite, the pod `patch` and the new order of pre-resources are all
unconditional. A user hit by the cases at L171, L208 or L135 has no workaround
other than changing the cluster or downgrading. Other Kubernetes behavior
changes in `docs/core-migration-guide.md` come with "To restore the legacy
behavior, you can set ...", e.g. `spark.kubernetes.executor.useDriverPodIP`.
The PR description also still says the resources "are created *before* the
driver pod" and "No behavior change for successful submissions", and that
`VolcanoFeatureStep` is "intentionally left unchanged", while its PodGroup is
now created after the pod too. Since the description becomes the commit
message, could you update it, including the user-facing change section, as
asked in [round
1](https://github.com/apache/spark/pull/58187#discussion_r4129910571)?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -209,6 +257,26 @@ private[spark] class Client(
log"submission ID ${MDC(SUBMISSION_ID, sId)} into Kubernetes")
}
}
+
+ // SPARK-38079: best-effort cleanup for the two failure catch blocks between
pod creation and
+ // gate removal. The pod and each pre-resource are deleted independently --
every delete call
+ // wrapped in its own Utils.tryLogNonFatalError -- so that one of them
failing (e.g. a delete
+ // call itself hitting a permission or network error) neither masks the
original exception
+ // the caller is about to (re)throw, nor skips the rest. In particular,
fabric8's
+ // resourceList(...).delete() deletes the given items sequentially and stops
at the first
+ // non-404 exception, so pre-resources are deleted one at a time here rather
than as a single
+ // resourceList(...).delete() call, which could otherwise leave every
pre-resource after the
+ // one that failed undeleted.
+ private def deletePodAndPreResources(pod: Pod, preResources:
Seq[HasMetadata]): Unit = {
+ Utils.tryLogNonFatalError {
+ kubernetesClient.pods().resource(pod).delete()
Review Comment:
**This isn't introduced by this PR, but the post-resource catch block
(L226-L229) still calls
`kubernetesClient.pods().resource(createdDriverPod).delete()` without this
guard.** If the post-resource apply fails and that delete also fails, e.g. on a
dropped connection or a webhook or RBAC denying the DELETE, the delete's
exception replaces the original one, which is then neither thrown nor logged.
Since this PR edits that block and guards the same delete here for the other
two catch blocks, could we wrap that one in `Utils.tryLogNonFatalError` as well?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -131,48 +147,80 @@ private[spark] class Client(
.withName(configMapName)
.endConfigMap()
.endVolume()
+ .addToNodeSelector(
+ pinnedNodeName.map(nodeName => Map("kubernetes.io/hostname" ->
nodeName))
+ .getOrElse(Map.empty[String, String]).asJava)
+ // SPARK-38079: holds the pod unschedulable -- so kubelet cannot
attempt to mount
+ // anything on it -- until its pre-resources exist (see below). Only
removed once
+ // that is true, right before the watch loop; see the "Remove the
pre-resources
+ // scheduling gate" step below for why this makes an
ownerless-resource window,
+ // and its consequences (a shutdown hook), unnecessary.
+ .addNewSchedulingGate(PRE_RESOURCES_SCHEDULING_GATE)
Review Comment:
**Gating the driver may also strand it with schedulers that don't support
scheduling gates.** For example, the YuniKorn shim before 1.7.0 creates a task
for the gated pod right away. If it allocates before L208 removes the gate,
which takes a few scheduling cycles and so needs a slow API server or a shorter
`service.schedulingInterval`, the bind is rejected (`pod ... has non-empty
.spec.schedulingGates`), the task fails, and the driver can stay `Pending`
after the gate is removed. This is narrow, since YuniKorn 1.6.x supports
Kubernetes only up to 1.31 and the docs now install 1.9.0, which skips gated
pods. It may still be worth a note in the docs, together with the `patch`
requirement.
--
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]
