You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/29 18:09:50 UTC
(texera) branch gh-readonly-queue/main/pr-6853-a61702fd10ccde130f0a24a1298adff24c4a38a5 created (now 6f5675b0c0)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6853-a61702fd10ccde130f0a24a1298adff24c4a38a5
in repository https://gitbox.apache.org/repos/asf/texera.git
at 6f5675b0c0 refactor(computing-unit-managing-service): share CU listing helpers and batch k8s calls (#6853)
This branch includes the following new commits:
new 6f5675b0c0 refactor(computing-unit-managing-service): share CU listing helpers and batch k8s calls (#6853)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: refactor(computing-unit-managing-service): share CU listing helpers and batch k8s calls (#6853)
Posted by gi...@apache.org.
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-a61702fd10ccde130f0a24a1298adff24c4a38a5
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 6f5675b0c06fd7cec823ba9151021c864f1a4568
Author: Kunwoo (Chris) <14...@users.noreply.github.com>
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) <14...@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
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("owner@example.com")
+ 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", "alice@example.com", "alice-avatar"))
+ userDao.insert(makeUser(501, "", "bob@example.com", ""))
+
+ 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", "carol@example.com", 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
+ }
}