HwangDongJun commented on code in PR #58187:
URL: https://github.com/apache/spark/pull/58187#discussion_r4163569067
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala:
##########
@@ -227,6 +230,203 @@ class ClientSuite extends SparkFunSuite with
BeforeAndAfter {
assert(configMap.getData.get(SPARK_CONF_FILE_NAME).contains("conf2key=conf2value"))
}
+ test("SPARK-38079: driver's own config map is created before the driver pod,
" +
+ "to avoid a mount race") {
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher)
+ submissionClient.run()
+ // The first resourceList(...) call is always the pre-resources
application, which
+ // happens before the driver pod is created. The driver's own config map
must be
+ // included there so that it exists in Kubernetes before the pod that
mounts it.
+ val firstResourceListCall =
createdResourcesArgumentCaptor.getAllValues.get(0)
+ val configMaps = firstResourceListCall.filter(_.isInstanceOf[ConfigMap])
+ assert(configMaps.nonEmpty,
+ "the driver's own config map must be sent as a pre-resource, before the
driver pod " +
+ "is created, to avoid a \"configmap ... not found\" mount race
(SPARK-38079)")
+
+ // Safety check: the pre-resource owner-reference refresh (the second
resourceList()
+ // call) must still set an owner reference on the config map, same as
before this
+ // change, so that it is still garbage-collected along with the driver pod.
+ val secondResourceListCall =
createdResourcesArgumentCaptor.getAllValues.get(1)
+ val refreshedConfigMap = secondResourceListCall
+ .filter(_.isInstanceOf[ConfigMap]).map(_.asInstanceOf[ConfigMap]).head
+ val ownerReferences = refreshedConfigMap.getMetadata.getOwnerReferences
+ assert(ownerReferences.size() === 1)
+ assert(ownerReferences.get(0).getName === POD_NAME)
+ assert(ownerReferences.get(0).getUid === DRIVER_POD_UID)
+ }
+
+ // SPARK-38079: making the driver's own config map (and other
credential-bearing resources,
+ // e.g. Kerberos keytab/delegation-token secrets) a pre-resource above means
they briefly
+ // exist without an owner reference (see the comment on
cleanupOrphanedPreResources in
+ // KubernetesClientApplication.scala). run() registers a shutdown hook to
best-effort clean
+ // those up if this process is terminated abruptly in that window. Actually
triggering a JVM
+ // shutdown hook from a test is impractical, so these tests instead call the
(package-private)
+ // cleanup method directly, the same way the hook itself would.
+ private def newRecoveryClientMocks(): (KubernetesClient, RESOURCE_LIST,
PodResource) = {
+ val recoveryClient = mock[KubernetesClient]
+ val recoveryResourceList = mock[RESOURCE_LIST]
+ val recoveryPodOperations = mock[PODS]
+ val recoveryPodsWithNamespace = mock[PODS_WITH_NAMESPACE]
+ val recoveryNamedPod = mock[PodResource]
+ doReturn(recoveryResourceList)
+
.when(recoveryClient).resourceList(ArgumentMatchers.any[Array[HasMetadata]]():
_*)
+ when(recoveryClient.pods()).thenReturn(recoveryPodOperations)
+
when(recoveryPodOperations.inNamespace(kconf.namespace)).thenReturn(recoveryPodsWithNamespace)
+
when(recoveryPodsWithNamespace.withName(POD_NAME)).thenReturn(recoveryNamedPod)
+ (recoveryClient, recoveryResourceList, recoveryNamedPod)
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup deletes orphaned pre-resources and
the driver pod " +
+ "if this submission created it") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
true)
+
+ verify(recoveryResourceList).delete()
+ verify(recoveryNamedPod).delete()
+ // The recovery client is built fresh for this cleanup and must not leak.
+ verify(recoveryClient).close()
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup does not delete the driver pod if
this submission " +
+ "never created it") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ // podCreatedByUs = false: e.g. the pod creation API call itself never
succeeded (or a
+ // differently-submitted application happens to be using the same pod
name), so this
+ // cleanup must not delete a pod it did not create.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
false)
+
+ verify(recoveryResourceList).delete()
+ verify(recoveryNamedPod, never()).delete()
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup is a no-op if pre-resources were
never applied") {
+ var recoveryClientFactoryInvoked = false
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => {
+ recoveryClientFactoryInvoked = true
+ mock[KubernetesClient]
+ }))
+
+ // preResourcesApplied = false: the pre-resource serverSideApply() call
itself never
+ // succeeded, so there is nothing to have been left orphaned -- and in
particular, the
+ // existing catch block for that call (in run()) has already handled
cleanup of whatever
+ // partial state that failed call may have left behind.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = false, podCreatedByUs =
false)
+
+ assert(!recoveryClientFactoryInvoked,
+ "the recovery client must not be built at all when there is nothing to
clean up")
+ }
+
+ test("SPARK-38079: shutdown-hook cleanup swallows exceptions from the
recovery client " +
+ "(best-effort only)") {
+ val (recoveryClient, recoveryResourceList, recoveryNamedPod) =
newRecoveryClientMocks()
+ doThrow(new RuntimeException("simulated API server failure"))
+ .when(recoveryResourceList).delete()
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ recoveryClientFactoryOverride = Some(() => recoveryClient))
+
+ // Must not throw: this runs on the JVM shutdown-hook thread, where an
uncaught exception
+ // would only be printed to stderr and otherwise has no one left to
meaningfully handle it.
+ submissionClient.cleanupOrphanedPreResources(
+ PRE_RESOURCES, POD_NAME, preResourcesApplied = true, podCreatedByUs =
true)
+
+ // The pod delete is independent of the (failed) pre-resource delete
above, and must still
+ // be attempted.
+ verify(recoveryNamedPod).delete()
+ }
+
+ // SPARK-38079: the tests above all call cleanupOrphanedPreResources
directly, since actually
+ // triggering a JVM shutdown hook from a test is impractical. That leaves
the wiring in run()
+ // itself -- does it register a hook before applying pre-resources, and
remove that exact hook
+ // once done with them -- uncovered by those tests alone. These two tests
close that gap by
+ // injecting a fake ShutdownHookOps instead.
+ test("SPARK-38079: cleanup hook is registered before pre-resources are
applied and " +
+ "removed once run() completes successfully") {
+ var registeredHook: Option[() => Unit] = None
+ var removedRef: Option[Any] = None
+ val fakeOps = ShutdownHookOps(
+ addHook = { hook =>
+ registeredHook = Some(hook)
+ "fake-hook-ref"
+ },
+ removeHook = { ref =>
+ removedRef = Some(ref)
+ true
+ })
+ val submissionClient = new Client(
+ kconf,
+ driverBuilder,
+ kubernetesClient,
+ loggingPodStatusWatcher,
+ shutdownHookOpsOverride = Some(fakeOps))
+
+ submissionClient.run()
+
+ assert(registeredHook.isDefined,
Review Comment:
All four of these are addressed now:
- The `.edit()` call's lambda is actually invoked in the mock setup --
`namedPods.edit(...)` applies the captured function to a real pod and returns
the result, and the test asserts on that returned pod, not just that `edit()`
was called with *some* function.
- A dedicated test (pod creation failure) asserts `resourceList(...)` is
never called when `create()` fails.
- `InOrder` now pins down `create()` → `resourceList(...)` → `edit(...)`
happening in that order, not just independently.
- The vararg matchers are `ArgumentMatchers.any[Array[HasMetadata]]()`
(Scala) / `ArgumentMatchers.any[HasMetadata]()` for the single-resource
`delete()` path, which match the whole array/argument rather than one vararg
element.
Also -- while addressing this, I noticed this same suite still had a couple
of the diff-describing-comment and direct-comment-ID patterns you flagged in
[the comment-diet
reply](https://github.com/apache/spark/pull/58187#discussion_r4129910586) (a
prior cleanup commit covered the other three files but missed this one). Fixed
those in a follow-up commit.
--
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]
