feat(computing-unit-managing-service): add admin endpoint to list all computing units

build.sbt

@@ -161,7 +161,12 @@
 lazy val ComputingUnitManagingService = (project in file("computing-unit-managing-service"))
   .dependsOn(WorkflowCore, Auth, Config, Resource)
   .settings(commonModuleSettings)
+  .configs(Test)
+  .dependsOn(DAO % "test->test", Auth % "test->test") // reuse MockTexeraDB embedded Postgres in tests
   .settings(
+    // Tests mutate JVM-wide singletons (SqlServer via MockTexeraDB, the
+    // KubernetesClient fabric8 client), so run suites serially to avoid cross-suite races.
+    Test / parallelExecution := false,
     dependencyOverrides ++= Seq(
       // override it as io.dropwizard 4 require 2.16.1 or higher
       "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion,
@@ -199,12 +204,12 @@
 
 lazy val WorkflowOperator = (project in file("common/workflow-operator")).settings(commonModuleSettingsWithVendored).dependsOn(WorkflowCore)
 lazy val WorkflowCompilingService = (project in file("workflow-compiling-service"))
   .dependsOn(WorkflowOperator, Auth, Config, Resource)
   .settings(commonModuleSettings)
   .settings(
     dependencyOverrides ++= Seq(
       // override it as io.dropwizard 4 require 2.16.1 or higher
       "com.fasterxml.jackson.module" %% "jackson-module-scala" % jacksonVersion,
       "com.fasterxml.jackson.core" % "jackson-databind" % jacksonVersion,
       "org.glassfish.jersey.core" % "jersey-common" % "3.0.12"
     ) ++ nettyDependencyOverrides

computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala

@@ -27,6 +27,7 @@ import org.apache.texera.common.config.StorageConfig
 import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer}
 import org.apache.texera.dao.SqlServer
 import org.apache.texera.service.resource.{
+  AdminComputingUnitResource,
   ComputingUnitAccessResource,
   ComputingUnitManagingResource,
   HealthCheckResource
@@ -66,6 +67,7 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ
 
     environment.jersey().register(new ComputingUnitManagingResource)
     environment.jersey().register(new ComputingUnitAccessResource)
+    environment.jersey().register(new AdminComputingUnitResource)
 
     RoleAnnotationEnforcer.enforce(
       environment.jersey.getResourceConfig,

computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.service.resource
+
+import io.dropwizard.auth.Auth
+import jakarta.annotation.security.RolesAllowed
+import jakarta.ws.rs.{GET, Path, Produces}
+import jakarta.ws.rs.core.MediaType
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_COMPUTING_UNIT
+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.jooq.DSLContext
+
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+object AdminComputingUnitResource {
+  private def context: DSLContext =
+    SqlServer
+      .getInstance()
+      .createDSLContext()
+
+  /**
+    * Render already-active `units` as admin dashboard rows (pure — no filtering). Every row gets
+    * WRITE access (an admin controls every unit it sees); `isOwner` reflects `callerUid`.
+    */
+  def buildDashboardUnits(
+      units: List[WorkflowComputingUnit],
+      ownerInfo: Map[Integer, (String, String)],
+      callerUid: Integer,
+      podPhases: Map[String, String],
+      podMetrics: Map[String, Map[String, String]]
+  ): List[DashboardWorkflowComputingUnit] =
+    units.map { unit =>
+      ComputingUnitHelpers.buildDashboardUnit(
+        unit,
+        isOwner = unit.getUid.equals(callerUid),
+        accessPrivilege = PrivilegeEnum.WRITE,
+        ownerInfo = ownerInfo,
+        podPhases = podPhases,
+        podMetrics = podMetrics
+      )
+    }
+}
+
+@Produces(Array(MediaType.APPLICATION_JSON))
+@Path("/computing-unit/admin")
+@RolesAllowed(Array("ADMIN"))
+class AdminComputingUnitResource {
+
+  import AdminComputingUnitResource._
+
+  /**
+    * List every non-terminated computing unit across all users (ADMIN-only). Like the per-user
+    * endpoint, a Kubernetes unit whose pod has vanished is marked terminated and dropped, so ghost
+    * units don't accumulate. Status/metrics use one namespace-wide call each (O(1) round trips).
+    */
+  @GET
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @Path("/list")
+  def listAllComputingUnits(
+      @Auth user: SessionUser
+  ): List[DashboardWorkflowComputingUnit] = {
+    val ctx = context
+
+    // Filter to active units in SQL so terminated rows are never loaded.
+    val activeUnits: List[WorkflowComputingUnit] =
+      ctx
+        .selectFrom(WORKFLOW_COMPUTING_UNIT)
+        .where(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME.isNull)
+        .fetchInto(classOf[WorkflowComputingUnit])
+        .asScala
+        .toList
+
+    // Pod phases decide which Kubernetes units are still alive.
+    val podPhases = ComputingUnitHelpers.podPhasesFor(activeUnits)
+
+    val liveUnits = ComputingUnitHelpers.reconcileVanishedKubernetesUnits(
+      new WorkflowComputingUnitDao(ctx.configuration()),
+      activeUnits,
+      podPhases
+    )
+
+    // Metrics only for survivors, so fetch after reconciliation.
+    val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits)
+
+    val userDao = new UserDao(ctx.configuration())
+    val ownerInfo = ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct)
+
+    buildDashboardUnits(liveUnits, ownerInfo, user.getUid, podPhases, podMetrics)
+  }
+}

computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala

@@ -50,8 +50,8 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
 }
 import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit
 import org.apache.texera.service.resource.ComputingUnitManagingResource._
-import org.apache.texera.service.resource.ComputingUnitState._
 import org.apache.texera.service.util.{
+  ComputingUnitHelpers,
   ComputingUnitManagingServiceException,
   InsufficientComputingUnitQuota,
   KubernetesClient
@@ -197,41 +197,6 @@ class ComputingUnitManagingResource {
     }
   }
 
-  private def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = {
-    unit.getType match {
-      // ── Local CUs are always "running" ──────────────────────────────
-      case WorkflowComputingUnitTypeEnum.local =>
-        Running
-
-      // ── Kubernetes CUs – only explicit "Running" counts as running ─
-      case WorkflowComputingUnitTypeEnum.kubernetes =>
-        val phaseOpt = KubernetesClient
-          .getPodByName(KubernetesClient.generatePodName(unit.getCuid))
-          .map(_.getStatus.getPhase)
-
-        if (phaseOpt.contains("Running")) Running else Pending
-
-      // ── Any other (unknown) type is treated as pending ──────────────
-      case _ =>
-        Pending
-    }
-  }
-
-  private def getComputingUnitMetrics(unit: WorkflowComputingUnit): WorkflowComputingUnitMetrics = {
-    unit.getType match {
-      case WorkflowComputingUnitTypeEnum.local =>
-        WorkflowComputingUnitMetrics("NaN", "NaN")
-      case WorkflowComputingUnitTypeEnum.kubernetes =>
-        val metrics = KubernetesClient.getPodMetrics(unit.getCuid)
-        WorkflowComputingUnitMetrics(
-          metrics.getOrElse("cpu", ""),
-          metrics.getOrElse("memory", "")
-        )
-      case _ =>
-        WorkflowComputingUnitMetrics("NaN", "NaN")
-    }
-  }
-
   private def getComputingUnitResourceLimit(
       unit: WorkflowComputingUnit
   ): WorkflowComputingUnitResourceLimit = {
@@ -476,8 +441,8 @@ class ComputingUnitManagingResource {
 
       DashboardWorkflowComputingUnit(
         insertedUnit,
-        getComputingUnitStatus(insertedUnit).toString,
-        getComputingUnitMetrics(insertedUnit),
+        ComputingUnitHelpers.getComputingUnitStatus(insertedUnit).toString,
+        ComputingUnitHelpers.getComputingUnitMetrics(insertedUnit),
         isOwner = true,
         accessPrivilege = PrivilegeEnum.WRITE,
         ownerGoogleAvatar,
@@ -528,63 +493,46 @@ class ComputingUnitManagingResource {
           (List.empty[WorkflowComputingUnit], Map.empty[Integer, PrivilegeEnum])
         }
 
-      val allUnits = ownedUnits ++ sharedUnits
-      val ownerUids: List[Integer] = allUnits.map(_.getUid).distinct
       val userDao = new UserDao(ctx.configuration())
-      val ownerInfoMap: Map[Integer, (String, String)] =
-        userDao
-          .fetchByUid(ownerUids: _*)
-          .asScala
-          .map { u =>
-            val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull
-            val name = Option(u.getName).filter(_.nonEmpty).orNull
-            u.getUid -> (avatar, name)
-          }
-          .toMap
-
-      // If a Kubernetes pod has already disappeared (e.g., manually deleted or TTL
-      // GC-ed by the cluster), we treat the corresponding computing unit as
-      // terminated from the system's point of view. Here we eagerly update its
-      // terminateTime in the database **before** we build the response list so
-      // that subsequent API calls will no longer return this unit.
-      allUnits.foreach { unit =>
-        if (
-          unit.getType == WorkflowComputingUnitTypeEnum.kubernetes &&
-          !KubernetesClient.podExists(unit.getCuid)
-        ) {
-          unit.setTerminateTime(new Timestamp(System.currentTimeMillis()))
-          computingUnitDao.update(unit)
-        }
-      }
 
-      // For shared units, we need to check the access privilege which are saved in different table
-      // to streamline the process, we combine owned units with default WRITE privilege and use sharedUnitInfo
-      // to get the privilege for shared units.
-      (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++ sharedUnits.map(u =>
-        (u, sharedUnitInfo(u.getCuid))
-      ))
-        .distinctBy { case (unit, _) => unit.getCuid }
-        .filter { case (unit, _) => unit.getTerminateTime == null }
-        .filter {
-          case (unit, _) =>
-            unit.getType match {
-              case WorkflowComputingUnitTypeEnum.kubernetes =>
-                KubernetesClient.podExists(unit.getCuid)
-              case _ => true
-            }
-        }
-        .map {
-          case (unit, privilege) =>
-            DashboardWorkflowComputingUnit(
-              computingUnit = unit,
-              isOwner = unit.getUid.equals(uid),
-              accessPrivilege = privilege,
-              status = getComputingUnitStatus(unit).toString,
-              metrics = getComputingUnitMetrics(unit),
-              ownerGoogleAvatar = ownerInfoMap.getOrElse(unit.getUid, (null, null))._1,
-              ownerName = ownerInfoMap.getOrElse(unit.getUid, (null, null))._2
-            )
-        }
+      // Pair each unit with the caller's privilege (owned default to WRITE), one row per cuid, so
+      // a unit that is both owned and shared is reconciled/rendered exactly once.
+      val unitsWithPrivilege =
+        (ownedUnits.map(u => (u, PrivilegeEnum.WRITE)) ++
+          sharedUnits.map(u => (u, sharedUnitInfo(u.getCuid))))
+          .distinctBy { case (unit, _) => unit.getCuid }
+          .filter { case (unit, _) => unit.getTerminateTime == null }
+      val privilegeByCuid = unitsWithPrivilege.map {
+        case (unit, privilege) => unit.getCuid -> privilege
+      }.toMap
+      val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit }
+
+      // Pod phases decide which Kubernetes units are still alive.
+      val podPhases = ComputingUnitHelpers.podPhasesFor(candidateUnits)
+
+      val liveUnits =
+        ComputingUnitHelpers.reconcileVanishedKubernetesUnits(
+          computingUnitDao,
+          candidateUnits,
+          podPhases
+        )
+
+      // Metrics only for survivors, so fetch after reconciliation.
+      val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits)
+
+      val ownerInfoMap =
+        ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct)
+
+      liveUnits.map { unit =>
+        ComputingUnitHelpers.buildDashboardUnit(
+          unit,
+          isOwner = unit.getUid.equals(uid),
+          accessPrivilege = privilegeByCuid(unit.getCuid),
+          ownerInfo = ownerInfoMap,
+          podPhases = podPhases,
+          podMetrics = podMetrics
+        )
+      }
     }
   }
 
@@ -613,8 +561,8 @@ class ComputingUnitManagingResource {
 
     DashboardWorkflowComputingUnit(
       computingUnit = unit,
-      status = getComputingUnitStatus(unit).toString,
-      metrics = getComputingUnitMetrics(unit),
+      status = ComputingUnitHelpers.getComputingUnitStatus(unit).toString,
+      metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit),
       isOwner = unit.getUid.equals(user.getUid),
       accessPrivilege = {
         val cuAccessDao = new ComputingUnitUserAccessDao(context.configuration())
@@ -748,7 +696,7 @@ class ComputingUnitManagingResource {
       throw new BadRequestException("User has no access to the computing unit")
     }
     val computingUnit = getComputingUnitByCuid(context, cuid.toInt)
-    getComputingUnitMetrics(computingUnit)
+    ComputingUnitHelpers.getComputingUnitMetrics(computingUnit)
   }
 
   @GET