HwangDongJun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4345297466
##########
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:
Both points are addressed.
- The re-stubs use `doAnswer(answer { ...; throw failure }).when(...)` now
(`failGatePatch`, `failPodDelete`, `failPodCall`), as the `create()` case did,
so they no longer run the `before` answer and `events` holds only what `run()`
called.
- New tests: a failed PATCH followed by a failed DELETE (the PATCH failure
is the one that propagates, and the events end with `remove gate`, `delete
pod`), a failure applying the pre-resources, and a failure refreshing the
pre-resources' owner references. The last two cover the paths you listed as
untested.
Replacing the guarded delete with a bare delete passed 17 of 17 before and
now fails the PATCH-then-DELETE test.
--
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]
