From 594bfe34d17483aece788716f7878d4c7e066847 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Wed, 14 Oct 2026 10:26:11 -0700
Subject: [PATCH 1/3] feat(computing-unit): add admin cluster capacity endpoint

---
 ...omputing-unit-manager-service-account.yaml |  29 ++
 .../resource/AdminComputingUnitResource.scala |  14 +-
 .../texera/service/util/ClusterCapacity.scala | 146 ++++++++
 .../service/util/ComputingUnitHelpers.scala   |  25 ++
 .../service/util/KubernetesClient.scala       |  21 ++
 .../AdminComputingUnitResourceSpec.scala      |  16 +
 .../service/util/ClusterCapacitySpec.scala    | 321 ++++++++++++++++++
 .../util/ComputingUnitHelpersSpec.scala       |  35 ++
 .../service/util/KubernetesClientSpec.scala   |  44 +++
 9 files changed, 650 insertions(+), 1 deletion(-)
 create mode 100644 computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
 create mode 100644 computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala

diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
index a1a0ed40cc7..b3435c4a58d 100644
--- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
+++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
@@ -63,3 +63,32 @@ roleRef:
   kind: Role
   name: {{ .Values.workflowComputingUnitManager.name }}
   apiGroup: rbac.authorization.k8s.io
+
+---
+# Read-only, cluster-wide, for the admin cluster-capacity endpoint. Nodes are cluster-scoped, and
+# what a new computing unit can claim of them depends on the requests of the pods of every
+# namespace, not only the pool's, so this cannot be the namespaced Role above. It only lists: the
+# endpoint reads the nodes and pods once per request and never watches or fetches one by name.
+# Named for the release because a ClusterRole is cluster-scoped and releases would collide.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+rules:
+  - apiGroups: [""]
+    resources: ["nodes", "pods"]
+    verbs: ["list"]
+
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+subjects:
+  - kind: ServiceAccount
+    name: {{ .Values.workflowComputingUnitManager.serviceAccountName }}
+    namespace: {{ .Release.Namespace }}
+roleRef:
+  kind: ClusterRole
+  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+  apiGroup: rbac.authorization.k8s.io
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
index 799c58b270d..1c20148cf8a 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
@@ -30,9 +30,10 @@ import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
 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.DashboardWorkflowComputingUnit
-import org.apache.texera.service.util.ComputingUnitHelpers
+import org.apache.texera.service.util.{ClusterCapacity, ComputingUnitHelpers}
 import org.jooq.DSLContext
 
+import scala.annotation.unused
 import scala.jdk.CollectionConverters.CollectionHasAsScala
 
 object AdminComputingUnitResource {
@@ -105,4 +106,15 @@ class AdminComputingUnitResource {
       )
     }
   }
