dongjoon-hyun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4340056240
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala:
##########
@@ -673,6 +673,17 @@ private[spark] object Config extends Logging {
.booleanConf
.createWithDefault(true)
+ val USE_DRIVER_SCHEDULING_GATE =
+ ConfigBuilder("spark.kubernetes.submission.useDriverSchedulingGate")
+ .doc("If true, in cluster mode the driver pod is created with a
scheduling gate that " +
+ "is removed once the resources the driver mounts exist. Removing the
gate needs the " +
+ "`patch` verb on pods, and the scheduler must honor scheduling gates.
If false, the " +
+ "driver pod can be scheduled before those resources exist.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
**On your open question about this switch: I'd keep it, but default it to
`false` in 4.4.0.** With `true`, every cluster-mode `spark-submit` needs
`patch` on pods and a scheduler that honors gates, while the PR description
says the race is rare ("without that delay the previous code reported none in
20") and fixes itself ("The kubelet retries the mount, so the driver recovers;
the cost is a short delay and a warning event").
- A submitter without `patch` fails every submission (the gated pod is
created, the PATCH gets a 403, and the pod is deleted), as you reproduced. For
example, the kubeflow/spark-operator controller role grants 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)).
Its image still bundles Spark 4.0.4 ([Dockerfile
L17](https://github.com/kubeflow/spark-operator/blob/v2.5.2/Dockerfile#L17)),
so moving it to 4.4 would need an RBAC change first, as
[kubeflow/spark-operator#3240](https://github.com/kubeflow/spark-operator/pull/3240)
did for the NetworkPolicy of 4.2.
- The [versioning policy](https://spark.apache.org/versioning-policy.html)
says feature releases have "No behavior or semantic changes (e.g. SQL
semantics, execution behavior, configuration defaults)", with exceptions "case
by case (e.g., security issues)". This is not a security fix, and unlike
`spark.kubernetes.executor.useDriverPodIP`, it adds a failure mode to
deployments that work today.
- The switch itself is needed. The gate lives in `Client.run()` rather than
in a feature step, so `spark.kubernetes.driver.pod.excludedFeatureSteps` can't
turn it off, and it is the only way out for the case at KCA L141 and for
YuniKorn before 1.7. kubeflow passes `spec.sparkConf` through to
`spark-submit`, so its users can set it per application.
So my suggestion is `createWithDefault(false)` with `.version("4.4.0")`, the
RBAC bullet and the Pod Spec rows made conditional on it, the K8s integration
tests run with the gate on so it stays covered, and the default flipped in a
separate master-only (5.0.0) change with its own migration guide entry. Keeping
`true` and making this PR master-only would be consistent too. This refines my
[round-2
comment](https://github.com/apache/spark/pull/58187#discussion_r4300454516),
where I only asked for a way back to the previous flow.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -180,28 +210,52 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
when(podOperations.inNamespace(kconf.namespace)).thenReturn(podsWithNamespace)
when(podsWithNamespace.withName(POD_NAME)).thenReturn(namedPods)
+ events.clear()
createdPodArgumentCaptor = ArgumentCaptor.forClass(classOf[Pod])
createdResourcesArgumentCaptor =
ArgumentCaptor.forClass(classOf[Array[HasMetadata]])
- when(podsWithNamespace.resource(fullExpectedPod())).thenReturn(namedPods)
+ patchContextCaptor = ArgumentCaptor.forClass(classOf[PatchContext])
+ patchBodyCaptor = ArgumentCaptor.forClass(classOf[String])
+
when(podsWithNamespace.resource(createdPodArgumentCaptor.capture())).thenReturn(namedPods)
+
when(podOperations.resource(ArgumentMatchers.any[Pod]())).thenReturn(namedPods)
Review Comment:
**Since this stub returns the same `namedPods` for any pod, no test checks
which pod object the gate PATCH (KCA L206) and its failure-path DELETE (KCA
L214) go through.** Replacing `resource(createdDriverPod)` with
`resource(podToCreate)` there passes every test. The UID check at L453-L455
covers only the post-resource delete at KCA L197.
That matters because `podToCreate` keeps the pod template's
`metadata.namespace`. `loadPodFromTemplate` reads the template as is
(`PodBuilderSuite` L190 expects the template's namespace to survive), and only
the CREATE's `inNamespace(conf.namespace)` replaces it, on a copy. With a
template in `team-a` and `spark.kubernetes.namespace=spark`, such a change
would send the PATCH to `team-a` and fail with a 403 or 404 after all resources
were created. With L214 changed too, it would leave the gated driver behind.
The code here is right. Could we capture the `podOperations.resource(...)`
arguments in the L319 and L472 tests and check the UID, as L453-L455 does?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -135,6 +136,21 @@ private[spark] class Client(
.build()
val driverPodName = resolvedDriverPod.getMetadata.getName
+ // The API server rejects a pod that has both spec.nodeName and scheduling
gates, and a pod
+ // with spec.nodeName skips the scheduler anyway, so only pods the
scheduler places are gated.
+ val gateDriverPod = conf.get(USE_DRIVER_SCHEDULING_GATE) &&
Review Comment:
**On a cluster whose mutating admission sets `spec.nodeName`, every gated
driver pod is now rejected at CREATE.** This check looks at the pod Spark
builds, before admission, but the API server validates the pod after the
mutating webhooks have run, and `ValidatePodCreate` rejects a pod that has both
`nodeName` and scheduling gates ([validation.go
L5609-L5612](https://github.com/kubernetes/kubernetes/blob/v1.35.0/pkg/apis/core/validation/validation.go#L5609-L5612)).
For example, Portainer KubeSolo v1.2.2 (Kubernetes 1.35.7-k3s1) has no
kube-scheduler. Its NodeSetter webhook "sets `spec.nodeName` on every new pod"
([README
L39](https://github.com/portainer/kubesolo/blob/v1.2.2/README.md?plain=1#L39))
on pod CREATE ([config.go
L34-L45](https://github.com/portainer/kubesolo/blob/v1.2.2/pkg/kubernetes/webhook/config.go#L34-L45),
[webhooks.go
L133-L135](https://github.com/portainer/kubesolo/blob/v1.2.2/pkg/kubernetes/webhook/webhooks.go#L133-L135)).
With the default configuration:
- every cluster-mode submission fails with `Pod "<name>" is invalid:
spec.nodeName: Forbidden: cannot be set until all schedulingGates have been
cleared` (422), where master succeeds;
- the catch block at L171-L175 logs the `kubectl auth can-i create pod`
hint, which doesn't apply;
- granting `patch` on pods, which the migration guide offers, doesn't help,
because the CREATE fails first. Only
`spark.kubernetes.submission.useDriverSchedulingGate=false` or a `nodeName` in
the pod template avoids it.
Could we mention this case in the docs next to the YuniKorn note? It is also
one more reason for the default discussed at `Config.scala` L685.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -223,12 +277,210 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
val configMap = configMaps.head
assert(configMap.getMetadata.getName ===
KubernetesClientUtils.configMapNameDriver)
+ assert(configMap.getMetadata.getName === mountedConfigMapName)
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: the driver pod is created gated, and the gate is removed
last") {
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list()", "apply",
+ "create pod",
+ "list()", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(DRIVER_RESOURCES_SCHEDULING_GATE))
+ }
+
+ test("SPARK-38079: pre resources are applied before the gated pod and owned
after it") {
+ val preResKconf = KubernetesTestConf.createDriverConf(
+ resourceNamePrefix = Some(KUBERNETES_RESOURCE_PREFIX))
+ when(driverBuilder.buildFromFeatures(preResKconf, kubernetesClient))
+ .thenReturn(BUILT_KUBERNETES_SPEC_WITH_PRERES)
+ kconf = preResKconf
+
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list(CustomResourceDefinition/preCRD)", "apply",
+ "create pod",
+ "list(CustomResourceDefinition/preCRD+owner)", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+ }
+
+ test("SPARK-38079: the gate is removed by name with an idempotent strategic
merge patch") {
+ newClient().run()
+
+ verify(namedPods).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(patchContextCaptor.getValue.getPatchType ===
PatchType.STRATEGIC_MERGE)
+ assert(patchBodyCaptor.getValue ===
+ s"""{"spec":{"schedulingGates":[{"$$patch":"delete",""" +
+ s""""name":"$DRIVER_RESOURCES_SCHEDULING_GATE"}]}}""")
+ }
+
+ test("SPARK-38079: gates already on the pod template are kept next to the
Spark gate") {
+ val templateGate = "example.com/keep-me"
+ val podWithGate = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .addNewSchedulingGate(templateGate)
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(podWithGate, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(templateGate, DRIVER_RESOURCES_SCHEDULING_GATE))
+ assert(!patchBodyCaptor.getValue.contains(templateGate))
+ }
+
+ test("SPARK-38079: the gate patch leaves a gate it does not name in place") {
+ // A fake API server that applies the patch body: only the named gate may
disappear.
Review Comment:
**Two new comments claim server-side behavior that these mocks can't show.**
- Here, nothing applies the patch body. `remaining` (L374-L375) is the
created pod's gates minus the name the test has just asserted, and every
assertion follows from L325-L327 (the body is a constant) and L346-L347. The
test never reads the patch type either. With `PatchType.JSON_MERGE`, a real API
server would replace the whole list with Spark's gate alone, dropping the
template's gate, yet this test passes; only L324 in the L319 test catches that.
Could we fold what this test is meant to show into the L330 test and drop it,
together with the module's only `ObjectMapper`?
- L452 says the deleted pod is "the one this submission created, not some
other pod of that name", but `resource(createdDriverPod).delete()` sends
`DELETE .../pods/<name>` with no UID precondition ([OperationSupport
L300-L308](https://github.com/fabric8io/kubernetes-client/blob/v7.9.0/kubernetes-client/src/main/java/io/fabric8/kubernetes/client/dsl/internal/OperationSupport.java#L300-L308)),
the same request as `withName(name).delete()`. L453-L455 check which object
reached the mock, not which pod the server deletes. Something like "The delete
goes through the pod returned by create()" would be accurate.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -223,12 +277,210 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
val configMap = configMaps.head
assert(configMap.getMetadata.getName ===
KubernetesClientUtils.configMapNameDriver)
+ assert(configMap.getMetadata.getName === mountedConfigMapName)
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: the driver pod is created gated, and the gate is removed
last") {
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list()", "apply",
+ "create pod",
+ "list()", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(DRIVER_RESOURCES_SCHEDULING_GATE))
+ }
+
+ test("SPARK-38079: pre resources are applied before the gated pod and owned
after it") {
+ val preResKconf = KubernetesTestConf.createDriverConf(
+ resourceNamePrefix = Some(KUBERNETES_RESOURCE_PREFIX))
+ when(driverBuilder.buildFromFeatures(preResKconf, kubernetesClient))
+ .thenReturn(BUILT_KUBERNETES_SPEC_WITH_PRERES)
+ kconf = preResKconf
+
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list(CustomResourceDefinition/preCRD)", "apply",
+ "create pod",
+ "list(CustomResourceDefinition/preCRD+owner)", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+ }
+
+ test("SPARK-38079: the gate is removed by name with an idempotent strategic
merge patch") {
+ newClient().run()
+
+ verify(namedPods).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(patchContextCaptor.getValue.getPatchType ===
PatchType.STRATEGIC_MERGE)
+ assert(patchBodyCaptor.getValue ===
+ s"""{"spec":{"schedulingGates":[{"$$patch":"delete",""" +
+ s""""name":"$DRIVER_RESOURCES_SCHEDULING_GATE"}]}}""")
+ }
+
+ test("SPARK-38079: gates already on the pod template are kept next to the
Spark gate") {
+ val templateGate = "example.com/keep-me"
+ val podWithGate = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .addNewSchedulingGate(templateGate)
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(podWithGate, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(templateGate, DRIVER_RESOURCES_SCHEDULING_GATE))
+ assert(!patchBodyCaptor.getValue.contains(templateGate))
+ }
+
+ test("SPARK-38079: the gate patch leaves a gate it does not name in place") {
+ // A fake API server that applies the patch body: only the named gate may
disappear.
+ val templateGate = "example.com/keep-me"
+ val podWithGate = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .addNewSchedulingGate(templateGate)
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(podWithGate, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+ val patch = new com.fasterxml.jackson.databind.ObjectMapper()
+ .readTree(patchBodyCaptor.getValue)
+ val ops = patch.at("/spec/schedulingGates").elements().asScala.toSeq
+ assert(ops.size === 1, "the patch must name exactly one gate")
+ assert(ops.head.get("$patch").asText === "delete")
+ assert(ops.head.get("name").asText === DRIVER_RESOURCES_SCHEDULING_GATE)
+ val remaining =
createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala
+ .map(_.getName).filterNot(_ == ops.head.get("name").asText)
+ assert(remaining === Seq(templateGate))
+ }
+
+ test("SPARK-38079: disabling the scheduling gate keeps the previous flow and
needs no patch") {
+ val sparkConf = new SparkConf(false)
+ .set(Config.USE_DRIVER_SCHEDULING_GATE, false)
+ kconf = KubernetesTestConf.createDriverConf(
+ sparkConf = sparkConf, resourceNamePrefix =
Some(KUBERNETES_RESOURCE_PREFIX))
+ when(driverBuilder.buildFromFeatures(kconf,
kubernetesClient)).thenReturn(BUILT_KUBERNETES_SPEC)
+
+ newClient().run()
+
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.isEmpty)
+ verify(namedPods, never()).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(!events.contains("remove gate"))
+ }
+
+ test("SPARK-38079: a pod pinned with spec.nodeName is neither gated nor
patched") {
+ val pinnedPod = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .withNodeName("some-node")
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(pinnedPod, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+ val created = createdPodArgumentCaptor.getValue
+ assert(created.getSpec.getNodeName === "some-node")
+ assert(created.getSpec.getSchedulingGates.isEmpty)
+ verify(namedPods, never()).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(events.toSeq === Seq(
+ "list()", "apply",
+ "create pod",
+ "list()", "apply",
+
s"list(Secret/secret+owner,ConfigMap/${KubernetesClientUtils.configMapNameDriver}+owner)",
+ "apply"))
+ }
+
+ test("SPARK-38079: a failure creating the pod deletes nothing but the pre
resources") {
+ val failure = new RuntimeException("create failed")
+ doAnswer(answer {
+ events += "create pod"
+ throw failure
+ }).when(namedPods).create()
+
+ val thrown = intercept[RuntimeException](newClient().run())
+
+ assert(thrown eq failure)
+ verify(namedPods, never()).delete()
+ verify(namedPods, never()).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(events.toSeq === Seq("list()", "apply", "create pod", "list()",
"delete resources"))
+ }
+
+ test("SPARK-38079: a failure applying the post resources deletes only the
pod") {
+ val failure = new RuntimeException("apply failed")
+ val failing = mock[RESOURCE_LIST]
+ when(failing.forceConflicts()).thenReturn(failing)
+ when(failing.serverSideApply()).thenThrow(failure)
+ resourceListFor = resources => if
(resources.exists(_.isInstanceOf[Secret])) failing
+ else resourceList
+
+ val thrown = intercept[RuntimeException](newClient().run())
+
+ assert(thrown eq failure)
+ assert(events.last === "delete pod")
+ assert(!events.contains("remove gate"))
+ assert(!events.contains("delete resources"))
+ // The pod deleted is the one this submission created, not some other pod
of that name.
+ val deleted = ArgumentCaptor.forClass(classOf[Pod])
+ verify(podOperations).resource(deleted.capture())
+ assert(deleted.getValue.getMetadata.getUid === DRIVER_POD_UID)
+ }
+
+ test("SPARK-38079: a failure deleting the pod does not hide the original
failure") {
+ val failure = new RuntimeException("apply failed")
+ val failing = mock[RESOURCE_LIST]
+ when(failing.forceConflicts()).thenReturn(failing)
+ when(failing.serverSideApply()).thenThrow(failure)
+ resourceListFor = resources => if
(resources.exists(_.isInstanceOf[Secret])) failing
+ else resourceList
+ when(namedPods.delete()).thenThrow(new RuntimeException("delete failed"))
+
+ val thrown = intercept[RuntimeException](newClient().run())
+
+ assert(thrown eq failure)
+ }
+
+ test("SPARK-38079: a failure removing the gate deletes the pod and
propagates") {
+ val failure = new RuntimeException("patch failed")
+ when(namedPods.patch(ArgumentMatchers.any[PatchContext](),
ArgumentMatchers.any[String]()))
Review Comment:
**No test makes the DELETE fail after a failed gate PATCH, so the new guard
at KCA L213-L215 is untested, and this `when(...).thenThrow` re-stub runs the
`before` answer.**
- The only delete-failure test (L458) fails the post-resource apply, so it
never reaches the PATCH, and in this test the delete succeeds. Replacing
L213-L215 with a bare delete keeps all 17 tests green, although a failed DELETE
would then replace the PATCH failure. The PR description's "each failure path"
is also broader than the tests: the pre-resource apply failure (KCA L156-L164)
and the owner-refresh failure (KCA L178-L187) have none.
- `when(namedPods.patch(any, any))` here and `when(namedPods.delete())` at
L465 call the mock while stubbing, so Mockito runs the L234-L242 answers
([MockHandlerImpl
L91-L103](https://github.com/mockito/mockito/blob/v5.23.0/mockito-core/src/main/java/org/mockito/internal/handler/MockHandlerImpl.java#L91-L103)).
`events` then starts with "remove gate" or "delete pod", the patch captors
hold `null`, and the call that `run()` really makes isn't recorded. No
assertion in these two tests reads them today, but the checks the other tests
use, such as `!events.contains("remove gate")` or `events.last === "delete
pod"`, would give the wrong answer here.
Could we stub these with `doAnswer(answer { events += "remove gate"; throw
failure }).when(namedPods).patch(...)`, as L424 does for `create()`, and add a
case where both the PATCH and the DELETE fail?
##########
docs/core-migration-guide.md:
##########
@@ -26,6 +26,8 @@ license: |
- Since Spark 4.4, reading an RDD checkpoint ignores names beginning with
`part-` unless the suffix consists of one or more ASCII digits. Previously,
names such as `part-00000.bak`, `part-backup`, or `part-` caused checkpoint
recovery to fail with `NumberFormatException`. Validation of numeric partition
names and the contiguous partition sequence is unchanged.
+- Since Spark 4.4, in Kubernetes cluster mode `spark-submit` creates the
driver pod with the scheduling gate `spark.apache.org/driver-resources` and
removes it once the resources the driver mounts exist, so the driver can no
longer start before its config map. Removing the gate requires the `patch` verb
on pods, which the account running `spark-submit` may not have; without it the
submission fails and the driver pod is deleted. A driver pod whose template
sets `spec.nodeName` is not gated and behaves as before. To restore the legacy
behavior, you can set `spark.kubernetes.submission.useDriverSchedulingGate` to
`false`, or grant `patch` on pods to the account that runs `spark-submit`.
Review Comment:
**This entry has three inaccuracies.**
- "To restore the legacy behavior, you can set ... to `false`, or grant
`patch` on pods": granting `patch` keeps the gated flow and only lets it
succeed. Every other "To restore the legacy behavior" entry in this file names
only the setting that reverts the change.
- It omits the scheduler requirement that the RBAC bullet, the config row,
the `Config.scala` doc and the PR description all state. That is the case an
upgrader can't diagnose: on YuniKorn before 1.7, granting `patch` turns the 403
into a driver that stays `Pending`.
- "so the driver can no longer start before its config map": it never could.
The kubelet waits for the conf volume before it starts any container
([kubelet.go
L2120](https://github.com/kubernetes/kubernetes/blob/v1.35.0/pkg/kubelet/kubelet.go#L2120)).
What changes is that the pod is no longer scheduled before its config map
exists, which avoids the transient `FailedMount` events.
If the default stays `true` (see `Config.scala` L685), how about something
like this?
```suggestion
- Since Spark 4.4, in Kubernetes cluster mode `spark-submit` creates the
driver pod with the scheduling gate `spark.apache.org/driver-resources` and
removes it once the resources it creates for the driver exist, so the driver
pod is no longer scheduled before its config map exists. Removing the gate
requires the `patch` verb on pods, which the account running `spark-submit` may
not have; without it the submission fails and the driver pod is deleted. A
scheduler that does not honor scheduling gates may leave the gated driver pod
pending; for example, Apache YuniKorn supports `spec.schedulingGates` since
v1.7.0. A driver pod that sets `spec.nodeName` is not gated and behaves as
before. To restore the legacy behavior, you can set
`spark.kubernetes.submission.useDriverSchedulingGate` to `false`. To keep the
gate, grant `patch` on pods to the account that runs `spark-submit`.
```
##########
docs/running-on-kubernetes.md:
##########
@@ -56,6 +56,12 @@ you may set up a test cluster on your local machine using
* You must have appropriate permissions to list, create, edit and delete
[pods](https://kubernetes.io/docs/concepts/workloads/pods/) in your cluster.
You can verify that you can list these resources
by running `kubectl auth can-i <list|create|edit|delete> pods`.
+ * In cluster mode, `spark-submit` also needs the `patch` verb on pods: it
creates the driver pod with a
+ [scheduling
gate](https://kubernetes.io/docs/concepts/scheduling-eviction/pod-scheduling-readiness/)
and removes
+ the gate once the resources the driver mounts exist. A driver pod whose
template sets `spec.nodeName` is not gated.
Review Comment:
**The gate waits only for what `spark-submit` creates for the driver, but
this line and six other places say it is removed once "the resources the driver
mounts" exist.** `run()` removes it right after applying the pre-resources,
`driverKubernetesResources` and the conf ConfigMap (KCA L155-L209). Objects the
driver mounts that Spark doesn't create aren't waited for: Secrets from
`spark.kubernetes.driver.secrets.*` (`MountSecretsFeatureStep` only references
them), `spark.kubernetes.hadoop.configMapName`,
`spark.kubernetes.kerberos.krb5.configMapName`, and the volumes of a pod
template. If one of those is created later, e.g. by another controller, the
kubelet still reports `FailedMount` with the gate on, which these docs say
can't happen.
The behavior is fine; only the wording promises more. Could we say "once the
resources `spark-submit` creates for the driver, such as its config map, exist"
here, at L1345-L1348 and L2316, in the `Config.scala` doc (L678-L681), the
migration guide, `Constants.scala` L95 and KCA L202? The PR title has the same
wording.
##########
docs/running-on-kubernetes.md:
##########
@@ -2276,13 +2293,30 @@ See the below table for the full list of pod
specifications that will be overwri
Additional node selectors will be added from the spark configuration to
both executor pods.
</td>
</tr>
+<tr>
+ <td>nodeName</td>
+ <td>Kept as the template sets it</td>
+ <td>
+ A driver pod whose template sets <code>nodeName</code> is created without
the scheduling gate described in
+ <code>spark.kubernetes.submission.useDriverSchedulingGate</code>, because
Kubernetes rejects a pod with both.
+ The kubelet can therefore try to mount the driver's resources before they
exist.
+ </td>
+</tr>
<tr>
<td>restartPolicy</td>
<td><code>"never"</code></td>
<td>
Spark assumes that both drivers and executors never restart.
</td>
</tr>
+<tr>
+ <td>schedulingGates</td>
+ <td>Adds <code>spark.apache.org/driver-resources</code> to driver pods</td>
Review Comment:
**A few smaller doc points around the gate:**
- This row says Spark adds the gate to driver pods unconditionally, and the
config doc (`Config.scala` L678-L681 and L1345 here) says "If true, ... created
with a scheduling gate" without the `nodeName` exception. "unless the pod sets
`nodeName` or `spark.kubernetes.submission.useDriverSchedulingGate` is `false`"
would cover both.
- The `nodeName` row (L2297) sits after `nodeSelector` (L2290) in an
otherwise alphabetical table. I asked for this row in round 2, when `nodeName`
was still rewritten, so this is only about the order.
- L58 still suggests `kubectl auth can-i <list|create|edit|delete> pods`,
and `edit` isn't an RBAC verb, so the documented check doesn't show the new
requirement, while the error log at KCA L212 asks for `kubectl auth can-i patch
pod`. `<list|create|patch|delete>` would line them up.
- L61, L2300 and the migration guide say a driver pod "whose template sets"
`nodeName` is not gated, but KCA L141-L142 check the resolved pod, which
includes user feature steps from `spark.kubernetes.driver.pod.featureSteps`. "A
driver pod that sets `spec.nodeName`" would be accurate.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -177,10 +193,30 @@ private[spark] class Client(
kubernetesClient.resourceList(otherKubernetesResources:
_*).forceConflicts().serverSideApply()
} catch {
case NonFatal(e) =>
- kubernetesClient.pods().resource(createdDriverPod).delete()
+ Utils.tryLogNonFatalError {
+ kubernetesClient.pods().resource(createdDriverPod).delete()
+ }
throw e
}
+ // Every resource the driver mounts exists now, so let the scheduler place
the pod. The
+ // strategic merge patch names the gate, so it is idempotent and leaves
other gates alone.
+ if (gateDriverPod) {
+ try {
+ kubernetesClient.pods().resource(createdDriverPod).patch(
+ PatchContext.of(PatchType.STRATEGIC_MERGE),
+ s"""{"spec":{"schedulingGates":[""" +
+
s"""{"$$patch":"delete","name":"$DRIVER_RESOURCES_SCHEDULING_GATE"}]}}""")
+ } catch {
+ case NonFatal(e) =>
+ logError("Please check \"kubectl auth can-i patch pod\" first. It
should be yes.")
Review Comment:
**Two small diagnostics points about the gate's failure paths:**
- This hint is logged for every `NonFatal` failure of the PATCH, including a
timeout or a connection failure after fabric8's retries, a 5xx, or a 404 when
the pod was deleted in the meantime. It matches the existing hints at L160 and
L174, so this is optional, but logging it only for a
`KubernetesClientException` with `getCode == 403`, and mentioning
`spark.kubernetes.submission.useDriverSchedulingGate=false` there, would point
users to both fixes.
- L147 appends the gate even if the resolved pod already has one named
`spark.apache.org/driver-resources` (from a template or a feature step). The
API server then rejects the duplicate at CREATE (`spec.schedulingGates[1]:
Duplicate value`), with the `can-i create pod` hint. That is unlikely, but
removing a gate of the same name before adding ours would make such a pod work,
since the PATCH then clears it.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -223,12 +277,210 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
val configMap = configMaps.head
assert(configMap.getMetadata.getName ===
KubernetesClientUtils.configMapNameDriver)
+ assert(configMap.getMetadata.getName === mountedConfigMapName)
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: the driver pod is created gated, and the gate is removed
last") {
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list()", "apply",
+ "create pod",
+ "list()", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
Review Comment:
**These call-time `+owner` markers read suite-level fixtures that `run()`
mutates and `before` never resets, so the `Secret/secret+owner` part can't fail
in a full run.** `ADDITIONAL_RESOURCES`, `PRE_RESOURCES` and the two
`BUILT_KUBERNETES_SPEC*` specs are `val`s, ScalaTest runs the suite on one
instance in declaration order, and `addOwnerReference` calls
`setOwnerReferences` on the objects themselves. Once the tests at L256 and L262
have run, the shared Secret keeps its owner reference. Only the ConfigMap,
which `run()` builds fresh, is really checked at call time.
- A change at KCA L191-L193 that sets the owner reference on
`Seq(configMap)` before the apply, but on `driverKubernetesResources` only
after it, passes all 17 tests in a full run. Production would then create the
driver Service, the Kerberos and credentials Secrets, OnDemand PVCs and the
NetworkPolicy without owners. L287, L301 and L394 catch it only when they run
first, e.g. with `-z`.
- The other way round, L312 expects an unowned `preCRD` only because L301 is
declared before the SPARK-37331 test (L484), which mutates the same CRD.
Declaring it later makes L312 fail.
The swap from [round
2](https://github.com/apache/spark/pull/58187#discussion_r4300454498) is still
caught through the ConfigMap, so this is incomplete rather than wrong. Turning
the four fixtures into `def`s would fix it.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -223,12 +277,210 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
val configMap = configMaps.head
assert(configMap.getMetadata.getName ===
KubernetesClientUtils.configMapNameDriver)
+ assert(configMap.getMetadata.getName === mountedConfigMapName)
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: the driver pod is created gated, and the gate is removed
last") {
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list()", "apply",
+ "create pod",
+ "list()", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(DRIVER_RESOURCES_SCHEDULING_GATE))
+ }
+
+ test("SPARK-38079: pre resources are applied before the gated pod and owned
after it") {
+ val preResKconf = KubernetesTestConf.createDriverConf(
+ resourceNamePrefix = Some(KUBERNETES_RESOURCE_PREFIX))
+ when(driverBuilder.buildFromFeatures(preResKconf, kubernetesClient))
+ .thenReturn(BUILT_KUBERNETES_SPEC_WITH_PRERES)
+ kconf = preResKconf
+
+ newClient().run()
+
+ val configMapName = KubernetesClientUtils.configMapNameDriver
+ assert(events.toSeq === Seq(
+ "list(CustomResourceDefinition/preCRD)", "apply",
+ "create pod",
+ "list(CustomResourceDefinition/preCRD+owner)", "apply",
+ s"list(Secret/secret+owner,ConfigMap/$configMapName+owner)", "apply",
+ "remove gate"))
+ }
+
+ test("SPARK-38079: the gate is removed by name with an idempotent strategic
merge patch") {
+ newClient().run()
+
+ verify(namedPods).patch(
+ ArgumentMatchers.any[PatchContext](), ArgumentMatchers.any[String]())
+ assert(patchContextCaptor.getValue.getPatchType ===
PatchType.STRATEGIC_MERGE)
+ assert(patchBodyCaptor.getValue ===
+ s"""{"spec":{"schedulingGates":[{"$$patch":"delete",""" +
+ s""""name":"$DRIVER_RESOURCES_SCHEDULING_GATE"}]}}""")
+ }
+
+ test("SPARK-38079: gates already on the pod template are kept next to the
Spark gate") {
+ val templateGate = "example.com/keep-me"
+ val podWithGate = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .addNewSchedulingGate(templateGate)
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(podWithGate, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala.map(_.getName)
===
+ Seq(templateGate, DRIVER_RESOURCES_SCHEDULING_GATE))
+ assert(!patchBodyCaptor.getValue.contains(templateGate))
+ }
+
+ test("SPARK-38079: the gate patch leaves a gate it does not name in place") {
+ // A fake API server that applies the patch body: only the named gate may
disappear.
+ val templateGate = "example.com/keep-me"
+ val podWithGate = new PodBuilder(BUILT_DRIVER_POD)
+ .editSpec()
+ .addNewSchedulingGate(templateGate)
+ .endSpec()
+ .build()
+ when(driverBuilder.buildFromFeatures(kconf, kubernetesClient)).thenReturn(
+ KubernetesDriverSpec(
+ SparkPod(podWithGate, BUILT_DRIVER_CONTAINER),
+ Nil,
+ ADDITIONAL_RESOURCES,
+ RESOLVED_JAVA_OPTIONS))
+
+ newClient().run()
+
+ val patch = new com.fasterxml.jackson.databind.ObjectMapper()
+ .readTree(patchBodyCaptor.getValue)
+ val ops = patch.at("/spec/schedulingGates").elements().asScala.toSeq
+ assert(ops.size === 1, "the patch must name exactly one gate")
+ assert(ops.head.get("$patch").asText === "delete")
+ assert(ops.head.get("name").asText === DRIVER_RESOURCES_SCHEDULING_GATE)
+ val remaining =
createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.asScala
+ .map(_.getName).filterNot(_ == ops.head.get("name").asText)
+ assert(remaining === Seq(templateGate))
+ }
+
+ test("SPARK-38079: disabling the scheduling gate keeps the previous flow and
needs no patch") {
+ val sparkConf = new SparkConf(false)
+ .set(Config.USE_DRIVER_SCHEDULING_GATE, false)
+ kconf = KubernetesTestConf.createDriverConf(
+ sparkConf = sparkConf, resourceNamePrefix =
Some(KUBERNETES_RESOURCE_PREFIX))
+ when(driverBuilder.buildFromFeatures(kconf,
kubernetesClient)).thenReturn(BUILT_KUBERNETES_SPEC)
+
+ newClient().run()
+
+
assert(createdPodArgumentCaptor.getValue.getSpec.getSchedulingGates.isEmpty)
Review Comment:
**The tests pin only the default path: no test compares the whole pod
created without the gate, and none checks that the gate is removed when
`waitAppCompletion` is `false`.** On master, the `before` stub
`resource(fullExpectedPod())` returned `null` for any other pod, so every test
pinned the created pod. Now `resource(createdPodArgumentCaptor.capture())`
(L218) matches any pod, `fullExpectedPod()` always carries the gate (L113), and
both full-pod checks (L258, L574) run with the gate on. So these changes pass
all 17 tests:
- `else resolvedDriverSpec.pod.pod` at KCA L151, which drops the Spark
container, the conf volume and `SPARK_CONF_DIR` from every opt-out or
`nodeName` submission;
- an `else` branch that clears the template's own gates;
- moving the gate block (KCA L204-L218) into the `if
(conf.get(WAIT_FOR_APP_COMPLETION))` branch. The only test with `false`,
SPARK-42813 (L602), checks only the log, and kubeflow/spark-operator always
submits with `waitAppCompletion=false` ([submission.go
L279-L285](https://github.com/kubeflow/spark-operator/blob/v2.5.2/internal/controller/sparkapplication/submission.go#L279-L285)),
so its drivers would stay gated.
This test also doesn't check the call sequence that its name promises
("keeps the previous flow"). Could we add a `gated: Boolean = true` parameter
to `fullExpectedPod` and assert `=== fullExpectedPod(gated = false)` here, the
same with `nodeName` at L394 and with a template gate that must survive, reuse
the L414-L419 sequence here, and assert `events.last === "remove gate"` in
SPARK-42813?
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/KubernetesClientApplication.scala:
##########
@@ -151,7 +167,7 @@ private[spark] class Client(
var createdDriverPod: Pod = null
try {
createdDriverPod =
-
kubernetesClient.pods().inNamespace(conf.namespace).resource(resolvedDriverPod).create()
+
kubernetesClient.pods().inNamespace(conf.namespace).resource(podToCreate).create()
Review Comment:
**If `spark-submit` stops between this CREATE and a successful gate PATCH,
the driver pod stays `SchedulingGated` with nothing to expire it.** For
example, `spark-submit` is killed (SIGKILL, OOM) after the post-resource apply,
or the API server becomes unreachable there and both the PATCH and the guarded
DELETE at L213-L215 fail. The pod is never bound, so it has no `startTime` and
`activeDeadlineSeconds` never applies; pod GC and the scheduler leave it alone
too.
We settled the shutdown hook in round 1, and for most of this window master
leaves a stuck pod as well (in `ContainerCreating`). What is new is the last
request: on master, a driver whose `spark-submit` died after the post-resource
apply simply ran. Also, `spark-submit --status` shows only `phase: Pending`,
which looks like a pod waiting for capacity. Could we add a sentence to the
docs saying that a driver left `SchedulingGated` by an interrupted submission
can be removed with `spark-submit --kill` or `kubectl delete pod`?
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -180,28 +210,52 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
when(podOperations.inNamespace(kconf.namespace)).thenReturn(podsWithNamespace)
when(podsWithNamespace.withName(POD_NAME)).thenReturn(namedPods)
+ events.clear()
createdPodArgumentCaptor = ArgumentCaptor.forClass(classOf[Pod])
createdResourcesArgumentCaptor =
ArgumentCaptor.forClass(classOf[Array[HasMetadata]])
- when(podsWithNamespace.resource(fullExpectedPod())).thenReturn(namedPods)
+ patchContextCaptor = ArgumentCaptor.forClass(classOf[PatchContext])
+ patchBodyCaptor = ArgumentCaptor.forClass(classOf[String])
+
when(podsWithNamespace.resource(createdPodArgumentCaptor.capture())).thenReturn(namedPods)
+
when(podOperations.resource(ArgumentMatchers.any[Pod]())).thenReturn(namedPods)
when(resourceList.forceConflicts()).thenReturn(resourceList)
+ when(resourceList.serverSideApply()).thenAnswer(answer {
+ events += "apply"
+ java.util.Collections.emptyList[HasMetadata]()
+ })
+ when(resourceList.delete()).thenAnswer(answer {
+ events += "delete resources"
+ java.util.Collections.emptyList[StatusDetails]()
+ })
when(namedPods.serverSideApply()).thenReturn(podWithOwnerReference())
- when(namedPods.create()).thenReturn(podWithOwnerReference())
+ when(namedPods.create()).thenAnswer(answer {
+ events += "create pod"
+ podWithOwnerReference()
+ })
+ when(namedPods.patch(patchContextCaptor.capture(),
patchBodyCaptor.capture()))
+ .thenAnswer(answer {
+ events += "remove gate"
+ podWithOwnerReference()
+ })
+ when(namedPods.delete()).thenAnswer(answer {
+ events += "delete pod"
+ java.util.Collections.emptyList[StatusDetails]()
+ })
when(namedPods.watch(loggingPodStatusWatcher)).thenReturn(mock[Watch])
val sId = submissionId(kconf.namespace, POD_NAME)
when(loggingPodStatusWatcher.watchOrStop(sId)).thenReturn(true)
- doReturn(resourceList)
- .when(kubernetesClient)
- .resourceList(createdResourcesArgumentCaptor.capture(): _*)
+ resourceListFor = _ => resourceList
+ doAnswer(new Answer[RESOURCE_LIST] {
Review Comment:
**This `doAnswer` replaced the only use of the `doReturn` helper at L48,
which is now unused. A few more leftovers and copies could go too:**
- the `keyToPaths` parameter of `podWithOwnerReference` (L117), whose only
non-default caller, in the `SPARK_CONF_DIR` test, was removed;
- the `s` prefix at KCA L208, which has nothing to interpolate;
- the three `KubernetesDriverSpec(SparkPod(<pod>, BUILT_DRIVER_CONTAINER),
Nil, ADDITIONAL_RESOURCES, RESOLVED_JAVA_OPTIONS)` copies (L338, L360, L401),
which are `BUILT_KUBERNETES_SPEC.copy(pod = SparkPod(<pod>,
BUILT_DRIVER_CONTAINER))`;
- the template-gate setup (L331-L342 and L353-L364) and the failing
post-resource setup (L439-L444 and L459-L464), each copied verbatim;
- `preResKconf` at L302-L306, which has the same arguments as `kconf` in
`before`;
- the `never()` checks at L389-L390, L412-L413 and L432-L434, which now
repeat what `events` already asserts (I asked for the delete one at L432 in
round 2, before `events` existed).
--
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]
