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..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
@@ -63,3 +63,30 @@ roleRef:
   kind: Role
   name: {{ .Values.workflowComputingUnitManager.name }}
   apiGroup: rbac.authorization.k8s.io
+
+---
+# 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 and its namespace because a ClusterRole is cluster-scoped.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ .Release.Name }}-{{ .Release.Namespace }}-{{ .Values.workflowComputingUnitManager.name }}-cluster-capacity
+rules:
+  - apiGroups: [""]
+    resources: ["nodes", "pods"]
+    verbs: ["list"]
+
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  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 }}-{{ .Release.Namespace }}-{{ .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..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
@@ -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,13 @@ class AdminComputingUnitResource {
       )
     }
   }
+
+  /**
+    * 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")
+  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..a8639304cb1
--- /dev/null
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ClusterCapacity.scala
@@ -0,0 +1,125 @@
+/*
+ * 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._
+
+/** 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)
+
+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
+
+  // Pods in these phases hold nothing on their node.
+  private val FinishedPhases = Set("Succeeded", "Failed")
+
+  /**
+    * 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.
+    *
+    * 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],
+      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 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)
+  }
+
+  // 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)
+      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, 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))
+    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)
+
+  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..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
@@ -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,26 @@ object ComputingUnitHelpers {
   ): Map[String, Map[String, String]] =
     if (units.exists(isKubernetes)) k8s.getAllPodMetrics else Map.empty
 
+  // 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)
+
+  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, so a missing read grant shows as "forbidden".
+      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..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
@@ -100,6 +100,25 @@ class KubernetesClient(
   def getAllPodMetrics: Map[String, Map[String, String]] =
     metricsByPodName(fetchPodMetricsItems())
 
+  /**
+    * 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(
+      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..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
@@ -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"
   }
+
+  // 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,
+      "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..a6ecd94641b
--- /dev/null
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ClusterCapacitySpec.scala
@@ -0,0 +1,315 @@
+/*
+ * 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._
+
+// 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._
+
+  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 {
+    // A container with no `resources` block (null) must not take the 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(
+        // The init container wins on CPU, the app containers on memory: the pod needs the larger.
+        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 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(
+      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..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
@@ -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,38 @@ class ComputingUnitHelpersSpec
     ComputingUnitHelpers.podMetricsFor(List(kubernetesUnit(51)), k8s) shouldBe metrics
   }
 
+  // ── clusterCapacity guard ────────────────────────────────────────────
+
+  // 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..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
@@ -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,43 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
     k8s.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi")
     k8s.getPodMetrics(999) shouldBe empty
   }
+
+  // ── cluster-wide capacity ──
+  // 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)
+
+    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