This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 33cc5d1609 refactor(computing-unit-managing-service): share CU listing
helpers and batch k8s calls (#6853)
33cc5d1609 is described below
commit 33cc5d16096aa05095f483ee8724e799c95d7898
Author: Kunwoo (Chris) <[email protected]>
AuthorDate: Sat Aug 1 11:59:32 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 b10661fd70..3779413e04 100644
--- a/build.sbt
+++ b/build.sbt
@@ -167,7 +167,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
+ }
}
