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-6853-2300d04717131519838cfec60e27b105f0178ab0 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 46b352f60edc3dde9adfe119626a0a7c0b93abeb Author: Kunwoo (Chris) <[email protected]> AuthorDate: Wed Jul 29 13:46:58 2026 -0400 refactor(computing-unit-managing-service): share CU listing helpers and batch k8s calls (#6853) ### What changes were proposed in this PR? Refactor the per-user computing-unit listing so its status/metrics logic lives in one reusable place and hits Kubernetes in bulk. No API response changes; this is groundwork for the admin listing endpoint (stacked follow-up PR). - Centralize listing logic in `ComputingUnitHelpers` (`resolveOwnerInfo`, `partitionLiveUnits`, `reconcileVanishedKubernetesUnits`, `buildDashboardUnit`, plus the `isKubernetes` / `podPhasesFor` / `podMetricsFor` guards), and remove the duplicated private status/metrics helpers from `ComputingUnitManagingResource`. - Add bulk `KubernetesClient.getAllPodPhases` / `getAllPodMetrics` (one namespace-wide `list()` / `top()` each) and a shared `containerUsage` helper; `listComputingUnits` now resolves pod status/metrics in bulk instead of probing pod existence twice per unit and re-fetching `top()` per unit. - Per-user reconciliation now runs over cuid-deduplicated units, reducing redundant DB updates. - Add a package-private `KubernetesClient` test seam (`setClientForTesting`) so the fabric8 client can be stubbed in tests; unused in production. ### Any related issues, documentation, discussions? Part of #6476 (Admin Computing Units Dashboard); prepares the shared listing helpers reused by the admin endpoint in the stacked follow-up. Originates from discussion #6322. ### How was this PR tested? Added `ComputingUnitManagingResourceSpec` and extended `ComputingUnitHelpersSpec` / `KubernetesClientSpec` (embedded Postgres via `MockTexeraDB`; stubbed fabric8 client for the bulk k8s paths). Ran `sbt 'ComputingUnitManagingService/testOnly org.apache.texera.service.util.ComputingUnitHelpersSpec org.apache.texera.service.util.KubernetesClientSpec org.apache.texera.service.resource.ComputingUnitManagingResourceSpec'` (34 tests, all green) and `sbt 'ComputingUnitManagingService/scalafmtCheck; ComputingUnitManagingService/Test/scalafmtCheck'` (clean). ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code, Claude Opus 4.8 --------- Signed-off-by: Kunwoo (Chris) <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- build.sbt | 5 + .../resource/ComputingUnitManagingResource.scala | 140 ++++------- .../texera/service/util/ComputingUnitHelpers.scala | 181 ++++++++++++- .../texera/service/util/KubernetesClient.scala | 66 +++-- .../ComputingUnitManagingResourceSpec.scala | 104 ++++++++ .../service/util/ComputingUnitHelpersSpec.scala | 280 ++++++++++++++++++++- .../texera/service/util/KubernetesClientSpec.scala | 129 ++++++++++ 7 files changed, 783 insertions(+), 122 deletions(-) diff --git a/build.sbt b/build.sbt index 720fc28b80..f8e6f42622 100644 --- a/build.sbt +++ b/build.sbt @@ -164,7 +164,12 @@ lazy val ComputingUnitManagingService = (project in file("computing-unit-managin .configs(Test) .dependsOn(DAO % "test->test") // reuse MockTexeraDB embedded Postgres in tests .settings(commonModuleSettings) + .configs(Test) + .dependsOn(DAO % "test->test", Auth % "test->test") // reuse MockTexeraDB embedded Postgres in tests .settings( + // MockTexeraDB swaps a JVM-wide singleton (SqlServer's embedded Postgres), + // so run suites serially to avoid cross-suite races. + Test / parallelExecution := false, dependencyOverrides ++= Seq( // override it as io.dropwizard 4 require 2.16.1 or higher "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion, diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala index aa02f73387..df67fb3bfe 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala @@ -50,8 +50,8 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{ } import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import org.apache.texera.service.resource.ComputingUnitManagingResource._ -import org.apache.texera.service.resource.ComputingUnitState._ import org.apache.texera.service.util.{ + ComputingUnitHelpers, ComputingUnitManagingServiceException, InsufficientComputingUnitQuota, KubernetesClient @@ -197,41 +197,6 @@ class ComputingUnitManagingResource { } } - private def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = { - unit.getType match { - // ── Local CUs are always "running" ────────────────────────────── - case WorkflowComputingUnitTypeEnum.local => - Running - - // ── Kubernetes CUs – only explicit "Running" counts as running ─ - case WorkflowComputingUnitTypeEnum.kubernetes => - val phaseOpt = KubernetesClient - .getPodByName(KubernetesClient.generatePodName(unit.getCuid)) - .map(_.getStatus.getPhase) - - if (phaseOpt.contains("Running")) Running else Pending - - // ── Any other (unknown) type is treated as pending ────────────── - case _ => - Pending - } - } - - private def getComputingUnitMetrics(unit: WorkflowComputingUnit): WorkflowComputingUnitMetrics = { - unit.getType match { - case WorkflowComputingUnitTypeEnum.local => - WorkflowComputingUnitMetrics("NaN", "NaN") - case WorkflowComputingUnitTypeEnum.kubernetes => - val metrics = KubernetesClient.getPodMetrics(unit.getCuid) - WorkflowComputingUnitMetrics( - metrics.getOrElse("cpu", ""), - metrics.getOrElse("memory", "") - ) - case _ => - WorkflowComputingUnitMetrics("NaN", "NaN") - } - } - private def getComputingUnitResourceLimit( unit: WorkflowComputingUnit ): WorkflowComputingUnitResourceLimit = { @@ -476,8 +441,8 @@ class ComputingUnitManagingResource { DashboardWorkflowComputingUnit( insertedUnit, - getComputingUnitStatus(insertedUnit).toString, - getComputingUnitMetrics(insertedUnit), + ComputingUnitHelpers.getComputingUnitStatus(insertedUnit).toString, + ComputingUnitHelpers.getComputingUnitMetrics(insertedUnit), isOwner = true, accessPrivilege = PrivilegeEnum.WRITE, ownerGoogleAvatar, @@ -528,63 +493,46 @@ class ComputingUnitManagingResource { (List.empty[WorkflowComputingUnit], Map.empty[Integer, PrivilegeEnum]) } - val allUnits = ownedUnits ++ sharedUnits - val ownerUids: List[Integer] = allUnits.map(_.getUid).distinct val userDao = new UserDao(ctx.configuration()) - val ownerInfoMap: Map[Integer, (String, String)] = - userDao - .fetchByUid(ownerUids: _*) - .asScala - .map { u => - val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull - val name = Option(u.getName).filter(_.nonEmpty).orNull - u.getUid -> (avatar, name) - } - .toMap - - // If a Kubernetes pod has already disappeared (e.g., manually deleted or TTL - // GC-ed by the cluster), we treat the corresponding computing unit as - // terminated from the system's point of view. Here we eagerly update its - // terminateTime in the database **before** we build the response list so - // that subsequent API calls will no longer return this unit. - allUnits.foreach { unit => - if ( - unit.getType == WorkflowComputingUnitTypeEnum.kubernetes && - !KubernetesClient.podExists(unit.getCuid) - ) { - unit.setTerminateTime(new Timestamp(System.currentTimeMillis())) - computingUnitDao.update(unit) - } - } - // For shared units, we need to check the access privilege which are saved in different table - // to streamline the process, we combine owned units with default WRITE privilege and use sharedUnitInfo - // to get the privilege for shared units. - (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++ sharedUnits.map(u => - (u, sharedUnitInfo(u.getCuid)) - )) - .distinctBy { case (unit, _) => unit.getCuid } - .filter { case (unit, _) => unit.getTerminateTime == null } - .filter { - case (unit, _) => - unit.getType match { - case WorkflowComputingUnitTypeEnum.kubernetes => - KubernetesClient.podExists(unit.getCuid) - case _ => true - } - } - .map { - case (unit, privilege) => - DashboardWorkflowComputingUnit( - computingUnit = unit, - isOwner = unit.getUid.equals(uid), - accessPrivilege = privilege, - status = getComputingUnitStatus(unit).toString, - metrics = getComputingUnitMetrics(unit), - ownerGoogleAvatar = ownerInfoMap.getOrElse(unit.getUid, (null, null))._1, - ownerName = ownerInfoMap.getOrElse(unit.getUid, (null, null))._2 - ) - } + // Pair each unit with the caller's privilege (owned default to WRITE), one row per cuid, so + // a unit that is both owned and shared is reconciled/rendered exactly once. + val unitsWithPrivilege = + (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++ + sharedUnits.map(u => (u, sharedUnitInfo(u.getCuid)))) + .distinctBy { case (unit, _) => unit.getCuid } + .filter { case (unit, _) => unit.getTerminateTime == null } + val privilegeByCuid = unitsWithPrivilege.map { + case (unit, privilege) => unit.getCuid -> privilege + }.toMap + val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit } + + // Pod phases decide which Kubernetes units are still alive. + val podPhases = ComputingUnitHelpers.podPhasesFor(candidateUnits) + + val liveUnits = + ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + computingUnitDao, + candidateUnits, + podPhases + ) + + // Metrics only for survivors, so fetch after reconciliation. + val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) + + val ownerInfoMap = + ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct) + + liveUnits.map { unit => + ComputingUnitHelpers.buildDashboardUnit( + unit, + isOwner = unit.getUid.equals(uid), + accessPrivilege = privilegeByCuid(unit.getCuid), + ownerInfo = ownerInfoMap, + podPhases = podPhases, + podMetrics = podMetrics + ) + } } } @@ -613,8 +561,8 @@ class ComputingUnitManagingResource { DashboardWorkflowComputingUnit( computingUnit = unit, - status = getComputingUnitStatus(unit).toString, - metrics = getComputingUnitMetrics(unit), + status = ComputingUnitHelpers.getComputingUnitStatus(unit).toString, + metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit), isOwner = unit.getUid.equals(user.getUid), accessPrivilege = { val cuAccessDao = new ComputingUnitUserAccessDao(context.configuration()) @@ -748,7 +696,7 @@ class ComputingUnitManagingResource { throw new BadRequestException("User has no access to the computing unit") } val computingUnit = getComputingUnitByCuid(context, cuid.toInt) - getComputingUnitMetrics(computingUnit) + ComputingUnitHelpers.getComputingUnitMetrics(computingUnit) } @GET diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala index 8ae31bba7f..ea7a65ef51 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala @@ -19,12 +19,56 @@ package org.apache.texera.service.util import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit -import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics +import org.apache.texera.service.resource.ComputingUnitManagingResource.{ + DashboardWorkflowComputingUnit, + WorkflowComputingUnitMetrics +} import org.apache.texera.service.resource.ComputingUnitState.{ComputingUnitState, Pending, Running} +import org.jooq.EnumType + +import java.sql.Timestamp +import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava} object ComputingUnitHelpers { - def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = { + + /** + * Owner (avatar, name) keyed by uid, resolved in one query. Blank values collapse to `null`; + * empty `uids` returns empty without querying. + */ + def resolveOwnerInfo( + userDao: UserDao, + uids: Seq[Integer] + ): Map[Integer, (String, String)] = { + if (uids.isEmpty) Map.empty + else + userDao + .fetchByUid(uids: _*) + .asScala + .map { u => + val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull + val name = Option(u.getName).filter(_.nonEmpty).orNull + u.getUid -> (avatar, name) + } + .toMap + } + + def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = + singleUnitStatus(unit, KubernetesClient) + + /** + * Single-unit status via a per-unit pod lookup (a targeted GET, cheaper than listing the whole + * namespace). The client is a by-name parameter — not the global singleton — so the kubernetes + * branch is unit-testable with a stub and the local/unknown branches never force the singleton; + * the public overload binds the production [[KubernetesClient]]. (Metrics has no analogous seam: + * its per-unit lookup already fans out to the whole namespace and the bulk (unit, podMetrics) + * overload already covers the cpu/memory resolution, so nothing there is worth pinning.) + */ + private[util] def singleUnitStatus( + unit: WorkflowComputingUnit, + k8s: => KubernetesClient + ): ComputingUnitState = { unit.getType match { // Local CUs are always “running” case WorkflowComputingUnitTypeEnum.local => @@ -32,9 +76,12 @@ object ComputingUnitHelpers { // Kubernetes CUs – only explicit “Running” counts as running case WorkflowComputingUnitTypeEnum.kubernetes => - val phaseOpt = KubernetesClient - .getPodByName(KubernetesClient.generatePodName(unit.getCuid)) - .map(_.getStatus.getPhase) + // Guard the pod status the same way the bulk getAllPodPhases does: a pod with no + // status yet has a null getStatus, so map through Option to avoid an NPE. + val client = k8s + val phaseOpt = client + .getPodByName(client.generatePodName(unit.getCuid)) + .flatMap(pod => Option(pod.getStatus).map(_.getPhase)) if (phaseOpt.contains("Running")) Running else Pending @@ -58,4 +105,128 @@ object ComputingUnitHelpers { WorkflowComputingUnitMetrics("NaN", "NaN") } } + + /** + * Resolves status from a pre-fetched pod-phase map instead of a per-unit cluster call, so a + * listing costs O(1) round trips rather than one per unit. + */ + def getComputingUnitStatus( + unit: WorkflowComputingUnit, + podPhases: Map[String, String] + ): ComputingUnitState = { + unit.getType match { + case WorkflowComputingUnitTypeEnum.local => + Running + case WorkflowComputingUnitTypeEnum.kubernetes => + // A missing entry or null phase both count as not-Running. + if (podPhases.get(KubernetesClient.generatePodName(unit.getCuid)).contains("Running")) + Running + else Pending + case _ => + Pending + } + } + + /** Resolves metrics from a pre-fetched pod-metrics map instead of a per-unit cluster call. */ + def getComputingUnitMetrics( + unit: WorkflowComputingUnit, + podMetrics: Map[String, Map[String, String]] + ): WorkflowComputingUnitMetrics = { + unit.getType match { + case WorkflowComputingUnitTypeEnum.local => + WorkflowComputingUnitMetrics("NaN", "NaN") + case WorkflowComputingUnitTypeEnum.kubernetes => + val metrics = podMetrics + .getOrElse(KubernetesClient.generatePodName(unit.getCuid), Map.empty[String, String]) + WorkflowComputingUnitMetrics( + metrics.getOrElse("cpu", ""), + metrics.getOrElse("memory", "") + ) + case _ => + WorkflowComputingUnitMetrics("NaN", "NaN") + } + } + + private def isKubernetes(unit: WorkflowComputingUnit): Boolean = + unit.getType match { + case WorkflowComputingUnitTypeEnum.kubernetes => true + case _ => false + } + + // Pod phases/metrics for the namespace; skipped (empty) when no Kubernetes unit is present, so a + // cluster-free listing issues no round trip. Same seam as singleUnitStatus: the public overload + // binds the production singleton, passing it by-name to the private[util] overload, which forces + // it only inside the guard's true branch — so the empty path never touches the client, and tests + // drive the private overload with a stub. + def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] = + podPhasesFor(units, KubernetesClient) + + private[util] def podPhasesFor( + units: Seq[WorkflowComputingUnit], + k8s: => KubernetesClient + ): Map[String, String] = + if (units.exists(isKubernetes)) k8s.getAllPodPhases else Map.empty + + def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] = + podMetricsFor(units, KubernetesClient) + + private[util] def podMetricsFor( + units: Seq[WorkflowComputingUnit], + k8s: => KubernetesClient + ): Map[String, Map[String, String]] = + if (units.exists(isKubernetes)) k8s.getAllPodMetrics else Map.empty + + /** A Kubernetes unit whose pod is absent from `podPhases` (deleted or TTL GC-ed). */ + private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean = + isKubernetes(unit) && !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid)) + + /** Partition into `(live, vanished)` by `podPhases`. Pure (no I/O), so it is unit-testable. */ + def partitionLiveUnits( + units: List[WorkflowComputingUnit], + podPhases: Map[String, String] + ): (List[WorkflowComputingUnit], List[WorkflowComputingUnit]) = + units.partition(unit => !isVanished(unit, podPhases)) + + /** + * Stamp `terminateTime` on vanished Kubernetes units (one batched update) and return the live + * ones. Shared by both listing endpoints so they agree on when a unit is terminated. + */ + def reconcileVanishedKubernetesUnits( + dao: WorkflowComputingUnitDao, + units: List[WorkflowComputingUnit], + podPhases: Map[String, String] + ): List[WorkflowComputingUnit] = { + val partitioned = partitionLiveUnits(units, podPhases) + val vanished = partitioned._2 + if (vanished.nonEmpty) { + val now = new Timestamp(System.currentTimeMillis()) + vanished.foreach(_.setTerminateTime(now)) + dao.update(vanished.asJava) + } + partitioned._1 + } + + /** + * Build one dashboard row; status/metrics come from the pre-fetched maps (no per-unit K8s + * call). Shared by both listing endpoints so row shape and owner-info fallback stay identical. + */ + def buildDashboardUnit( + unit: WorkflowComputingUnit, + isOwner: Boolean, + accessPrivilege: EnumType, + ownerInfo: Map[Integer, (String, String)], + podPhases: Map[String, String], + podMetrics: Map[String, Map[String, String]] + ): DashboardWorkflowComputingUnit = { + val owner = ownerInfo.getOrElse(unit.getUid, (null, null)) + DashboardWorkflowComputingUnit( + computingUnit = unit, + status = getComputingUnitStatus(unit, podPhases).toString, + metrics = getComputingUnitMetrics(unit, podMetrics), + isOwner = isOwner, + accessPrivilege = accessPrivilege, + ownerGoogleAvatar = owner._1, + ownerName = owner._2 + ) + } } diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala index 4f1d391cb3..6f97a5cf6b 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala @@ -20,17 +20,20 @@ package org.apache.texera.service.util import io.fabric8.kubernetes.api.model._ -import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetricsList +import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetrics import io.fabric8.kubernetes.client.KubernetesClientBuilder import org.apache.texera.common.config.KubernetesConfig import scala.jdk.CollectionConverters._ -object KubernetesClient { +/** + * Thin wrapper over the fabric8 Kubernetes client. The production singleton is the companion + * object below, bound to a real in-cluster client. The fabric8 client is a constructor + * parameter (not a mutable global) so tests can construct an instance backed by a stubbed + * client and exercise the passthrough wrappers without a live cluster. + */ +class KubernetesClient(client: io.fabric8.kubernetes.client.KubernetesClient) { - // Initialize the Kubernetes client - private val client: io.fabric8.kubernetes.client.KubernetesClient = - new KubernetesClientBuilder().build() private val namespace: String = KubernetesConfig.computeUnitPoolNamespace private val podNamePrefix = "computing-unit" @@ -48,19 +51,51 @@ object KubernetesClient { Option(client.pods().inNamespace(namespace).withName(podName).get()) } + /** + * Phase of every pod in the namespace, keyed by pod name, in one call — so a bulk listing + * avoids a per-unit lookup. Unfiltered so callers can test a unit's presence by its pod-name + * key; a pod with no status yet maps to a `null` phase but still appears. + */ + def getAllPodPhases: Map[String, String] = + phasesByPodName(client.pods().inNamespace(namespace).list().getItems.asScala) + + /** Pure fabric8 -> map transform: a pod with no status yet maps to a `null` phase. */ + private[util] def phasesByPodName(pods: Iterable[Pod]): Map[String, String] = + pods + .map(pod => pod.getMetadata.getName -> Option(pod.getStatus).map(_.getPhase).orNull) + .toMap + + // Flatten a pod's per-container resource usage into a single metric -> value map. + private def containerUsage(podMetrics: PodMetrics): Map[String, String] = + podMetrics.getContainers.asScala.flatMap { container => + container.getUsage.asScala.map { + case (metric, value) => metric -> value.toString + } + }.toMap + + /** Pure fabric8 -> map transform over the raw per-pod metrics items. */ + private[util] def metricsByPodName( + items: Iterable[PodMetrics] + ): Map[String, Map[String, String]] = + items.map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics)).toMap + + // One namespace-wide metrics call, returning the raw per-pod items. + private def fetchPodMetricsItems(): Iterable[PodMetrics] = + client.top().pods().metrics(namespace).getItems.asScala + + /** + * CPU/memory of every pod in the namespace, keyed by pod name, in one call — the bulk + * counterpart to the single-unit lookup. + */ + def getAllPodMetrics: Map[String, Map[String, String]] = + metricsByPodName(fetchPodMetricsItems()) + def getPodMetrics(cuid: Int): Map[String, String] = { - val podMetricsList: PodMetricsList = client.top().pods().metrics(namespace) val targetPodName = generatePodName(cuid) - - podMetricsList.getItems.asScala + fetchPodMetricsItems() .collectFirst { case podMetrics if podMetrics.getMetadata.getName == targetPodName => - podMetrics.getContainers.asScala.flatMap { container => - container.getUsage.asScala.map { - case (metric, value) => - metric -> value.toString - } - }.toMap + containerUsage(podMetrics) } .getOrElse(Map.empty[String, String]) } @@ -182,3 +217,6 @@ object KubernetesClient { client.pods().inNamespace(namespace).withName(generatePodName(cuid)).delete() } } + +/** Production singleton bound to a real in-cluster fabric8 client. */ +object KubernetesClient extends KubernetesClient(new KubernetesClientBuilder().build()) diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala new file mode 100644 index 0000000000..54e495473d --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + UserRoleEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit} +import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +// Drives the per-user computing-unit endpoints against the embedded database using +// local units (so no Kubernetes calls are made). +class ComputingUnitManagingResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + private val uid = 800 + private lazy val user: SessionUser = { + val u = new User() + u.setUid(uid) + u.setName("owner") + u.setEmail("[email protected]") + u.setRole(UserRoleEnum.REGULAR) + u.setPassword("password") + u.setGoogleAvatar("owner-avatar") + new SessionUser(u) + } + + private def localUnit(cuid: Int, name: String): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit() + unit.setCuid(cuid) + unit.setUid(uid) + unit.setName(name) + unit.setType(WorkflowComputingUnitTypeEnum.local) + unit + } + + override protected def beforeAll(): Unit = { + super.beforeAll() + initializeDBAndReplaceDSLContext() + new UserDao(getDSLContext.configuration()).insert(user.getUser) + val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + unitDao.insert(localUnit(800, "cu-a")) + unitDao.insert(localUnit(801, "cu-b")) + } + + override protected def afterAll(): Unit = + try shutdownDB() + finally super.afterAll() + + private val resource = new ComputingUnitManagingResource + + "getComputingUnitInfo" should "return the owner's local unit with WRITE access and Running status" in { + val info = resource.getComputingUnitInfo(800, user) + + info.computingUnit.getCuid shouldBe 800 + info.status shouldBe "Running" + info.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") + info.isOwner shouldBe true + info.accessPrivilege shouldBe PrivilegeEnum.WRITE + info.ownerName shouldBe "owner" + } + + "getComputingUnitMetricsEndpoint" should "return NaN metrics for an owned local unit" in { + resource.getComputingUnitMetricsEndpoint("800", user) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + "listComputingUnits" should "return the caller's owned, non-terminated units" in { + val result = resource.listComputingUnits(user) + + result.map(_.computingUnit.getCuid.intValue()) should contain theSameElementsAs Seq(800, 801) + all(result.map(_.isOwner)) shouldBe true + all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE + all(result.map(_.status)) shouldBe "Running" + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala index 39a11ea2a5..cf54089a13 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala @@ -19,25 +19,78 @@ package org.apache.texera.service.util -import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum -import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import io.fabric8.kubernetes.api.model.{Pod, PodBuilder} +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + UserRoleEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit} import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics import org.apache.texera.service.resource.ComputingUnitState.{Pending, Running} +import org.mockito.Mockito.{mock, when} +import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { +class ComputingUnitHelpersSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + override protected def beforeAll(): Unit = { + super.beforeAll() + initializeDBAndReplaceDSLContext() + } + + override protected def afterAll(): Unit = + try shutdownDB() + finally super.afterAll() + + private lazy val userDao = new UserDao(getDSLContext.configuration()) + private lazy val computingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) - private def localUnit(): WorkflowComputingUnit = { + private def makeUser(uid: Int, name: String, email: String, avatar: String): User = { + val u = new User() + u.setUid(uid) + u.setName(name) + u.setEmail(email) + u.setRole(UserRoleEnum.REGULAR) + u.setPassword("password") + u.setGoogleAvatar(avatar) + u + } + + private def makeUnit( + cuid: Int, + uid: Int, + tpe: WorkflowComputingUnitTypeEnum + ): WorkflowComputingUnit = { val unit = new WorkflowComputingUnit() - unit.setType(WorkflowComputingUnitTypeEnum.local) + unit.setCuid(cuid) + unit.setUid(uid) + unit.setType(tpe) unit } - // WorkflowComputingUnitTypeEnum only defines `local` and `kubernetes`, so an - // untyped unit (getType == null) is what exercises the pure "unknown" branch. + private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit = + makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.local) + + private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit = + makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.kubernetes) + + // A null-type unit (the enum has only local/kubernetes) exercises the "unknown" branch. private def untypedUnit(): WorkflowComputingUnit = new WorkflowComputingUnit() + private def podWithPhase(phase: String): Pod = + new PodBuilder().withNewStatus().withPhase(phase).endStatus().build() + + // A pod whose status has not been populated yet (getStatus == null). + private def statuslessPod(): Pod = new PodBuilder().build() + "getComputingUnitStatus" should "return Running for a local unit" in { ComputingUnitHelpers.getComputingUnitStatus(localUnit()) shouldBe Running } @@ -46,6 +99,29 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { ComputingUnitHelpers.getComputingUnitStatus(untypedUnit()) shouldBe Pending } + // The kubernetes branch does a per-unit pod lookup, so singleUnitStatus is driven through a + // stubbed client (the public getComputingUnitStatus binds the production singleton). + "singleUnitStatus" should "return Running for a kubernetes unit whose pod phase is Running" in { + val k8s = mock(classOf[KubernetesClient]) + when(k8s.generatePodName(40)).thenReturn("computing-unit-40") + when(k8s.getPodByName("computing-unit-40")).thenReturn(Some(podWithPhase("Running"))) + ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(40), k8s) shouldBe Running + } + + it should "return Pending for a kubernetes unit whose pod has no status yet" in { + val k8s = mock(classOf[KubernetesClient]) + when(k8s.generatePodName(41)).thenReturn("computing-unit-41") + when(k8s.getPodByName("computing-unit-41")).thenReturn(Some(statuslessPod())) + ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(41), k8s) shouldBe Pending + } + + it should "return Pending for a kubernetes unit whose pod is absent" in { + val k8s = mock(classOf[KubernetesClient]) + when(k8s.generatePodName(42)).thenReturn("computing-unit-42") + when(k8s.getPodByName("computing-unit-42")).thenReturn(None) + ComputingUnitHelpers.singleUnitStatus(kubernetesUnit(42), k8s) shouldBe Pending + } + "getComputingUnitMetrics" should "return NaN metrics for a local unit" in { ComputingUnitHelpers.getComputingUnitMetrics(localUnit()) shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") @@ -55,4 +131,194 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit()) shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") } + + // ── Bulk variants resolving from pre-fetched pod maps ──────────────── + + "getComputingUnitStatus(unit, podPhases)" should "return Running for a local unit" in { + ComputingUnitHelpers.getComputingUnitStatus(localUnit(), Map.empty) shouldBe Running + } + + it should "return Running for a kubernetes unit whose pod phase is Running" in { + val unit = kubernetesUnit(7) + val podPhases = Map(KubernetesClient.generatePodName(7) -> "Running") + ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Running + } + + it should "return Pending for a kubernetes unit whose pod is absent or not Running" in { + val unit = kubernetesUnit(8) + ComputingUnitHelpers.getComputingUnitStatus(unit, Map.empty) shouldBe Pending + ComputingUnitHelpers.getComputingUnitStatus( + unit, + Map(KubernetesClient.generatePodName(8) -> "Pending") + ) shouldBe Pending + } + + it should "treat a null phase as not Running" in { + val unit = kubernetesUnit(9) + val podPhases = Map(KubernetesClient.generatePodName(9) -> (null: String)) + ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Pending + } + + "getComputingUnitMetrics(unit, podMetrics)" should "return NaN metrics for a local unit" in { + ComputingUnitHelpers.getComputingUnitMetrics(localUnit(), Map.empty) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + it should "resolve cpu/memory for a kubernetes unit from the map" in { + val unit = kubernetesUnit(10) + val podMetrics = Map( + KubernetesClient.generatePodName(10) -> Map("cpu" -> "500m", "memory" -> "256Mi") + ) + ComputingUnitHelpers.getComputingUnitMetrics(unit, podMetrics) shouldBe + WorkflowComputingUnitMetrics("500m", "256Mi") + } + + it should "return empty cpu/memory for a kubernetes unit absent from the map" in { + ComputingUnitHelpers.getComputingUnitMetrics(kubernetesUnit(11), Map.empty) shouldBe + WorkflowComputingUnitMetrics("", "") + } + + // ── partitionLiveUnits ─────────────────────────────────────────────── + + "partitionLiveUnits" should "treat local units as always live" in { + val units = List(localUnit(cuid = 1), localUnit(cuid = 2)) + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(units, Map.empty) + live.map(_.getCuid) shouldBe List(1, 2) + vanished shouldBe empty + } + + it should "classify a kubernetes unit as live iff its pod is present in the map" in { + val present = kubernetesUnit(20) + val gone = kubernetesUnit(21) + val podPhases = Map(KubernetesClient.generatePodName(20) -> "Running") + + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(present, gone), podPhases) + + live.map(_.getCuid) shouldBe List(20) + vanished.map(_.getCuid) shouldBe List(21) + } + + it should "treat an untyped (null-type) unit as live (never kubernetes)" in { + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(untypedUnit()), Map.empty) + live should have size 1 + vanished shouldBe empty + } + + // ── buildDashboardUnit ─────────────────────────────────────────────── + + "buildDashboardUnit" should "populate the row from the caller flags and pre-fetched maps" in { + val unit = kubernetesUnit(cuid = 30, uid = 100) + val podName = KubernetesClient.generatePodName(30) + + val row = ComputingUnitHelpers.buildDashboardUnit( + unit, + isOwner = true, + accessPrivilege = PrivilegeEnum.READ, + ownerInfo = Map((100: Integer) -> ("avatar", "owner")), + podPhases = Map(podName -> "Running"), + podMetrics = Map(podName -> Map("cpu" -> "100m", "memory" -> "64Mi")) + ) + + row.computingUnit.getCuid shouldBe 30 + row.isOwner shouldBe true + row.accessPrivilege shouldBe PrivilegeEnum.READ + row.status shouldBe "Running" + row.metrics shouldBe WorkflowComputingUnitMetrics("100m", "64Mi") + row.ownerGoogleAvatar shouldBe "avatar" + row.ownerName shouldBe "owner" + } + + it should "fall back to null owner info when the owner is missing from the map" in { + val row = ComputingUnitHelpers.buildDashboardUnit( + localUnit(cuid = 31, uid = 200), + isOwner = false, + accessPrivilege = PrivilegeEnum.WRITE, + ownerInfo = Map.empty, + podPhases = Map.empty, + podMetrics = Map.empty + ) + + row.ownerGoogleAvatar shouldBe null + row.ownerName shouldBe null + row.status shouldBe "Running" + row.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") + } + + // ── Bulk variants: unknown (untyped) branch ────────────────────────── + + "getComputingUnitStatus(unit, podPhases)" should "return Pending for an unknown (untyped) unit" in { + ComputingUnitHelpers.getComputingUnitStatus(untypedUnit(), Map.empty) shouldBe Pending + } + + "getComputingUnitMetrics(unit, podMetrics)" should "return NaN for an unknown (untyped) unit" in { + ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit(), Map.empty) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + // ── podPhasesFor / podMetricsFor guards ────────────────────────────── + + "podPhasesFor" should "return empty (issuing no cluster call) when no kubernetes unit is present" in { + ComputingUnitHelpers.podPhasesFor(List(localUnit(), untypedUnit())) shouldBe empty + } + + it should "fetch all pod phases once when a kubernetes unit is present" in { + val k8s = mock(classOf[KubernetesClient]) + val phases = Map("computing-unit-50" -> "Running") + when(k8s.getAllPodPhases).thenReturn(phases) + ComputingUnitHelpers.podPhasesFor(List(kubernetesUnit(50)), k8s) shouldBe phases + } + + "podMetricsFor" should "return empty (issuing no cluster call) when no kubernetes unit is present" in { + ComputingUnitHelpers.podMetricsFor(List(localUnit(), untypedUnit())) shouldBe empty + } + + it should "fetch all pod metrics once when a kubernetes unit is present" in { + val k8s = mock(classOf[KubernetesClient]) + val metrics = Map("computing-unit-51" -> Map("cpu" -> "100m", "memory" -> "64Mi")) + when(k8s.getAllPodMetrics).thenReturn(metrics) + ComputingUnitHelpers.podMetricsFor(List(kubernetesUnit(51)), k8s) shouldBe metrics + } + + // ── resolveOwnerInfo (backed by the embedded database) ─────────────── + + "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in { + userDao.insert(makeUser(500, "alice", "[email protected]", "alice-avatar")) + userDao.insert(makeUser(501, "", "[email protected]", "")) + + val info = ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq[Integer](500, 501)) + info(500) shouldBe (("alice-avatar", "alice")) + info(501) shouldBe ((null, null)) + } + + it should "return an empty map (and issue no query) for no uids" in { + ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq.empty) shouldBe empty + } + + // ── reconcileVanishedKubernetesUnits (backed by the embedded database) ─ + + "reconcileVanishedKubernetesUnits" should "terminate vanished kubernetes units and return the live ones" in { + userDao.insert(makeUser(600, "carol", "[email protected]", null)) + + val present = kubernetesUnit(600, 600) + present.setName("present") + val gone = kubernetesUnit(601, 600) + gone.setName("gone") + val local = localUnit(602, 600) + local.setName("local") + Seq(present, gone, local).foreach(computingUnitDao.insert(_)) + + // Only the pod for cuid 600 exists; cuid 601's pod has vanished. + val podPhases = Map(KubernetesClient.generatePodName(600) -> "Running") + val live = + ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + computingUnitDao, + List(present, gone, local), + podPhases + ) + + live.map(_.getCuid) should contain theSameElementsAs Seq(600, 602) + computingUnitDao.fetchOneByCuid(601).getTerminateTime should not be null + computingUnitDao.fetchOneByCuid(600).getTerminateTime shouldBe null + computingUnitDao.fetchOneByCuid(602).getTerminateTime shouldBe null + } } 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 8ce3b107ae..f0d96347a9 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 @@ -19,11 +19,97 @@ package org.apache.texera.service.util +import io.fabric8.kubernetes.api.model.metrics.v1beta1.{ + ContainerMetricsBuilder, + PodMetrics, + PodMetricsBuilder, + PodMetricsList, + PodMetricsListBuilder +} +import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity} +import io.fabric8.kubernetes.client.dsl.{ + MetricAPIGroupDSL, + MixedOperation, + NonNamespaceOperation, + PodMetricOperation, + PodResource +} +import io.fabric8.kubernetes.client.{KubernetesClient => Fabric8Client} +import org.apache.texera.common.config.KubernetesConfig +import org.mockito.Mockito.{mock, when} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import scala.jdk.CollectionConverters._ + +// Two layers are exercised here: +// * the pure fabric8 -> map transforms (phasesByPodName / metricsByPodName) with +// builder-constructed model objects, so the transform logic needs no client, and +// * the thin namespace-wide wrappers (getAllPodPhases / getAllPodMetrics / getPodMetrics), +// which are driven through a freshly constructed KubernetesClient whose fabric8 client is a +// Mockito stub — no live cluster and no mutable global. +// The status/metrics *decision* logic that consumes these maps (Running vs Pending, cpu/memory +// resolution) is covered by ComputingUnitHelpersSpec. class KubernetesClientSpec extends AnyFlatSpec with Matchers { + private val namespace: String = KubernetesConfig.computeUnitPoolNamespace + + // A fabric8 client stubbed just enough to answer the namespace-wide pod-list and pod-metrics + // calls the wrappers make. RETURNS_DEEP_STUBS can't be used: fabric8's fluent API returns type + // variables, so each step of the chain is mocked explicitly. + private def stubbedClient(pods: Seq[Pod], metrics: Seq[PodMetrics]): Fabric8Client = { + val client = mock(classOf[Fabric8Client]) + + val podsMixed = mock(classOf[MixedOperation[_, _, _]]) + .asInstanceOf[MixedOperation[Pod, PodList, PodResource]] + val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]]) + .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]] + when(client.pods()).thenReturn(podsMixed) + when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace) + when(podsInNamespace.list()).thenReturn(new PodListBuilder().addAllToItems(pods.asJava).build()) + + val top = mock(classOf[MetricAPIGroupDSL]) + val podMetricOp = mock(classOf[PodMetricOperation]) + val metricsList: PodMetricsList = + new PodMetricsListBuilder().addAllToItems(metrics.asJava).build() + when(client.top()).thenReturn(top) + when(top.pods()).thenReturn(podMetricOp) + when(podMetricOp.metrics(namespace)).thenReturn(metricsList) + + client + } + + private def pod(cuid: Int, phase: String): Pod = + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(cuid)) + .endMetadata() + .withNewStatus() + .withPhase(phase) + .endStatus() + .build() + + // A pod whose status has not been populated yet (getStatus == null). + private def statuslessPod(cuid: Int): Pod = + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(cuid)) + .endMetadata() + .build() + + private def podMetrics(cuid: Int, cpu: String, memory: String): PodMetrics = + new PodMetricsBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(cuid)) + .endMetadata() + .addToContainers( + new ContainerMetricsBuilder() + .withName("main") + .withUsage(Map("cpu" -> new Quantity(cpu), "memory" -> new Quantity(memory)).asJava) + .build() + ) + .build() + "generatePodName" should "prefix the cuid with computing-unit" in { KubernetesClient.generatePodName(42) shouldBe "computing-unit-42" } @@ -31,4 +117,47 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers { it should "handle a cuid of 0" in { KubernetesClient.generatePodName(0) shouldBe "computing-unit-0" } + + "phasesByPodName" should "map every pod name to its phase" in { + val phases = KubernetesClient.phasesByPodName(Seq(pod(1, "Running"), pod(2, "Pending"))) + phases(KubernetesClient.generatePodName(1)) shouldBe "Running" + phases(KubernetesClient.generatePodName(2)) shouldBe "Pending" + } + + it should "map a pod with no status to a null phase but still include it" in { + val phases = KubernetesClient.phasesByPodName(Seq(statuslessPod(3))) + phases should contain key KubernetesClient.generatePodName(3) + phases(KubernetesClient.generatePodName(3)) shouldBe null + } + + "metricsByPodName" should "flatten each pod's container usage into a cpu/memory map" in { + val metrics = + KubernetesClient.metricsByPodName(Seq(podMetrics(1, "250m", "128Mi"))) + metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") + } + + // ── namespace-wide wrappers, driven through a stubbed fabric8 client ── + // These pin the fabric8 fluent-chain plumbing (list() / top().pods().metrics()); the value + // transform they delegate to is already pinned by the phasesByPodName / metricsByPodName tests, + // so they only assert that the namespace items flow through keyed by pod name. + + "getAllPodPhases" should "list the namespace pods and key them by pod name" in { + val k8s = + new KubernetesClient(stubbedClient(Seq(pod(1, "Running"), pod(2, "Pending")), Seq.empty)) + k8s.getAllPodPhases.keySet shouldBe + Set(KubernetesClient.generatePodName(1), KubernetesClient.generatePodName(2)) + } + + "getAllPodMetrics" should "fetch the namespace metrics and key them by pod name" in { + val k8s = new KubernetesClient(stubbedClient(Seq.empty, Seq(podMetrics(1, "250m", "128Mi")))) + k8s.getAllPodMetrics.keySet shouldBe Set(KubernetesClient.generatePodName(1)) + } + + // getPodMetrics adds its own collectFirst-by-name lookup on top of the transform, so it asserts + // both the matched pod's usage and the no-match fallback. + "getPodMetrics" should "return the matching pod's usage and an empty map when none matches" in { + val k8s = new KubernetesClient(stubbedClient(Seq.empty, Seq(podMetrics(1, "250m", "128Mi")))) + k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") + k8s.getPodMetrics(999) shouldBe empty + } }