+
+  /**
+    * Total, allocated and available CPU (millicores), memory (bytes) and GPU (devices) across the
+    * cluster's schedulable nodes (ADMIN-only). Allocated is what the scheduler has committed to
+    * pods' requests, so available is an upper bound on what a new computing unit can claim. 404
+    * when computing units do not run on Kubernetes.
+    */
+  @GET
+  @Path("/cluster-capacity")
+  def getClusterCapacity(@Auth @unused user: SessionUser): ClusterCapacity =
+    ComputingUnitHelpers.clusterCapacity()
 }
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
new file mode 100644
index 00000000000..0387da054de
--- /dev/null
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.util
+
+import io.fabric8.kubernetes.api.model.{Container, Node, Pod, PodSpec, Quantity}
+import io.fabric8.kubernetes.client.readiness.Readiness
+
+import scala.jdk.CollectionConverters._
+
+/**
+  * One resource across the cluster's schedulable nodes. CPU is in millicores, memory in bytes
+  * and GPU in devices, so a consumer can compute a percentage or pick a display unit without
+  * parsing Kubernetes quantity strings. `available` is `total - allocated`, never below zero.
+  */
+case class ResourceCapacity(total: Long, allocated: Long, available: Long)
+
+case class ClusterCapacity(cpu: ResourceCapacity, memory: ResourceCapacity, gpu: ResourceCapacity)
+
+object ClusterCapacity {
+
+  // Quantities in base units (cores, bytes, devices), keyed by Kubernetes resource name.
+  private type Amounts = Map[String, BigDecimal]
+
+  private val Cpu = "cpu"
+  private val Memory = "memory"
+  private val MillicoresPerCore = 1000
+
+  // A pod in either phase has stopped and no longer holds anything on its node.
+  private val FinishedPhases = Set("Succeeded", "Failed")
+
+  /**
+    * Aggregates what the nodes offer against what the scheduler has already handed out.
+    *
+    * `total` is the allocatable of every Ready, uncordoned node, and `allocated` is the requests
+    * of the pods occupying those nodes. These are the figures `kubectl describe node`
+    * reports as "Allocated resources", and they are what kube-scheduler fits a new pod against,
+    * so `available` is what bounds a new computing unit. Live usage is deliberately not used: it
+    * reads well below what is committed, and metrics-server reports no GPU at all.
+    *
+    * Not modelled, so `available` is an upper bound on what a unit can claim:
+    *   - per-node fit: the figures are sums across nodes, so capacity spread thinly over many
+    *     nodes can still refuse a unit that needs one node's worth;
+    *   - the pool namespace's ResourceQuota, which refuses a unit once it is used up however
+    *     much the nodes have free;
+    *   - which nodes a unit may land on (taints, node selectors, runtime-class scheduling), so
+    *     a node it cannot use still counts toward `total`;
+    *   - pod-level `spec.resources`, which newer clusters let a pod set in place of
+    *     per-container requests;
+    *   - native sidecars (init containers with restartPolicy Always), which the scheduler adds
+    *     to the app containers instead of taking the larger of the two, so a pod using one is
+    *     charged less than it holds.
+    */
+  def fromNodesAndPods(
+      nodes: Iterable[Node],
+      pods: Iterable[Pod],
+      gpuResourceKey: String
+  ): ClusterCapacity = {
+    val schedulable = nodes.filter(isSchedulable)
+    val nodeNames = schedulable.map(_.getMetadata.getName).toSet
+
+    val total = sum(schedulable.map(node => amounts(node.getStatus.getAllocatable)))
+    val allocated = sum(pods.flatMap(pod => occupiedRequests(pod, nodeNames)))
+
+    def capacity(resource: String, unitsPerAmount: Int): ResourceCapacity = {
+      val totalUnits = toUnits(total, resource, unitsPerAmount)
+      val allocatedUnits = toUnits(allocated, resource, unitsPerAmount)
+      ResourceCapacity(totalUnits, allocatedUnits, (totalUnits - allocatedUnits).max(0))
+    }
+    ClusterCapacity(
+      cpu = capacity(Cpu, MillicoresPerCore),
+      memory = capacity(Memory, 1),
+      gpu = capacity(gpuResourceKey, 1)
+    )
+  }
+
+  // A cordoned node takes no new pods, and neither does one that is not Ready (the node
+  // controller taints it for exactly that reason), so neither can hold a new computing unit.
+  private def isSchedulable(node: Node): Boolean = {
+    val cordoned =
+      Option(node.getSpec).flatMap(spec => Option(spec.getUnschedulable)).exists(_.booleanValue)
+    !cordoned && Readiness.isNodeReady(node)
+  }
+
+  // The scheduler's accounting: a pod bound to a node counts until it finishes, whether or not its
+  // containers have started, and a pod it has not bound yet counts nowhere.
+  private def occupiedRequests(pod: Pod, nodeNames: Set[String]): Option[Amounts] =
+    for {
+      spec <- Option(pod.getSpec)
+      nodeName <- Option(spec.getNodeName) if nodeNames.contains(nodeName)
+      if !Option(pod.getStatus).flatMap(s => Option(s.getPhase)).exists(FinishedPhases.contains)
+    } yield podRequests(spec)
+
+  // Init containers run one at a time before the app containers, so a pod needs the larger of
+  // the app containers' sum and its biggest init container, plus the runtime-class overhead.
+  private def podRequests(spec: PodSpec): Amounts = {
+    val appContainers = sum(containers(spec.getContainers).map(requests))
+    val peakInit = peak(containers(spec.getInitContainers).map(requests))
+    sum(Seq(peak(Seq(appContainers, peakInit)), amounts(spec.getOverhead)))
+  }
+
+  private def containers(list: java.util.List[Container]): Seq[Container] =
+    Option(list).map(_.asScala.toSeq).getOrElse(Seq.empty)
+
+  private def requests(container: Container): Amounts =
+    Option(container.getResources)
+      .map(resources => amounts(resources.getRequests))
+      .getOrElse(Map.empty)
+
+  private def amounts(quantities: java.util.Map[String, Quantity]): Amounts =
+    Option(quantities)
+      .map(_.asScala.map {
+        case (resource, quantity) =>
+          resource -> BigDecimal(quantity.getNumericalAmount)
+      }.toMap)
+      .getOrElse(Map.empty)
+
+  // Both treat a resource absent from an amount as zero. Requests are never negative, so zero is
+  // also the identity for `peak`'s max.
+  private def sum(all: Iterable[Amounts]): Amounts =
+    all.flatten.groupMapReduce(_._1)(_._2)(_ + _)
+
+  private def peak(all: Iterable[Amounts]): Amounts =
+    all.flatten.groupMapReduce(_._1)(_._2)(_ max _)
+
+  private def toUnits(all: Amounts, resource: String, unitsPerAmount: Int): Long =
+    (all.getOrElse(resource, BigDecimal(0)) * unitsPerAmount)
+      .setScale(0, BigDecimal.RoundingMode.HALF_UP)
+      .toLong
+}
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 e0506b52fb5..a6ffc3f29a0 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
@@ -18,6 +18,9 @@
 
 package org.apache.texera.service.util
 
