HwangDongJun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4345279110
##########
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:
Taken as suggested. `fullExpectedPod` has a `gated: Boolean = true`
parameter, and these tests now compare the whole created pod:
- the default (`false`): `=== fullExpectedPod(gated = false)`, with the call
sequence;
- `false` with a template gate: the pod keeps the template's gate;
- `spec.nodeName`: the whole pod with `nodeName` and no gate, and a second
test where the template gate survives;
- SPARK-42813 (`waitAppCompletion=false`): `events.last === "remove gate"`.
I first ran your three changes against the previous tests: `else
resolvedDriverSpec.pod.pod`, an `else` branch that clears the template's gates,
and moving the gate block under `WAIT_FOR_APP_COMPLETION` all passed 17 of 17.
With the new tests the first fails 3 tests, the second 2 and the third 1
(SPARK-42813).
--
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]
