This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7511-29d7cd6cff43542d11effd970ecfc9f1ba78ce11 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 8aec22903261ddc69d43fe60def80e9b08547aa1 Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Aug 10 10:30:02 2026 -0700 test(computing-unit): cover the pod lookup, creation and deletion paths (#7511) ### What changes were proposed in this PR? The spec covered the pure transforms and the namespace-wide wrappers and stopped. The single-pod half of the class — `getPodByName`, `podExists`, `getPodLimits`, `createPod`, `deletePod` and the pod URI — was untested, which is 42 of the file's 62 lines. None of it needs a cluster: the fabric8 client is already a constructor parameter, so the existing Mockito fixture extends to the rest of the fluent chain, with the pod that `createPod` builds captured and inspected rather than sent anywhere. Adds 8 tests. The three that matter: - **the guard that refuses to overwrite a live pod** — creating over a running unit would detach it from its owner; - **the `Option(...)` wrapper around the by-name lookup** — fabric8 returns `null` for an absent pod rather than throwing, so the wrapper is all that stands between a caller and an NPE; - **the shared-memory volume appearing only when a size is requested** — `/dev/shm` defaults to 64 Mi, too small for the Python workers, and the volume must not appear when unrequested. Also covered: the pod URI's service and namespace segments, the first container's resource limits and the empty-map fallback, env values reaching the container as strings, and `deletePod` targeting the cuid's own pod. **Verified by mutation**, all reverted (production diff empty): | Mutation | Result | |---|---| | pod name loses the cuid suffix | red | | URI drops the namespace segment | red | | `getPodByName` no longer null-guards | red | | `createPod` overwrites an existing pod | red | | cpu limit written from the memory value | red | | pod hostname is not the pod name | red | | `deletePod` targets a fixed cuid | red | No production file is touched. ### Any related issues, documentation, discussions? Closes #7510 ### How was this PR tested? ``` sbt "ComputingUnitManagingService/testOnly org.apache.texera.service.util.KubernetesClientSpec" ``` ``` [info] Total number of tests run: 16 [info] Tests: succeeded 16, failed 0, canceled 0, ignored 0, pending 0 ``` 8 new on top of the existing 8. `Test/scalafmtCheck` and `Test/scalafix --check` both pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) Co-authored-by: Meng Wang <[email protected]> --- .../texera/service/util/KubernetesClientSpec.scala | 163 ++++++++++++++++++++- 1 file changed, 160 insertions(+), 3 deletions(-) diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala index f0d96347a9..f8e8353f7e 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala @@ -26,17 +26,29 @@ import io.fabric8.kubernetes.api.model.metrics.v1beta1.{ PodMetricsList, PodMetricsListBuilder } -import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity} +import io.fabric8.kubernetes.api.model.{ + ContainerBuilder, + Pod, + PodBuilder, + PodList, + PodListBuilder, + Quantity, + ResourceRequirementsBuilder +} import io.fabric8.kubernetes.client.dsl.{ MetricAPIGroupDSL, MixedOperation, + NamespaceableResource, NonNamespaceOperation, PodMetricOperation, - PodResource + PodResource, + Resource } import io.fabric8.kubernetes.client.{KubernetesClient => Fabric8Client} import org.apache.texera.common.config.KubernetesConfig -import org.mockito.Mockito.{mock, when} +import org.mockito.ArgumentCaptor +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.{mock, times, verify, when} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -160,4 +172,149 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers { k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") k8s.getPodMetrics(999) shouldBe empty } + // ── single-pod lookups, creation and deletion ── + // These reach the rest of the fluent chain: withName(...).get() for the lookups, + // resource(pod).inNamespace(...).create() for creation, and .delete() for removal. Everything + // is driven through the constructor seam, so no cluster is involved. + + /** Extends the namespace stub with the by-name pod operations `withName(...)` returns. */ + private def clientWithNamedPod(podName: String, found: Pod): (Fabric8Client, PodResource) = { + val client = stubbedClient(Seq.empty, Seq.empty) + val podsInNamespace = client.pods().inNamespace(namespace) + val podResource = mock(classOf[PodResource]) + when(podsInNamespace.withName(podName)).thenReturn(podResource) + when(podResource.get()).thenReturn(found) + (client, podResource) + } + + /** A pod carrying one container whose resource limits are set. */ + private def podWithLimits(cuid: Int, limits: Map[String, String]): Pod = + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(cuid)) + .endMetadata() + .withNewSpec() + .addToContainers( + new ContainerBuilder() + .withName("main") + .withResources( + new ResourceRequirementsBuilder() + .withLimits(limits.map { case (k, v) => k -> new Quantity(v) }.asJava) + .build() + ) + .build() + ) + .endSpec() + .build() + + "generatePodURI" should "address the pod through its headless service inside the namespace" in { + // The URI is how a computing unit is reached once it is up, so every segment matters: a pod + // name alone, or the wrong namespace, resolves to nothing. + val uri = KubernetesClient.generatePodURI(7) + uri should startWith(KubernetesClient.generatePodName(7) + ".") + uri should include(s".${KubernetesConfig.computeUnitServiceName}.$namespace.svc.cluster.local:") + uri should endWith(s":${KubernetesConfig.computeUnitPortNumber}") + } + + "getPodByName" should "wrap a found pod and report a missing one as None" in { + // fabric8 returns null rather than throwing for an absent pod, so the Option() wrapper is the + // only thing standing between a caller and an NPE. + val name = KubernetesClient.generatePodName(1) + val (found, _) = clientWithNamedPod(name, pod(1, "Running")) + val (absent, _) = clientWithNamedPod(name, null) + + new KubernetesClient(found).getPodByName(name).map(_.getMetadata.getName) shouldBe Some(name) + new KubernetesClient(absent).getPodByName(name) shouldBe None + } + + "podExists" should "follow the by-name lookup in both directions" in { + val name = KubernetesClient.generatePodName(2) + new KubernetesClient(clientWithNamedPod(name, pod(2, "Running"))._1).podExists(2) shouldBe true + new KubernetesClient(clientWithNamedPod(name, null)._1).podExists(2) shouldBe false + } + + "getPodLimits" should "read the first container's limits and fall back to an empty map" in { + val name = KubernetesClient.generatePodName(3) + val withLimits = + clientWithNamedPod(name, podWithLimits(3, Map("cpu" -> "2", "memory" -> "4Gi")))._1 + val missing = clientWithNamedPod(name, null)._1 + + new KubernetesClient(withLimits).getPodLimits(3) shouldBe Map("cpu" -> "2", "memory" -> "4Gi") + new KubernetesClient(missing).getPodLimits(3) shouldBe empty + } + + "createPod" should "refuse to overwrite a pod that already exists" in { + // Creating over a live unit would silently detach the running one from its owner. + val name = KubernetesClient.generatePodName(4) + val k8s = new KubernetesClient(clientWithNamedPod(name, pod(4, "Running"))._1) + + val thrown = intercept[Exception] { + k8s.createPod(4, "1", "2Gi", "0", Map.empty) + } + thrown.getMessage should include("already exists") + } + + it should "build the pod from the requested limits and env, and create it in the namespace" in { + val name = KubernetesClient.generatePodName(5) + val (client, _) = clientWithNamedPod(name, null) + val namespaceable = mock(classOf[NamespaceableResource[Pod]]) + val resource = mock(classOf[Resource[Pod]]) + val captor = ArgumentCaptor.forClass(classOf[Pod]) + when(client.resource(any(classOf[Pod]))).thenReturn(namespaceable) + when(namespaceable.inNamespace(namespace)).thenReturn(resource) + // create()'s return value is not asserted; the pod is inspected through the captor below. + when(resource.create()).thenReturn(null) + + new KubernetesClient(client).createPod(5, "2", "4Gi", "1", Map("UID" -> 9, "MODE" -> "batch")) + + verify(client).resource(captor.capture()) + val built = captor.getValue + built.getSpec.getHostname shouldBe name + built.getSpec.getSubdomain shouldBe KubernetesConfig.computeUnitServiceName + val container = built.getSpec.getContainers.asScala.head + val limits = container.getResources.getLimits.asScala.map { case (k, v) => k -> v.toString } + limits("cpu") shouldBe "2" + limits("memory") shouldBe "4Gi" + // Env values arrive as Any and reach the container as strings. + container.getEnv.asScala.map(e => e.getName -> e.getValue).toMap shouldBe + Map("UID" -> "9", "MODE" -> "batch") + } + + it should "mount a shared-memory volume only when a size is asked for" in { + // /dev/shm defaults to 64Mi in Kubernetes, which is too small for the Python workers, so the + // volume is the fix — but it must not appear when no size was requested. + def build(shm: Option[String]): Pod = { + val name = KubernetesClient.generatePodName(6) + val (client, _) = clientWithNamedPod(name, null) + val namespaceable = mock(classOf[NamespaceableResource[Pod]]) + val resource = mock(classOf[Resource[Pod]]) + val captor = ArgumentCaptor.forClass(classOf[Pod]) + when(client.resource(any(classOf[Pod]))).thenReturn(namespaceable) + when(namespaceable.inNamespace(namespace)).thenReturn(resource) + // create()'s return value is not asserted; the pod is inspected through the captor below. + when(resource.create()).thenReturn(null) + new KubernetesClient(client).createPod(6, "1", "2Gi", "0", Map.empty, shm) + verify(client).resource(captor.capture()) + captor.getValue + } + + val withShm = build(Some("1Gi")) + withShm.getSpec.getVolumes.asScala.map(_.getName) should contain("dshm") + withShm.getSpec.getVolumes.asScala + .find(_.getName == "dshm") + .flatMap(v => Option(v.getEmptyDir)) + .map(_.getSizeLimit.toString) shouldBe Some("1Gi") + + Option(build(None).getSpec.getVolumes).map(_.asScala.map(_.getName)).getOrElse(Nil) should + not contain "dshm" + } + + "deletePod" should "delete the pod for the cuid inside the namespace" in { + val name = KubernetesClient.generatePodName(8) + val (client, podResource) = clientWithNamedPod(name, pod(8, "Running")) + + new KubernetesClient(client).deletePod(8) + + verify(podResource, times(1)).delete() + } }
