feat(computing-unit): add admin cluster capacity endpoint

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

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()
 }

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
+}

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,

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()

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")
+  }
 }