+import io.fabric8.kubernetes.client.KubernetesClientException
+import jakarta.ws.rs.NotFoundException
+import org.apache.texera.common.config.KubernetesConfig
 import org.apache.texera.dao.jooq.generated.enums.{
   WorkflowComputingUnitTerminationReasonEnum,
   WorkflowComputingUnitTypeEnum
@@ -276,6 +279,28 @@ object ComputingUnitHelpers {
   ): Map[String, Map[String, String]] =
     if (units.exists(isKubernetes)) k8s.getAllPodMetrics else Map.empty
 
+  // Same seam as podSnapshotsFor: the public overload binds the production singleton, passed
+  // by-name so a deployment without Kubernetes units is refused without ever forcing the
+  // singleton, which builds a fabric8 client, and tests drive the private overload with a stub.
+  def clusterCapacity(): ClusterCapacity =
+    clusterCapacity(KubernetesConfig.kubernetesComputingUnitEnabled, KubernetesClient)
+
+  private[util] def clusterCapacity(enabled: Boolean, k8s: => KubernetesClient): ClusterCapacity = {
+    if (!enabled) {
+      throw new NotFoundException(
+        "Cluster capacity is only available when computing units run on Kubernetes."
+      )
+    }
+    try {
+      k8s.getClusterCapacity
+    } catch {
+      // Keeps the cluster's own message: a chart without the cluster-wide read grant fails with
+      // a "forbidden" that names the missing permission, which is what an admin has to act on.
+      case e: KubernetesClientException =>
+        throw ComputingUnitManagingServiceException.fromKubernetes(e)
+    }
+  }
+
   /** A Kubernetes unit whose pod is absent from `podSnapshots` (deleted or TTL GC-ed). */
   private def isVanished(
       unit: WorkflowComputingUnit,
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 e88c0ea50cf..e3a680ccb0d 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
@@ -100,6 +100,27 @@ class KubernetesClient(
   def getAllPodMetrics: Map[String, Map[String, String]] =
     metricsByPodName(fetchPodMetricsItems())
 
+  /**
+    * CPU/memory/GPU capacity of the cluster's schedulable nodes. Cluster-scoped, unlike the
+    * namespace-wide reads above, so it needs the chart's cluster-capacity ClusterRole. Finished
+    * pods are excluded by the API server so a cluster that keeps many of them does not ship them
+    * all; [[ClusterCapacity]] still skips them, which is the rule and is tested there, so the
+    * selectors are an optimisation only.
+    */
+  def getClusterCapacity: ClusterCapacity =
+    ClusterCapacity.fromNodesAndPods(
+      client.nodes().list().getItems.asScala,
+      client
+        .pods()
+        .inAnyNamespace()
+        .withoutField("status.phase", "Succeeded")
+        .withoutField("status.phase", "Failed")
+        .list()
+        .getItems
+        .asScala,
+      KubernetesConfig.gpuResourceKey
+    )
+
   def getPodMetrics(cuid: Int): Map[String, String] = {
     val targetPodName = generatePodName(cuid)
     fetchPodMetricsItems()
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
index f1735935c15..77087c2ec5b 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
@@ -20,7 +20,9 @@
 package org.apache.texera.service.resource
 
 import jakarta.annotation.security.RolesAllowed
+import jakarta.ws.rs.NotFoundException
 import org.apache.texera.auth.SessionUser
+import org.apache.texera.common.config.KubernetesConfig
 import org.apache.texera.dao.MockTexeraDB
 import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_COMPUTING_UNIT
 import org.apache.texera.dao.jooq.generated.enums.{
@@ -157,4 +159,18 @@ class AdminComputingUnitResourceSpec
     result.head.isOwner shouldBe false
     result.head.ownerName shouldBe "owner"
   }
+
+  // The test environment has kubernetes.enabled false (a local deployment), so the endpoint
+  // refuses; the enabled path is pinned in ComputingUnitHelpersSpec. Cancelled when an exported
+  // KUBERNETES_COMPUTING_UNIT_ENABLED would reach the real client instead.
+  "getClusterCapacity" should "answer 404 on a deployment without Kubernetes computing units" in {
+    assume(
+      !KubernetesConfig.kubernetesComputingUnitEnabled,
+      "kubernetes.enabled is set in this environment"
+    )
+    val refused = intercept[NotFoundException] {
+      new AdminComputingUnitResource().getClusterCapacity(new SessionUser(makeUser(740, "admin")))
+    }
+    refused.getMessage should include("Kubernetes")
+  }
 }
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
new file mode 100644
index 00000000000..b7c06285b95
--- /dev/null
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
@@ -0,0 +1,321 @@
+/*
+ * 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.util
+
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import io.fabric8.kubernetes.api.model.{
+  ContainerBuilder,
+  Node,
+  NodeBuilder,
+  Pod,
+  PodBuilder,
+  Quantity,
+  ResourceRequirements,
+  ResourceRequirementsBuilder
+}
+import io.dropwizard.jackson.Jackson
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.jdk.CollectionConverters._
+
+// The aggregation is a pure function over fabric8 model objects, so every case is driven with
+// builder-constructed nodes and pods: no client, no cluster. The rules under test are the ones
+// kube-scheduler applies when it decides whether a new pod fits, because that is the question an
+// admin is asking when a computing unit refuses to start.
+class ClusterCapacitySpec extends AnyFlatSpec with Matchers {
+
+  import ClusterCapacitySpec._
+
+  private val Gi: Long = 1024L * 1024L * 1024L
+
+  private def capacity(nodes: Seq[Node], pods: Seq[Pod]): ClusterCapacity =
+    ClusterCapacity.fromNodesAndPods(nodes, pods, gpuKey)
+
+  "fromNodesAndPods" should "report all zeros for an empty cluster" in {
+    val result = capacity(Seq.empty, Seq.empty)
+    val zero = ResourceCapacity(total = 0, allocated = 0, available = 0)
+    result shouldBe ClusterCapacity(cpu = zero, memory = zero, gpu = zero)
+  }
+
+  it should "parse decimal-suffixed and fractional quantities exactly" in {
+    val result = capacity(
+      Seq(node("a", cpu = "1.5", memory = "512M")),
+      Seq(pod("p", Some("a"), Seq(Map("cpu" -> "250m", "memory" -> "128M"))))
+    )
+    result.cpu shouldBe ResourceCapacity(total = 1500, allocated = 250, available = 1250)
+    result.memory shouldBe
+      ResourceCapacity(total = 512000000L, allocated = 128000000L, available = 384000000L)
+  }
+
+  it should "allocate the requests of pods bound to a node and report the remainder as available" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi", Some("2")), node("b", "8", "32Gi")),
+      Seq(
+        pod("p1", Some("a"), Seq(Map("cpu" -> "1", "memory" -> "2Gi", gpuKey -> "1"))),
+        pod("p2", Some("b"), Seq(Map("cpu" -> "500m", "memory" -> "1Gi")))
+      )
+    )
+    result.cpu shouldBe ResourceCapacity(total = 12000, allocated = 1500, available = 10500)
+    result.memory shouldBe ResourceCapacity(
+      total = 48 * Gi,
+      allocated = 3 * Gi,
+      available = 45 * Gi
+    )
+    result.gpu shouldBe ResourceCapacity(total = 2, allocated = 1, available = 1)
+  }
+
+  it should "sum the requests of every container in a pod" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi")),
+      Seq(pod("p", Some("a"), Seq(Map("cpu" -> "1"), Map("cpu" -> "250m", "memory" -> "1Gi"))))
+    )
+    result.cpu.allocated shouldBe 1250
+    result.memory.allocated shouldBe Gi
+  }
+
+  it should "count a container that declares no resources, or empty ones, as zero" in {
+    // System pods on a shared cluster are not always written with resources, and a container
+    // with no `resources` block at all (null) must not take the whole endpoint down.
+    val bare = new PodBuilder()
+      .withNewMetadata()
+      .withName("bare")
+      .endMetadata()
+      .withNewSpec()
+      .withNodeName("a")
+      .addToContainers(new ContainerBuilder().withName("no-resources").build())
+      .addToContainers(
+        new ContainerBuilder()
+          .withName("no-requests")
+          .withResources(new ResourceRequirements())
+          .build()
+      )
+      .endSpec()
+      .build()
+    capacity(Seq(node("a", "4", "16Gi")), Seq(bare)).cpu.allocated shouldBe 0
+  }
+
+  it should "charge a pod the larger of its containers and its biggest init container" in {
+    val result = capacity(
+      Seq(node("a", "8", "16Gi")),
+      Seq(
+        // Init containers run one at a time before the app containers, so the pod needs
+        // max(sum(containers), max(init)): the init container wins on CPU, the containers on memory.
+        pod(
+          "p",
+          Some("a"),
+          containers = Seq(Map("cpu" -> "1", "memory" -> "4Gi")),
+          initContainers = Seq(Map("cpu" -> "3", "memory" -> "1Gi"), Map("cpu" -> "2"))
+        )
+      )
+    )
+    result.cpu.allocated shouldBe 3000
+    result.memory.allocated shouldBe 4 * Gi
+  }
+
+  it should "add a pod's runtime-class overhead to what it is charged" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi")),
+      Seq(
+        pod(
+          "p",
+          Some("a"),
+          containers = Seq(Map("cpu" -> "1", "memory" -> "1Gi")),
+          overhead = Map("cpu" -> "250m", "memory" -> "120Mi")
+        )
+      )
+    )
+    result.cpu.allocated shouldBe 1250
+    result.memory.allocated shouldBe Gi + 120L * 1024 * 1024
+  }
+
+  it should "leave out a cordoned node together with the pods running on it" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi"), node("cordoned", "8", "32Gi", Some("4"), cordoned = true)),
+      Seq(
+        pod("p1", Some("a"), Seq(Map("cpu" -> "1"))),
+        pod("p2", Some("cordoned"), Seq(Map("cpu" -> "6", gpuKey -> "4")))
+      )
+    )
+    result.cpu shouldBe ResourceCapacity(total = 4000, allocated = 1000, available = 3000)
+    result.gpu shouldBe ResourceCapacity(total = 0, allocated = 0, available = 0)
+  }
+
+  it should "leave out a node that is not Ready together with the pods running on it" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi"), node("down", "8", "32Gi", ready = false)),
+      Seq(pod("p", Some("down"), Seq(Map("cpu" -> "6"))))
+    )
+    result.cpu shouldBe ResourceCapacity(total = 4000, allocated = 0, available = 4000)
+  }
+
+  it should "leave out a node that reports no Ready condition at all" in {
+    val noConditions = new NodeBuilder(node("blank", "8", "32Gi"))
+      .editStatus()
+      .withConditions()
+      .endStatus()
+      .build()
+    capacity(Seq(noConditions), Seq.empty).cpu.total shouldBe 0
+  }
+
+  it should "leave out a node with no status at all" in {
+    val statusless = new NodeBuilder().withNewMetadata().withName("new").endMetadata().build()
+    capacity(Seq(statusless, node("a", "4", "16Gi")), Seq.empty).cpu.total shouldBe 4000
+  }
+
+  it should "read GPUs under the configured resource key, whatever the vendor" in {
+    val nodes = Seq(node("a", "4", "16Gi", Some("2")))
+    ClusterCapacity.fromNodesAndPods(nodes, Seq.empty, gpuKey).gpu.total shouldBe 2
+    ClusterCapacity.fromNodesAndPods(nodes, Seq.empty, "amd.com/gpu").gpu.total shouldBe 0
+  }
+
+  it should "ignore pods the scheduler has not bound to a node" in {
+    val result = capacity(
+      Seq(node("a", "4", "16Gi")),
+      Seq(
+        pod("pending", None, Seq(Map("cpu" -> "2")), phase = "Pending"),
+        pod("blank", Some(""), Seq(Map("cpu" -> "2")), phase = "Pending")
+      )
+    )
+    result.cpu.allocated shouldBe 0
+  }
+
+  it should "count Pending, Running and Unknown pods but not finished ones" in {
+    val result = capacity(
+      Seq(node("a", "16", "64Gi")),
+      Seq(
+        pod("pending", Some("a"), Seq(Map("cpu" -> "1")), phase = "Pending"),
+        pod("running", Some("a"), Seq(Map("cpu" -> "2")), phase = "Running"),
+        pod("unknown", Some("a"), Seq(Map("cpu" -> "4")), phase = "Unknown"),
+        pod("done", Some("a"), Seq(Map("cpu" -> "8")), phase = "Succeeded"),
+        pod("crashed", Some("a"), Seq(Map("cpu" -> "16")), phase = "Failed")
+      )
+    )
+    result.cpu.allocated shouldBe 7000
+  }
+
+  it should "count a pod whose status has not been populated yet" in {
+    val statusless =
+      new PodBuilder(pod("p", Some("a"), Seq(Map("cpu" -> "2")))).withStatus(null).build()
+    capacity(Seq(node("a", "4", "16Gi")), Seq(statusless)).cpu.allocated shouldBe 2000
+  }
+
+  it should "count a pod with no spec as zero" in {
+    val specless = new PodBuilder().withNewMetadata().withName("p").endMetadata().build()
+    capacity(Seq(node("a", "4", "16Gi")), Seq(specless)).cpu.allocated shouldBe 0
+  }
+
+  it should "floor available at zero when requests exceed what the nodes advertise" in {
+    // Allocatable can shrink under running pods (a kubelet reservation raised after they landed).
+    val result = capacity(
+      Seq(node("a", "2", "4Gi")),
+      Seq(pod("p", Some("a"), Seq(Map("cpu" -> "3", "memory" -> "6Gi"))))
+    )
+    result.cpu shouldBe ResourceCapacity(total = 2000, allocated = 3000, available = 0)
+    result.memory shouldBe ResourceCapacity(total = 4 * Gi, allocated = 6 * Gi, available = 0)
+  }
+
+  // Consumers read these exact field names, so a rename is a breaking API change. Serialised the
+  // way the service does: Dropwizard's mapper with the Scala module registered in
+  // ComputingUnitManagingService.initialize.
+  "ClusterCapacity" should "serialise as cpu/memory/gpu objects of total, allocated and available" in {
+    val mapper = Jackson.newObjectMapper().registerModule(DefaultScalaModule)
+    val serialised = mapper.writeValueAsString(
+      ClusterCapacity(
+        cpu = ResourceCapacity(total = 4000, allocated = 1000, available = 3000),
+        memory = ResourceCapacity(total = 8 * Gi, allocated = Gi, available = 7 * Gi),
+        gpu = ResourceCapacity(total = 2, allocated = 1, available = 1)
+      )
+    )
+    mapper.readTree(serialised) shouldBe mapper.readTree("""{
+      "cpu": {"total": 4000, "allocated": 1000, "available": 3000},
+      "memory": {"total": 8589934592, "allocated": 1073741824, "available": 7516192768},
+      "gpu": {"total": 2, "allocated": 1, "available": 1}
+    }""")
+  }
+}
+
+object ClusterCapacitySpec {
+
+  private val gpuKey = "nvidia.com/gpu"
+
+  private def quantities(values: Map[String, String]): java.util.Map[String, Quantity] =
+    values.map { case (resource, amount) => resource -> new Quantity(amount) }.asJava
+
+  // `unschedulable` is left unset (null, as the API server omits it) unless the node is cordoned.
+  private[util] def node(
+      name: String,
+      cpu: String,
+      memory: String,
+      gpu: Option[String] = None,
+      ready: Boolean = true,
+      cordoned: Boolean = false
+  ): Node = {
+    val allocatable = Map("cpu" -> cpu, "memory" -> memory) ++ gpu.map(gpuKey -> _)
+    val builder = new NodeBuilder()
+      .withNewMetadata()
+      .withName(name)
+      .endMetadata()
+      .withNewSpec()
+      .endSpec()
+      .withNewStatus()
+      .withAllocatable(quantities(allocatable))
+      .addNewCondition()
+      .withType("Ready")
+      .withStatus(if (ready) "True" else "False")
+      .endCondition()
+      .endStatus()
+    if (cordoned) builder.editSpec().withUnschedulable(true).endSpec()
+    builder.build()
+  }
+
+  // A pod with one app container per entry of `containers`, each asking for the given requests.
+  private[util] def pod(
+      name: String,
+      nodeName: Option[String],
+      containers: Seq[Map[String, String]],
+      phase: String = "Running",
+      initContainers: Seq[Map[String, String]] = Seq.empty,
+      overhead: Map[String, String] = Map.empty
+  ): Pod = {
+    def container(requested: Map[String, String]) =
+      new ContainerBuilder()
+        .withName("c")
+        .withResources(
+          new ResourceRequirementsBuilder().withRequests(quantities(requested)).build()
+        )
+        .build()
+
+    new PodBuilder()
+      .withNewMetadata()
+      .withName(name)
+      .endMetadata()
+      .withNewSpec()
+      .withNodeName(nodeName.orNull)
+      .withContainers(containers.map(container).asJava)
+      .withInitContainers(initContainers.map(container).asJava)
+      .withOverhead(quantities(overhead))
+      .endSpec()
+      .withNewStatus()
+      .withPhase(phase)
+      .endStatus()
+      .build()
+  }
+}
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 2b7eb009572..ea482f6ed26 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
@@ -20,6 +20,8 @@
 package org.apache.texera.service.util
 
 import io.fabric8.kubernetes.api.model.{Pod, PodBuilder}
+import io.fabric8.kubernetes.client.KubernetesClientException
+import jakarta.ws.rs.NotFoundException
 import org.apache.texera.dao.MockTexeraDB
 import org.apache.texera.dao.jooq.generated.enums.{
   PrivilegeEnum,
@@ -800,6 +802,39 @@ class ComputingUnitHelpersSpec
     ComputingUnitHelpers.podMetricsFor(List(kubernetesUnit(51)), k8s) shouldBe metrics
   }
 
+  // ── clusterCapacity guard ────────────────────────────────────────────
+
+  // Same by-name seam as the two guards above. Passing `fail(...)` as the client proves the
+  // disabled branch never forces it.
+  "clusterCapacity" should "refuse, without touching the cluster, when kubernetes units are disabled" in {
+    val refused = intercept[NotFoundException] {
+      ComputingUnitHelpers.clusterCapacity(enabled = false, fail("the client must not be forced"))
+    }
+    refused.getMessage should include("Kubernetes")
+  }
+
+  it should "return the cluster's capacity when kubernetes units are enabled" in {
+    val k8s = mock(classOf[KubernetesClient])
+    val capacity = ClusterCapacity(
+      cpu = ResourceCapacity(total = 4000, allocated = 1000, available = 3000),
+      memory = ResourceCapacity(total = 8L << 30, allocated = 1L << 30, available = 7L << 30),
+      gpu = ResourceCapacity(total = 2, allocated = 1, available = 1)
+    )
+    when(k8s.getClusterCapacity).thenReturn(capacity)
+    ComputingUnitHelpers.clusterCapacity(enabled = true, k8s) shouldBe capacity
+  }
+
+  it should "surface a Kubernetes API failure as a service error that keeps the cluster's message" in {
+    val denied = new KubernetesClientException("nodes is forbidden: cannot list resource \"nodes\"")
+    val k8s = mock(classOf[KubernetesClient])
+    when(k8s.getClusterCapacity).thenThrow(denied)
+
+    val thrown = intercept[ComputingUnitManagingServiceException] {
+      ComputingUnitHelpers.clusterCapacity(enabled = true, k8s)
+    }
+    thrown shouldBe InternalError(denied.getMessage)
+  }
+
   // ── resolveOwnerInfo (backed by the embedded database) ───────────────
 
   "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in {
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 2723c35a53b..e5a3a36a8f0 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
@@ -29,6 +29,9 @@ import io.fabric8.kubernetes.api.model.metrics.v1beta1.{
 import io.fabric8.kubernetes.api.model.{
   Container,
   ContainerBuilder,
+  Node,
+  NodeList,
+  NodeListBuilder,
   Pod,
   PodBuilder,
   PodList,
@@ -37,6 +40,8 @@ import io.fabric8.kubernetes.api.model.{
   ResourceRequirementsBuilder
 }
 import io.fabric8.kubernetes.client.dsl.{
+  AnyNamespaceOperation,
+  FilterWatchListDeletable,
   MetricAPIGroupDSL,
   MixedOperation,
   NamespaceableResource,
@@ -291,6 +296,45 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
     k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
     k8s.getPodMetrics(999) shouldBe empty
   }
+
+  // ── cluster-wide capacity ──
+  // Reads cluster-scoped nodes and the pods of every namespace, narrowed on the server to the
+  // pods that have not finished. The stub answers only that chain: a wrapper that listed just the
+  // pool namespace sees no pods and one that skipped the selectors gets null back, and either
+  // fails; the arithmetic itself is pinned by ClusterCapacitySpec.
+
+  private def clusterClient(nodes: Seq[Node], pods: Seq[Pod]): Fabric8Client = {
+    val client = stubbedClient(Seq.empty, Seq.empty)
+
+    val nodeOp = mock(classOf[NonNamespaceOperation[_, _, _]])
+      .asInstanceOf[NonNamespaceOperation[Node, NodeList, Resource[Node]]]
+    when(client.nodes()).thenReturn(nodeOp)
+    when(nodeOp.list()).thenReturn(new NodeListBuilder().addAllToItems(nodes.asJava).build())
+
+    def filterable() =
+      mock(classOf[FilterWatchListDeletable[_, _, _]])
+        .asInstanceOf[FilterWatchListDeletable[Pod, PodList, PodResource]]
+    val podsAnywhere = mock(classOf[AnyNamespaceOperation[_, _, _]])
+      .asInstanceOf[AnyNamespaceOperation[Pod, PodList, PodResource]]
+    val notSucceeded = filterable()
+    val notFinished = filterable()
+    when(client.pods().inAnyNamespace()).thenReturn(podsAnywhere)
+    when(podsAnywhere.withoutField("status.phase", "Succeeded")).thenReturn(notSucceeded)
+    when(notSucceeded.withoutField("status.phase", "Failed")).thenReturn(notFinished)
+    when(notFinished.list()).thenReturn(new PodListBuilder().addAllToItems(pods.asJava).build())
+
+    client
+  }
+
+  "getClusterCapacity" should "aggregate every node against the unfinished pods of every namespace" in {
+    val node = ClusterCapacitySpec.node("node-1", cpu = "4", memory = "16Gi")
+    val systemPod = ClusterCapacitySpec.pod("coredns", Some("node-1"), Seq(Map("cpu" -> "100m")))
+
+    val capacity = new KubernetesClient(clusterClient(Seq(node), Seq(systemPod))).getClusterCapacity
+
+    capacity.cpu shouldBe ResourceCapacity(total = 4000, allocated = 100, available = 3900)
+  }
+
   // ── single-pod lookups, creation and deletion ──
   // These reach the rest of the fluent chain: withName(...).get() for the lookups,
   // resource(pod).inNamespace(...).create() for creation, and .delete() for removal. Everything

From ad6bd0df3d4e7f0121adbdf413e9fd7f05d07602 Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Wed, 14 Oct 2026 10:37:23 -0700
Subject: [PATCH 2/3] refactor(computing-unit): shorten cluster capacity
 comments

---
 ...omputing-unit-manager-service-account.yaml |  8 ++--
 .../resource/AdminComputingUnitResource.scala |  6 +--
 .../texera/service/util/ClusterCapacity.scala | 45 +++++--------------
 .../service/util/ComputingUnitHelpers.scala   |  8 ++--
 .../service/util/KubernetesClient.scala       |  8 ++--
 .../AdminComputingUnitResourceSpec.scala      |  6 +--
 .../service/util/ClusterCapacitySpec.scala    | 16 +++----
 .../util/ComputingUnitHelpersSpec.scala       |  3 +-
 .../service/util/KubernetesClientSpec.scala   |  6 +--
 9 files changed, 34 insertions(+), 72 deletions(-)

diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
index b3435c4a58d..c8bfb0ccba3 100644
--- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
+++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
@@ -65,11 +65,9 @@ roleRef:
   apiGroup: rbac.authorization.k8s.io
 
 ---
-# Read-only, cluster-wide, for the admin cluster-capacity endpoint. Nodes are cluster-scoped, and
-# what a new computing unit can claim of them depends on the requests of the pods of every
-# namespace, not only the pool's, so this cannot be the namespaced Role above. It only lists: the
-# endpoint reads the nodes and pods once per request and never watches or fetches one by name.
-# Named for the release because a ClusterRole is cluster-scoped and releases would collide.
+# Read-only and cluster-wide for the admin cluster-capacity endpoint: nodes are cluster-scoped and
+# every namespace's pods take their capacity, so the namespaced Role above cannot cover it.
+# Named for the release because a ClusterRole is cluster-scoped.
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
index 1c20148cf8a..768d0828b80 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala
@@ -108,10 +108,8 @@ class AdminComputingUnitResource {
   }
 
   /**
-    * Total, allocated and available CPU (millicores), memory (bytes) and GPU (devices) across the
-    * cluster's schedulable nodes (ADMIN-only). Allocated is what the scheduler has committed to
-    * pods' requests, so available is an upper bound on what a new computing unit can claim. 404
-    * when computing units do not run on Kubernetes.
+    * Cluster CPU, memory and GPU as total, allocated and available, an upper bound for a new unit
+    * (ADMIN-only). 404 when units do not run on Kubernetes.
     */
   @GET
   @Path("/cluster-capacity")
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
index 0387da054de..a8639304cb1 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
@@ -24,11 +24,7 @@ import io.fabric8.kubernetes.client.readiness.Readiness
 
 import scala.jdk.CollectionConverters._
 
-/**
-  * One resource across the cluster's schedulable nodes. CPU is in millicores, memory in bytes
-  * and GPU in devices, so a consumer can compute a percentage or pick a display unit without
-  * parsing Kubernetes quantity strings. `available` is `total - allocated`, never below zero.
-  */
+/** CPU is in millicores, memory in bytes and GPU in devices. `available` is never below zero. */
 case class ResourceCapacity(total: Long, allocated: Long, available: Long)
 
 case class ClusterCapacity(cpu: ResourceCapacity, memory: ResourceCapacity, gpu: ResourceCapacity)
@@ -42,30 +38,17 @@ object ClusterCapacity {
   private val Memory = "memory"
   private val MillicoresPerCore = 1000
 
-  // A pod in either phase has stopped and no longer holds anything on its node.
+  // Pods in these phases hold nothing on their node.
   private val FinishedPhases = Set("Succeeded", "Failed")
 
   /**
-    * Aggregates what the nodes offer against what the scheduler has already handed out.
+    * Sums what the nodes offer against the requests the scheduler has handed out, the "Allocated
+    * resources" of `kubectl describe node`, so `available` bounds what a new unit can claim. Live
+    * usage is not used: it reads below what is committed and has no GPU metric.
     *
-    * `total` is the allocatable of every Ready, uncordoned node, and `allocated` is the requests
-    * of the pods occupying those nodes. These are the figures `kubectl describe node`
-    * reports as "Allocated resources", and they are what kube-scheduler fits a new pod against,
-    * so `available` is what bounds a new computing unit. Live usage is deliberately not used: it
-    * reads well below what is committed, and metrics-server reports no GPU at all.
-    *
-    * Not modelled, so `available` is an upper bound on what a unit can claim:
-    *   - per-node fit: the figures are sums across nodes, so capacity spread thinly over many
-    *     nodes can still refuse a unit that needs one node's worth;
-    *   - the pool namespace's ResourceQuota, which refuses a unit once it is used up however
-    *     much the nodes have free;
-    *   - which nodes a unit may land on (taints, node selectors, runtime-class scheduling), so
-    *     a node it cannot use still counts toward `total`;
-    *   - pod-level `spec.resources`, which newer clusters let a pod set in place of
-    *     per-container requests;
-    *   - native sidecars (init containers with restartPolicy Always), which the scheduler adds
-    *     to the app containers instead of taking the larger of the two, so a pod using one is
-    *     charged less than it holds.
+    * Not modelled, so `available` is an upper bound: per-node fit, the pool's ResourceQuota, node
+    * placement (taints, selectors, runtime classes), pod-level `spec.resources`, and native
+    * sidecars, which the scheduler adds to the app containers, so a pod using one is undercharged.
     */
   def fromNodesAndPods(
       nodes: Iterable[Node],
@@ -90,16 +73,14 @@ object ClusterCapacity {
     )
   }
 
-  // A cordoned node takes no new pods, and neither does one that is not Ready (the node
-  // controller taints it for exactly that reason), so neither can hold a new computing unit.
+  // A cordoned or not Ready node takes no new pods, so it cannot hold a new unit.
   private def isSchedulable(node: Node): Boolean = {
     val cordoned =
       Option(node.getSpec).flatMap(spec => Option(spec.getUnschedulable)).exists(_.booleanValue)
     !cordoned && Readiness.isNodeReady(node)
   }
 
-  // The scheduler's accounting: a pod bound to a node counts until it finishes, whether or not its
-  // containers have started, and a pod it has not bound yet counts nowhere.
+  // As the scheduler does, a bound pod holds its requests until it finishes, an unbound one none.
   private def occupiedRequests(pod: Pod, nodeNames: Set[String]): Option[Amounts] =
     for {
       spec <- Option(pod.getSpec)
@@ -107,8 +88,8 @@ object ClusterCapacity {
       if !Option(pod.getStatus).flatMap(s => Option(s.getPhase)).exists(FinishedPhases.contains)
     } yield podRequests(spec)
 
-  // Init containers run one at a time before the app containers, so a pod needs the larger of
-  // the app containers' sum and its biggest init container, plus the runtime-class overhead.
+  // Init containers run one at a time, so a pod needs the larger of its containers' sum and its
+  // biggest init container, plus overhead.
   private def podRequests(spec: PodSpec): Amounts = {
     val appContainers = sum(containers(spec.getContainers).map(requests))
     val peakInit = peak(containers(spec.getInitContainers).map(requests))
@@ -131,8 +112,6 @@ object ClusterCapacity {
       }.toMap)
       .getOrElse(Map.empty)
 
-  // Both treat a resource absent from an amount as zero. Requests are never negative, so zero is
-  // also the identity for `peak`'s max.
   private def sum(all: Iterable[Amounts]): Amounts =
     all.flatten.groupMapReduce(_._1)(_._2)(_ + _)
 
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 a6ffc3f29a0..92d7fc70651 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
@@ -279,9 +279,8 @@ object ComputingUnitHelpers {
   ): Map[String, Map[String, String]] =
     if (units.exists(isKubernetes)) k8s.getAllPodMetrics else Map.empty
 
-  // Same seam as podSnapshotsFor: the public overload binds the production singleton, passed
-  // by-name so a deployment without Kubernetes units is refused without ever forcing the
-  // singleton, which builds a fabric8 client, and tests drive the private overload with a stub.
+  // Same seam as podSnapshotsFor: by-name, so a deployment without Kubernetes units never forces
+  // the singleton, and tests pass a stub to the private overload.
   def clusterCapacity(): ClusterCapacity =
     clusterCapacity(KubernetesConfig.kubernetesComputingUnitEnabled, KubernetesClient)
 
@@ -294,8 +293,7 @@ object ComputingUnitHelpers {
     try {
       k8s.getClusterCapacity
     } catch {
-      // Keeps the cluster's own message: a chart without the cluster-wide read grant fails with
-      // a "forbidden" that names the missing permission, which is what an admin has to act on.
+      // Keeps the cluster's own message, so a missing read grant shows as "forbidden".
       case e: KubernetesClientException =>
         throw ComputingUnitManagingServiceException.fromKubernetes(e)
     }
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 e3a680ccb0d..13b44b11dbf 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
@@ -101,11 +101,9 @@ class KubernetesClient(
     metricsByPodName(fetchPodMetricsItems())
 
   /**
-    * CPU/memory/GPU capacity of the cluster's schedulable nodes. Cluster-scoped, unlike the
-    * namespace-wide reads above, so it needs the chart's cluster-capacity ClusterRole. Finished
-    * pods are excluded by the API server so a cluster that keeps many of them does not ship them
-    * all; [[ClusterCapacity]] still skips them, which is the rule and is tested there, so the
-    * selectors are an optimisation only.
+    * Needs the chart's cluster-capacity ClusterRole, since nodes and every namespace's pods are
+    * cluster-scoped. The API server drops finished pods so they are not shipped, and
+    * `ClusterCapacity` skips them too.
     */
   def getClusterCapacity: ClusterCapacity =
     ClusterCapacity.fromNodesAndPods(
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
index 77087c2ec5b..a9b4ef2a0fc 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
@@ -160,9 +160,9 @@ class AdminComputingUnitResourceSpec
     result.head.ownerName shouldBe "owner"
   }
 
-  // The test environment has kubernetes.enabled false (a local deployment), so the endpoint
-  // refuses; the enabled path is pinned in ComputingUnitHelpersSpec. Cancelled when an exported
-  // KUBERNETES_COMPUTING_UNIT_ENABLED would reach the real client instead.
+  // kubernetes.enabled is false here, so the endpoint refuses; the enabled path is in
+  // ComputingUnitHelpersSpec. Cancelled if an exported KUBERNETES_COMPUTING_UNIT_ENABLED would
+  // reach the real client.
   "getClusterCapacity" should "answer 404 on a deployment without Kubernetes computing units" in {
     assume(
       !KubernetesConfig.kubernetesComputingUnitEnabled,
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
index b7c06285b95..a6ecd94641b 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
@@ -36,10 +36,8 @@ import org.scalatest.matchers.should.Matchers
 
 import scala.jdk.CollectionConverters._
 
-// The aggregation is a pure function over fabric8 model objects, so every case is driven with
-// builder-constructed nodes and pods: no client, no cluster. The rules under test are the ones
-// kube-scheduler applies when it decides whether a new pod fits, because that is the question an
-// admin is asking when a computing unit refuses to start.
+// A pure function over fabric8 objects, so cases use builders with no client. The rules are the
+// ones kube-scheduler applies when it decides whether a new pod fits.
 class ClusterCapacitySpec extends AnyFlatSpec with Matchers {
 
   import ClusterCapacitySpec._
@@ -92,8 +90,7 @@ class ClusterCapacitySpec extends AnyFlatSpec with Matchers {
   }
 
   it should "count a container that declares no resources, or empty ones, as zero" in {
-    // System pods on a shared cluster are not always written with resources, and a container
-    // with no `resources` block at all (null) must not take the whole endpoint down.
+    // A container with no `resources` block (null) must not take the endpoint down.
     val bare = new PodBuilder()
       .withNewMetadata()
       .withName("bare")
@@ -116,8 +113,7 @@ class ClusterCapacitySpec extends AnyFlatSpec with Matchers {
     val result = capacity(
       Seq(node("a", "8", "16Gi")),
       Seq(
-        // Init containers run one at a time before the app containers, so the pod needs
-        // max(sum(containers), max(init)): the init container wins on CPU, the containers on memory.
+        // The init container wins on CPU, the app containers on memory: the pod needs the larger.
         pod(
           "p",
           Some("a"),
@@ -232,9 +228,7 @@ class ClusterCapacitySpec extends AnyFlatSpec with Matchers {
     result.memory shouldBe ResourceCapacity(total = 4 * Gi, allocated = 6 * Gi, available = 0)
   }
 
-  // Consumers read these exact field names, so a rename is a breaking API change. Serialised the
-  // way the service does: Dropwizard's mapper with the Scala module registered in
-  // ComputingUnitManagingService.initialize.
+  // Consumers read these field names, so a rename breaks them. Mapper set up as the service does.
   "ClusterCapacity" should "serialise as cpu/memory/gpu objects of total, allocated and available" in {
     val mapper = Jackson.newObjectMapper().registerModule(DefaultScalaModule)
     val serialised = mapper.writeValueAsString(
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 ea482f6ed26..aa1d2ea00df 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
@@ -804,8 +804,7 @@ class ComputingUnitHelpersSpec
 
   // ── clusterCapacity guard ────────────────────────────────────────────
 
-  // Same by-name seam as the two guards above. Passing `fail(...)` as the client proves the
-  // disabled branch never forces it.
+  // Passing `fail(...)` as the client proves the disabled branch never forces it.
   "clusterCapacity" should "refuse, without touching the cluster, when kubernetes units are disabled" in {
     val refused = intercept[NotFoundException] {
       ComputingUnitHelpers.clusterCapacity(enabled = false, fail("the client must not be forced"))
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 e5a3a36a8f0..60e9a653850 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
@@ -298,10 +298,8 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
   }
 
   // ── cluster-wide capacity ──
-  // Reads cluster-scoped nodes and the pods of every namespace, narrowed on the server to the
-  // pods that have not finished. The stub answers only that chain: a wrapper that listed just the
-  // pool namespace sees no pods and one that skipped the selectors gets null back, and either
-  // fails; the arithmetic itself is pinned by ClusterCapacitySpec.
+  // The stub answers only the all-namespace, unfinished-pod chain, so a wrapper that lists one
+  // namespace or skips the selectors fails; the arithmetic is pinned by ClusterCapacitySpec.
 
   private def clusterClient(nodes: Seq[Node], pods: Seq[Pod]): Fabric8Client = {
     val client = stubbedClient(Seq.empty, Seq.empty)

From 07c2e8faf1d0699781217dce1631d6910b8d950b Mon Sep 17 00:00:00 2001
From: Kunwoo Park <kunwp1@uci.edu>
Date: Sun, 25 Oct 2026 00:16:40 -0700
Subject: [PATCH 3/3] fix(computing-unit): include the release namespace in
 cluster capacity RBAC names

ClusterRole and ClusterRoleBinding names are cluster-wide, but Helm release
names are only unique per namespace, so two same-name releases in different
namespaces rendered the same names and the second install failed.
---
 .../workflow-computing-unit-manager-service-account.yaml  | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
index c8bfb0ccba3..fe311644539 100644
--- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
+++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-service-account.yaml
@@ -67,11 +67,11 @@ roleRef:
 ---
 # Read-only and cluster-wide for the admin cluster-capacity endpoint: nodes are cluster-scoped and
 # every namespace's pods take their capacity, so the namespaced Role above cannot cover it.
-# Named for the release because a ClusterRole is cluster-scoped.
+# Named for the release and its namespace because a ClusterRole is cluster-scoped.
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
-  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+  name: {{ .Release.Name }}-{{ .Release.Namespace }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
 rules:
   - apiGroups: [""]
     resources: ["nodes", "pods"]
@@ -81,12 +81,12 @@ rules:
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
-  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+  name: {{ .Release.Name }}-{{ .Release.Namespace }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
 subjects:
   - kind: ServiceAccount
     name: {{ .Values.workflowComputingUnitManager.serviceAccountName }}
     namespace: {{ .Release.Namespace }}
 roleRef:
   kind: ClusterRole
-  name: {{ .Release.Name }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+  name: {{ .Release.Name }}-{{ .Release.Namespace }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
   apiGroup: rbac.authorization.k8s.io