From 885b3fe56ab66a26889ad4e16606fdcba0e7d45a Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Thu, 16 Jul 2026 19:55:31 -0400 Subject: [PATCH 1/8] feat(computing-unit-managing-service): add admin endpoint to list all computing units Add an ADMIN-only GET /computing-unit/admin/list that returns every non-terminated computing unit across all users, backing the admin Computing Units dashboard. Row assembly is extracted into a pure buildDashboardUnits function so it is unit-testable without a database or Kubernetes. Register the resource and extend the service's startup access-control coverage check. --- .../ComputingUnitManagingService.scala | 2 + .../resource/AdminComputingUnitResource.scala | 119 ++++++++++++++++++ .../ComputingUnitManagingServiceRunSpec.scala | 2 + .../AdminComputingUnitResourceSpec.scala | 83 ++++++++++++ 4 files changed, 206 insertions(+) create mode 100644 computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala create mode 100644 computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala index db63bbf2eb2..f0dffc89e11 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala +++ b/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, 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 new file mode 100644 index 00000000000..92f46c409cb --- /dev/null +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala @@ -0,0 +1,119 @@ +/* + * 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.SqlServer.withTransaction +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() + + /** + * Assemble the dashboard rows for the admin view from the raw computing units and their + * owners' display info. Terminated units (non-null `terminate_time`) are excluded, and the + * remaining units keep the same [[DashboardWorkflowComputingUnit]] shape produced by the + * per-user listing endpoint. + * + * @param units all computing units to consider (across every owner) + * @param ownerInfo map of owner uid -> (googleAvatar, userName) + * @param callerUid the uid of the requesting admin, used to populate `isOwner` + */ + def buildDashboardUnits( + units: List[WorkflowComputingUnit], + ownerInfo: Map[Integer, (String, String)], + callerUid: Integer + ): List[DashboardWorkflowComputingUnit] = { + units + .filter(_.getTerminateTime == null) + .map { unit => + val (avatar, name) = ownerInfo.getOrElse(unit.getUid, (null, null)) + DashboardWorkflowComputingUnit( + computingUnit = unit, + status = ComputingUnitHelpers.getComputingUnitStatus(unit).toString, + metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit), + isOwner = unit.getUid.equals(callerUid), + // Admins have full control over every computing unit they can see. + accessPrivilege = PrivilegeEnum.WRITE, + ownerGoogleAvatar = avatar, + ownerName = name + ) + } + } +} + +@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. + * + * @return the computing units (owned by any user) that have not been terminated. + */ + @GET + @Produces(Array(MediaType.APPLICATION_JSON)) + @Path("/list") + def listAllComputingUnits( + @Auth user: SessionUser + ): List[DashboardWorkflowComputingUnit] = { + withTransaction(context) { ctx => + val computingUnitDao = new WorkflowComputingUnitDao(ctx.configuration()) + val userDao = new UserDao(ctx.configuration()) + + val activeUnits = + computingUnitDao.findAll().asScala.toList.filter(_.getTerminateTime == null) + + val ownerUids: List[Integer] = activeUnits.map(_.getUid).distinct + val ownerInfo: Map[Integer, (String, String)] = + if (ownerUids.isEmpty) Map.empty + else + 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 + + buildDashboardUnits(activeUnits, ownerInfo, user.getUid) + } + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala index d2162d48c77..6962c2dc0d5 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala @@ -21,6 +21,7 @@ package org.apache.texera.service import org.apache.texera.auth.RoleAnnotationEnforcer import org.apache.texera.service.resource.{ + AdminComputingUnitResource, ComputingUnitAccessResource, ComputingUnitManagingResource, HealthCheckResource @@ -36,6 +37,7 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { Seq( classOf[ComputingUnitManagingResource], classOf[ComputingUnitAccessResource], + classOf[AdminComputingUnitResource], classOf[HealthCheckResource] ) ) shouldBe empty 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 new file mode 100644 index 00000000000..af145455050 --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala @@ -0,0 +1,83 @@ +/* + * 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 jakarta.annotation.security.RolesAllowed +import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} +import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.sql.Timestamp + +class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { + + private def localUnit( + cuid: Int, + uid: Int, + name: String, + terminated: Boolean = false + ): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit() + unit.setCuid(cuid) + unit.setUid(uid) + unit.setName(name) + unit.setType(WorkflowComputingUnitTypeEnum.local) + if (terminated) unit.setTerminateTime(new Timestamp(0L)) + unit + } + + // The class-level @RolesAllowed(Array("ADMIN")) is what makes Jersey's + // RolesAllowedDynamicFeature return 403 for any non-ADMIN (e.g. REGULAR) caller. + "AdminComputingUnitResource" should "only permit the ADMIN role (non-ADMIN callers are rejected)" in { + val annotation = classOf[AdminComputingUnitResource].getAnnotation(classOf[RolesAllowed]) + annotation should not be null + annotation.value.toSeq shouldBe Seq("ADMIN") + } + + "buildDashboardUnits" should "return active units owned by multiple users, excluding terminated ones" in { + val units = List( + localUnit(cuid = 1, uid = 100, name = "alice-cu"), + localUnit(cuid = 2, uid = 200, name = "bob-cu"), + localUnit(cuid = 3, uid = 100, name = "alice-terminated", terminated = true) + ) + val ownerInfo: Map[Integer, (String, String)] = Map( + (100: Integer) -> ("alice-avatar", "alice"), + (200: Integer) -> (null, "bob") + ) + + // caller is an admin who owns none of these units + val result = AdminComputingUnitResource.buildDashboardUnits(units, ownerInfo, callerUid = 999) + + result.map(_.computingUnit.getCuid) should contain theSameElementsAs Seq(1, 2) + result.map(_.computingUnit.getUid).distinct should contain theSameElementsAs Seq(100, 200) + + val byCuid = result.map(u => u.computingUnit.getCuid.intValue() -> u).toMap + byCuid(1).ownerName shouldBe "alice" + byCuid(1).ownerGoogleAvatar shouldBe "alice-avatar" + byCuid(2).ownerName shouldBe "bob" + byCuid(2).ownerGoogleAvatar shouldBe null + + // local units report Running/NaN and the admin caller does not own them + all(result.map(_.status)) shouldBe "Running" + all(result.map(_.isOwner)) shouldBe false + all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE + } +} From cb566cd6208c5cb0e9bb3ed0ecf64c85cb5e27d7 Mon Sep 17 00:00:00 2001 From: kunwp1 Date: Fri, 17 Jul 2026 00:20:20 -0400 Subject: [PATCH 2/8] refactor(computing-unit-managing-service): share CU listing helpers and batch k8s calls Centralize computing-unit listing logic in ComputingUnitHelpers so the per-user and admin listing endpoints share one implementation: - Add resolveOwnerInfo, partitionLiveUnits, reconcileVanishedKubernetesUnits, and buildDashboardUnit; both listing endpoints now route through them. - Remove the byte-identical private getComputingUnitStatus/getComputingUnitMetrics from ComputingUnitManagingResource; single-unit callers use the shared helpers. - Add bulk KubernetesClient.getAllPodPhases / getAllPodMetrics (one namespace-wide list()/top() each) and a shared containerUsage helper; listComputingUnits now resolves pod status/metrics in bulk instead of per-unit (previously podExists was probed twice per unit and top() was re-fetched per unit). - Admin endpoint filters active units in SQL and reconciles vanished pods via the shared helper. Behavior of the API responses is unchanged; per-user reconciliation now runs over cuid-deduplicated units, reducing redundant DB updates without adding new ones. --- .../resource/AdminComputingUnitResource.scala | 118 ++++++++------ .../ComputingUnitManagingResource.scala | 150 +++++++----------- .../service/util/ComputingUnitHelpers.scala | 147 ++++++++++++++++- .../service/util/KubernetesClient.scala | 63 ++++++-- .../AdminComputingUnitResourceSpec.scala | 111 +++++++++++-- .../util/ComputingUnitHelpersSpec.scala | 120 +++++++++++++- 6 files changed, 538 insertions(+), 171 deletions(-) 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 92f46c409cb..8c866891b93 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 @@ -25,12 +25,12 @@ 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.SqlServer.withTransaction -import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum +import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_COMPUTING_UNIT +import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import org.apache.texera.service.resource.ComputingUnitManagingResource.DashboardWorkflowComputingUnit -import org.apache.texera.service.util.ComputingUnitHelpers +import org.apache.texera.service.util.{ComputingUnitHelpers, KubernetesClient} import org.jooq.DSLContext import scala.jdk.CollectionConverters.CollectionHasAsScala @@ -42,36 +42,35 @@ object AdminComputingUnitResource { .createDSLContext() /** - * Assemble the dashboard rows for the admin view from the raw computing units and their - * owners' display info. Terminated units (non-null `terminate_time`) are excluded, and the - * remaining units keep the same [[DashboardWorkflowComputingUnit]] shape produced by the - * per-user listing endpoint. + * Assemble the admin dashboard rows for a set of active computing units. The `units` are + * expected to already be the active (non-terminated, pod-still-present) set; this is a pure + * mapping and does no filtering of its own. Every row is marked with WRITE access — admins + * have full control over every unit they can see — and `isOwner` reflects the requesting + * admin. Kubernetes status/metrics are resolved from the pre-fetched maps (no per-unit call). * - * @param units all computing units to consider (across every owner) - * @param ownerInfo map of owner uid -> (googleAvatar, userName) - * @param callerUid the uid of the requesting admin, used to populate `isOwner` + * @param units active computing units to render (across every owner) + * @param ownerInfo map of owner uid -> (googleAvatar, userName) + * @param callerUid the uid of the requesting admin, used to populate `isOwner` + * @param podPhases map of pod name -> phase (see [[KubernetesClient.getAllPodPhases]]) + * @param podMetrics map of pod name -> (metric -> value) (see KubernetesClient.getAllPodMetrics) */ def buildDashboardUnits( units: List[WorkflowComputingUnit], ownerInfo: Map[Integer, (String, String)], - callerUid: Integer - ): List[DashboardWorkflowComputingUnit] = { - units - .filter(_.getTerminateTime == null) - .map { unit => - val (avatar, name) = ownerInfo.getOrElse(unit.getUid, (null, null)) - DashboardWorkflowComputingUnit( - computingUnit = unit, - status = ComputingUnitHelpers.getComputingUnitStatus(unit).toString, - metrics = ComputingUnitHelpers.getComputingUnitMetrics(unit), - isOwner = unit.getUid.equals(callerUid), - // Admins have full control over every computing unit they can see. - accessPrivilege = PrivilegeEnum.WRITE, - ownerGoogleAvatar = avatar, - ownerName = name - ) - } - } + 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)) @@ -84,7 +83,16 @@ class AdminComputingUnitResource { /** * List every non-terminated computing unit across all users. ADMIN-only. * - * @return the computing units (owned by any user) that have not been terminated. + * Mirrors the reconciliation done by the per-user listing endpoint: a Kubernetes unit whose + * pod has vanished (manually deleted or TTL GC-ed by the cluster) is eagerly marked + * terminated in the database and excluded from the response, so ghost units do not + * accumulate in the admin view. + * + * Kubernetes status/metrics are resolved from a single namespace-wide `list`/`top` call each, + * so the number of cluster round trips is constant rather than proportional to the number of + * units. + * + * @return the computing units (owned by any user) that are active and whose pods still exist. */ @GET @Produces(Array(MediaType.APPLICATION_JSON)) @@ -92,28 +100,40 @@ class AdminComputingUnitResource { def listAllComputingUnits( @Auth user: SessionUser ): List[DashboardWorkflowComputingUnit] = { - withTransaction(context) { ctx => - val computingUnitDao = new WorkflowComputingUnitDao(ctx.configuration()) - val userDao = new UserDao(ctx.configuration()) + val ctx = context - val activeUnits = - computingUnitDao.findAll().asScala.toList.filter(_.getTerminateTime == null) + def isKubernetes(unit: WorkflowComputingUnit): Boolean = + unit.getType == WorkflowComputingUnitTypeEnum.kubernetes - val ownerUids: List[Integer] = activeUnits.map(_.getUid).distinct - val ownerInfo: Map[Integer, (String, String)] = - if (ownerUids.isEmpty) Map.empty - else - 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 + // Filter to active units in SQL so historically-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 - buildDashboardUnits(activeUnits, ownerInfo, user.getUid) - } + // Pod phases (one namespace-wide `list`) are needed to decide which units are still alive; + // only fetch them when there is a Kubernetes unit to resolve. + val podPhases: Map[String, String] = + if (activeUnits.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty + + // A Kubernetes unit whose pod is gone is stamped terminated and dropped from the response. + val liveUnits = ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + new WorkflowComputingUnitDao(ctx.configuration()), + activeUnits, + podPhases + ) + + // Metrics (one namespace-wide `top`) are only rendered for surviving units, so defer this + // call until after reconciliation and skip it when no live Kubernetes unit remains. + val podMetrics: Map[String, Map[String, String]] = + if (liveUnits.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty + + val userDao = new UserDao(ctx.configuration()) + val ownerInfo = ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct) + + buildDashboardUnits(liveUnits, ownerInfo, user.getUid, podPhases, podMetrics) } } diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala index aa02f73387e..d1e5ef3931c 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala @@ -50,8 +50,8 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{ } import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import org.apache.texera.service.resource.ComputingUnitManagingResource._ -import org.apache.texera.service.resource.ComputingUnitState._ import org.apache.texera.service.util.{ + ComputingUnitHelpers, ComputingUnitManagingServiceException, InsufficientComputingUnitQuota, KubernetesClient @@ -197,41 +197,6 @@ class ComputingUnitManagingResource { } } - private def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = { - unit.getType match { - // ── Local CUs are always "running" ────────────────────────────── - case WorkflowComputingUnitTypeEnum.local => - Running - - // ── Kubernetes CUs – only explicit "Running" counts as running ─ - case WorkflowComputingUnitTypeEnum.kubernetes => - val phaseOpt = KubernetesClient - .getPodByName(KubernetesClient.generatePodName(unit.getCuid)) - .map(_.getStatus.getPhase) - - if (phaseOpt.contains("Running")) Running else Pending - - // ── Any other (unknown) type is treated as pending ────────────── - case _ => - Pending - } - } - - private def getComputingUnitMetrics(unit: WorkflowComputingUnit): WorkflowComputingUnitMetrics = { - unit.getType match { - case WorkflowComputingUnitTypeEnum.local => - WorkflowComputingUnitMetrics("NaN", "NaN") - case WorkflowComputingUnitTypeEnum.kubernetes => - val metrics = KubernetesClient.getPodMetrics(unit.getCuid) - WorkflowComputingUnitMetrics( - metrics.getOrElse("cpu", ""), - metrics.getOrElse("memory", "") - ) - case _ => - WorkflowComputingUnitMetrics("NaN", "NaN") - } - } - private def getComputingUnitResourceLimit( unit: WorkflowComputingUnit ): WorkflowComputingUnitResourceLimit = { @@ -476,8 +441,8 @@ class ComputingUnitManagingResource { DashboardWorkflowComputingUnit( insertedUnit, - getComputingUnitStatus(insertedUnit).toString, - getComputingUnitMetrics(insertedUnit), + ComputingUnitHelpers.getComputingUnitStatus(insertedUnit).toString, + ComputingUnitHelpers.getComputingUnitMetrics(insertedUnit), isOwner = true, accessPrivilege = PrivilegeEnum.WRITE, ownerGoogleAvatar, @@ -529,62 +494,57 @@ class ComputingUnitManagingResource { } 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 units default to WRITE), keeping only + // one row per cuid. Do this first so we reconcile/render each unit exactly once even when a + // unit is both owned and shared. + 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 } + + // Pod phases (one namespace-wide `list`) decide which Kubernetes units are still alive; + // only fetch them when there is a Kubernetes unit to resolve. + val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit } + val podPhases: Map[String, String] = + if (candidateUnits.exists(_.getType == WorkflowComputingUnitTypeEnum.kubernetes)) + KubernetesClient.getAllPodPhases + else Map.empty + + // A Kubernetes unit whose pod has vanished (manually deleted or TTL GC-ed by the cluster) + // is treated as terminated: its terminateTime is persisted and it is dropped here so + // subsequent API calls no longer return it. + val liveCuids = ComputingUnitHelpers + .reconcileVanishedKubernetesUnits(computingUnitDao, candidateUnits, podPhases) + .map(_.getCuid) + .toSet + val liveUnitsWithPrivilege = + unitsWithPrivilege.filter { case (unit, _) => liveCuids.contains(unit.getCuid) } + + // Metrics (one namespace-wide `top`) are only rendered for surviving units, so defer this + // call until after reconciliation and skip it when no live Kubernetes unit remains. + val podMetrics: Map[String, Map[String, String]] = + if (liveUnitsWithPrivilege.exists { + case (unit, _) => unit.getType == WorkflowComputingUnitTypeEnum.kubernetes + }) KubernetesClient.getAllPodMetrics + else Map.empty + + val ownerInfoMap = + ComputingUnitHelpers.resolveOwnerInfo(userDao, allUnits.map(_.getUid).distinct) + + liveUnitsWithPrivilege.map { + case (unit, privilege) => + ComputingUnitHelpers.buildDashboardUnit( + unit, + isOwner = unit.getUid.equals(uid), + accessPrivilege = privilege, + ownerInfo = ownerInfoMap, + podPhases = podPhases, + podMetrics = podMetrics + ) + } } } @@ -613,8 +573,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 +708,7 @@ class ComputingUnitManagingResource { throw new BadRequestException("User has no access to the computing unit") } val computingUnit = getComputingUnitByCuid(context, cuid.toInt) - getComputingUnitMetrics(computingUnit) + ComputingUnitHelpers.getComputingUnitMetrics(computingUnit) } @GET diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala index 8ae31bba7fb..067a99b6336 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala @@ -19,11 +19,42 @@ package org.apache.texera.service.util import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit -import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics +import org.apache.texera.service.resource.ComputingUnitManagingResource.{ + DashboardWorkflowComputingUnit, + WorkflowComputingUnitMetrics +} import org.apache.texera.service.resource.ComputingUnitState.{ComputingUnitState, Pending, Running} +import org.jooq.EnumType + +import java.sql.Timestamp +import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava} object ComputingUnitHelpers { + + /** + * Resolve owner display info for the given owner uids in a single `fetchByUid` call, + * keyed by uid. Blank avatars/names collapse to `null` so callers can pass them straight + * into the dashboard shape. Returns an empty map when `uids` is empty (no query issued). + */ + def resolveOwnerInfo( + userDao: UserDao, + uids: Seq[Integer] + ): Map[Integer, (String, String)] = { + if (uids.isEmpty) Map.empty + else + userDao + .fetchByUid(uids: _*) + .asScala + .map { u => + val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull + val name = Option(u.getName).filter(_.nonEmpty).orNull + u.getUid -> (avatar, name) + } + .toMap + } + def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState = { unit.getType match { // Local CUs are always “running” @@ -58,4 +89,118 @@ object ComputingUnitHelpers { WorkflowComputingUnitMetrics("NaN", "NaN") } } + + /** + * Bulk variant of [[getComputingUnitStatus]]: resolves a unit's status from a pre-fetched + * map of pod name -> phase (see [[KubernetesClient.getAllPodPhases]]) instead of issuing a + * per-unit Kubernetes call. Used by listings that resolve many units at once so the number + * of cluster round trips is O(1) rather than O(units). + */ + def getComputingUnitStatus( + unit: WorkflowComputingUnit, + podPhases: Map[String, String] + ): ComputingUnitState = { + unit.getType match { + case WorkflowComputingUnitTypeEnum.local => + Running + case WorkflowComputingUnitTypeEnum.kubernetes => + // A missing entry or a null phase both yield `false` here (null != "Running"). + if (podPhases.get(KubernetesClient.generatePodName(unit.getCuid)).contains("Running")) + Running + else Pending + case _ => + Pending + } + } + + /** + * Bulk variant of [[getComputingUnitMetrics]]: resolves a unit's metrics from a pre-fetched + * map of pod name -> (metric -> value) (see [[KubernetesClient.getAllPodMetrics]]) instead of + * re-fetching the whole namespace metrics list per unit. + */ + def getComputingUnitMetrics( + unit: WorkflowComputingUnit, + podMetrics: Map[String, Map[String, String]] + ): WorkflowComputingUnitMetrics = { + unit.getType match { + case WorkflowComputingUnitTypeEnum.local => + WorkflowComputingUnitMetrics("NaN", "NaN") + case WorkflowComputingUnitTypeEnum.kubernetes => + val metrics = podMetrics + .getOrElse(KubernetesClient.generatePodName(unit.getCuid), Map.empty[String, String]) + WorkflowComputingUnitMetrics( + metrics.getOrElse("cpu", ""), + metrics.getOrElse("memory", "") + ) + case _ => + WorkflowComputingUnitMetrics("NaN", "NaN") + } + } + + /** + * A Kubernetes unit is considered "vanished" when its pod is absent from the pre-fetched + * `podPhases` map (manually deleted or TTL GC-ed by the cluster). Local/other units always + * count as present. + */ + private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean = + unit.getType == WorkflowComputingUnitTypeEnum.kubernetes && + !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid)) + + /** + * Split `units` into `(live, vanished)` using a pre-fetched pod-phase map (see + * [[KubernetesClient.getAllPodPhases]]). Pure — does no I/O — so it can be unit-tested. + */ + def partitionLiveUnits( + units: List[WorkflowComputingUnit], + podPhases: Map[String, String] + ): (List[WorkflowComputingUnit], List[WorkflowComputingUnit]) = { + val (vanished, live) = units.partition(isVanished(_, podPhases)) + (live, vanished) + } + + /** + * Reconcile a set of units against the cluster: any Kubernetes unit whose pod has vanished is + * stamped with `terminateTime` and persisted in a single batched update, then the surviving + * (live) units are returned. Shared by the per-user and admin listing endpoints so both agree + * on when a unit is treated as terminated. + */ + def reconcileVanishedKubernetesUnits( + dao: WorkflowComputingUnitDao, + units: List[WorkflowComputingUnit], + podPhases: Map[String, String] + ): List[WorkflowComputingUnit] = { + val (live, vanished) = partitionLiveUnits(units, podPhases) + if (vanished.nonEmpty) { + val now = new Timestamp(System.currentTimeMillis()) + vanished.foreach(_.setTerminateTime(now)) + dao.update(vanished.asJava) + } + live + } + + /** + * Assemble a single dashboard row from a unit, its caller-relative ownership/privilege, and + * the pre-fetched owner-info/pod maps. Kubernetes status and metrics are resolved from the + * maps (no per-unit Kubernetes call). Shared by both listing endpoints so the row shape and + * owner-info fallback stay identical. + */ + def buildDashboardUnit( + unit: WorkflowComputingUnit, + isOwner: Boolean, + accessPrivilege: EnumType, + ownerInfo: Map[Integer, (String, String)], + podPhases: Map[String, String], + podMetrics: Map[String, Map[String, String]] + ): DashboardWorkflowComputingUnit = { + val (avatar, name) = ownerInfo.getOrElse(unit.getUid, (null, null)) + DashboardWorkflowComputingUnit( + computingUnit = unit, + status = getComputingUnitStatus(unit, podPhases).toString, + metrics = getComputingUnitMetrics(unit, podMetrics), + isOwner = isOwner, + accessPrivilege = accessPrivilege, + ownerGoogleAvatar = avatar, + ownerName = name + ) + } } 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 4f1d391cb30..34390b02f93 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala @@ -20,7 +20,7 @@ package org.apache.texera.service.util import io.fabric8.kubernetes.api.model._ -import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetricsList +import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetrics import io.fabric8.kubernetes.client.KubernetesClientBuilder import org.apache.texera.common.config.KubernetesConfig @@ -48,19 +48,64 @@ object KubernetesClient { Option(client.pods().inNamespace(namespace).withName(podName).get()) } + /** + * Fetch the phase of every pod in the namespace in a single list call, keyed by pod name. + * Intended for bulk listings (e.g. the admin view) so that N units do not trigger N separate + * `getPodByName` round trips. Left unfiltered (rather than label-scoped) to match the + * name-based existence semantics of `getPodByName`/`podExists`: a caller checks + * `contains(generatePodName(cuid))` to decide whether a unit's pod still exists, and the + * `computing-unit-` names never collide with unrelated pods. + * + * A pod that exists but has no status yet maps to a `null` phase; callers can still rely + * on `contains(podName)` to decide whether the pod exists at all. + */ + def getAllPodPhases: Map[String, String] = { + client + .pods() + .inNamespace(namespace) + .list() + .getItems + .asScala + .map(pod => pod.getMetadata.getName -> Option(pod.getStatus).map(_.getPhase).orNull) + .toMap + } + + // Flatten a pod's per-container resource usage into a single metric -> value map. + private def containerUsage(podMetrics: PodMetrics): Map[String, String] = + podMetrics.getContainers.asScala.flatMap { container => + container.getUsage.asScala.map { + case (metric, value) => metric -> value.toString + } + }.toMap + + /** + * Fetch CPU/memory usage for every pod in the namespace in a single `top` call, keyed by + * pod name. Intended for bulk listings so that N units do not each re-fetch the whole + * namespace metrics list (which `getPodMetrics` does per call). + */ + def getAllPodMetrics: Map[String, Map[String, String]] = { + client + .top() + .pods() + .metrics(namespace) + .getItems + .asScala + .map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics)) + .toMap + } + def getPodMetrics(cuid: Int): Map[String, String] = { - val podMetricsList: PodMetricsList = client.top().pods().metrics(namespace) val targetPodName = generatePodName(cuid) - podMetricsList.getItems.asScala + client + .top() + .pods() + .metrics(namespace) + .getItems + .asScala .collectFirst { case podMetrics if podMetrics.getMetadata.getName == targetPodName => - podMetrics.getContainers.asScala.flatMap { container => - container.getUsage.asScala.map { - case (metric, value) => - metric -> value.toString - } - }.toMap + containerUsage(podMetrics) } .getOrElse(Map.empty[String, String]) } 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 af145455050..597c50289b1 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 @@ -22,28 +22,32 @@ package org.apache.texera.service.resource import jakarta.annotation.security.RolesAllowed import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.service.util.KubernetesClient import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.sql.Timestamp - class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { - private def localUnit( + private def makeUnit( cuid: Int, uid: Int, name: String, - terminated: Boolean = false + tpe: WorkflowComputingUnitTypeEnum ): WorkflowComputingUnit = { - val unit = new WorkflowComputingUnit() - unit.setCuid(cuid) - unit.setUid(uid) - unit.setName(name) - unit.setType(WorkflowComputingUnitTypeEnum.local) - if (terminated) unit.setTerminateTime(new Timestamp(0L)) - unit + val u = new WorkflowComputingUnit() + u.setCuid(cuid) + u.setUid(uid) + u.setName(name) + u.setType(tpe) + u } + private def localUnit(cuid: Int, uid: Int, name: String): WorkflowComputingUnit = + makeUnit(cuid, uid, name, WorkflowComputingUnitTypeEnum.local) + + private def kubernetesUnit(cuid: Int, uid: Int, name: String): WorkflowComputingUnit = + makeUnit(cuid, uid, name, WorkflowComputingUnitTypeEnum.kubernetes) + // The class-level @RolesAllowed(Array("ADMIN")) is what makes Jersey's // RolesAllowedDynamicFeature return 403 for any non-ADMIN (e.g. REGULAR) caller. "AdminComputingUnitResource" should "only permit the ADMIN role (non-ADMIN callers are rejected)" in { @@ -52,11 +56,10 @@ class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { annotation.value.toSeq shouldBe Seq("ADMIN") } - "buildDashboardUnits" should "return active units owned by multiple users, excluding terminated ones" in { + "buildDashboardUnits" should "render active units owned by multiple users with their owner info" in { val units = List( localUnit(cuid = 1, uid = 100, name = "alice-cu"), - localUnit(cuid = 2, uid = 200, name = "bob-cu"), - localUnit(cuid = 3, uid = 100, name = "alice-terminated", terminated = true) + localUnit(cuid = 2, uid = 200, name = "bob-cu") ) val ownerInfo: Map[Integer, (String, String)] = Map( (100: Integer) -> ("alice-avatar", "alice"), @@ -64,7 +67,13 @@ class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { ) // caller is an admin who owns none of these units - val result = AdminComputingUnitResource.buildDashboardUnits(units, ownerInfo, callerUid = 999) + val result = AdminComputingUnitResource.buildDashboardUnits( + units, + ownerInfo, + callerUid = 999, + podPhases = Map.empty, + podMetrics = Map.empty + ) result.map(_.computingUnit.getCuid) should contain theSameElementsAs Seq(1, 2) result.map(_.computingUnit.getUid).distinct should contain theSameElementsAs Seq(100, 200) @@ -80,4 +89,76 @@ class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { all(result.map(_.isOwner)) shouldBe false all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE } + + it should "set isOwner only for units owned by the caller" in { + val units = List( + localUnit(cuid = 1, uid = 100, name = "owned-by-caller"), + localUnit(cuid = 2, uid = 200, name = "owned-by-other") + ) + + val result = AdminComputingUnitResource + .buildDashboardUnits( + units, + ownerInfo = Map.empty, + callerUid = 100, + podPhases = Map.empty, + podMetrics = Map.empty + ) + .map(u => u.computingUnit.getCuid.intValue() -> u.isOwner) + .toMap + + result(1) shouldBe true + result(2) shouldBe false + } + + it should "fall back to null owner info when the owner is missing from the map" in { + val units = List(localUnit(cuid = 1, uid = 100, name = "orphan-cu")) + + val result = AdminComputingUnitResource.buildDashboardUnits( + units, + ownerInfo = Map.empty, + callerUid = 999, + podPhases = Map.empty, + podMetrics = Map.empty + ) + + result.head.ownerName shouldBe null + result.head.ownerGoogleAvatar shouldBe null + } + + it should "resolve kubernetes status and metrics from the pre-fetched maps" in { + val unit = kubernetesUnit(cuid = 5, uid = 300, name = "k8s-cu") + val podName = KubernetesClient.generatePodName(5) + + val podPhases = Map(podName -> "Running") + val podMetrics = Map(podName -> Map("cpu" -> "250m", "memory" -> "128Mi")) + + val result = AdminComputingUnitResource.buildDashboardUnits( + List(unit), + ownerInfo = Map((300: Integer) -> ("k8s-avatar", "k8s-owner")), + callerUid = 999, + podPhases = podPhases, + podMetrics = podMetrics + ) + + result.head.status shouldBe "Running" + result.head.metrics.cpuUsage shouldBe "250m" + result.head.metrics.memoryUsage shouldBe "128Mi" + } + + it should "report a kubernetes unit as Pending with empty metrics when its pod is absent from the maps" in { + val unit = kubernetesUnit(cuid = 6, uid = 300, name = "k8s-no-pod") + + val result = AdminComputingUnitResource.buildDashboardUnits( + List(unit), + ownerInfo = Map.empty, + callerUid = 999, + podPhases = Map.empty, + podMetrics = Map.empty + ) + + result.head.status shouldBe "Pending" + result.head.metrics.cpuUsage shouldBe "" + result.head.metrics.memoryUsage shouldBe "" + } } 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 39a11ea2a54..e6fc0124804 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala @@ -19,7 +19,7 @@ package org.apache.texera.service.util -import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics import org.apache.texera.service.resource.ComputingUnitState.{Pending, Running} @@ -28,12 +28,22 @@ import org.scalatest.matchers.should.Matchers class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { - private def localUnit(): WorkflowComputingUnit = { + private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit = { val unit = new WorkflowComputingUnit() + unit.setCuid(cuid) + unit.setUid(uid) unit.setType(WorkflowComputingUnitTypeEnum.local) unit } + private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit() + unit.setCuid(cuid) + unit.setUid(uid) + unit.setType(WorkflowComputingUnitTypeEnum.kubernetes) + unit + } + // WorkflowComputingUnitTypeEnum only defines `local` and `kubernetes`, so an // untyped unit (getType == null) is what exercises the pure "unknown" branch. private def untypedUnit(): WorkflowComputingUnit = new WorkflowComputingUnit() @@ -55,4 +65,110 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit()) shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") } + + // ── Bulk variants resolving from pre-fetched pod maps ──────────────── + + "getComputingUnitStatus(unit, podPhases)" should "return Running for a local unit" in { + ComputingUnitHelpers.getComputingUnitStatus(localUnit(), Map.empty) shouldBe Running + } + + it should "return Running for a kubernetes unit whose pod phase is Running" in { + val unit = kubernetesUnit(7) + val podPhases = Map(KubernetesClient.generatePodName(7) -> "Running") + ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Running + } + + it should "return Pending for a kubernetes unit whose pod is absent or not Running" in { + val unit = kubernetesUnit(8) + ComputingUnitHelpers.getComputingUnitStatus(unit, Map.empty) shouldBe Pending + ComputingUnitHelpers.getComputingUnitStatus( + unit, + Map(KubernetesClient.generatePodName(8) -> "Pending") + ) shouldBe Pending + } + + it should "treat a null phase as not Running" in { + val unit = kubernetesUnit(9) + val podPhases = Map(KubernetesClient.generatePodName(9) -> (null: String)) + ComputingUnitHelpers.getComputingUnitStatus(unit, podPhases) shouldBe Pending + } + + "getComputingUnitMetrics(unit, podMetrics)" should "return NaN metrics for a local unit" in { + ComputingUnitHelpers.getComputingUnitMetrics(localUnit(), Map.empty) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + it should "resolve cpu/memory for a kubernetes unit from the map" in { + val unit = kubernetesUnit(10) + val podMetrics = Map( + KubernetesClient.generatePodName(10) -> Map("cpu" -> "500m", "memory" -> "256Mi") + ) + ComputingUnitHelpers.getComputingUnitMetrics(unit, podMetrics) shouldBe + WorkflowComputingUnitMetrics("500m", "256Mi") + } + + it should "return empty cpu/memory for a kubernetes unit absent from the map" in { + ComputingUnitHelpers.getComputingUnitMetrics(kubernetesUnit(11), Map.empty) shouldBe + WorkflowComputingUnitMetrics("", "") + } + + // ── partitionLiveUnits ─────────────────────────────────────────────── + + "partitionLiveUnits" should "treat local units as always live" in { + val units = List(localUnit(cuid = 1), localUnit(cuid = 2)) + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(units, Map.empty) + live.map(_.getCuid) shouldBe List(1, 2) + vanished shouldBe empty + } + + it should "classify a kubernetes unit as live iff its pod is present in the map" in { + val present = kubernetesUnit(20) + val gone = kubernetesUnit(21) + val podPhases = Map(KubernetesClient.generatePodName(20) -> "Running") + + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(present, gone), podPhases) + + live.map(_.getCuid) shouldBe List(20) + vanished.map(_.getCuid) shouldBe List(21) + } + + // ── buildDashboardUnit ─────────────────────────────────────────────── + + "buildDashboardUnit" should "populate the row from the caller flags and pre-fetched maps" in { + val unit = kubernetesUnit(cuid = 30, uid = 100) + val podName = KubernetesClient.generatePodName(30) + + val row = ComputingUnitHelpers.buildDashboardUnit( + unit, + isOwner = true, + accessPrivilege = PrivilegeEnum.READ, + ownerInfo = Map((100: Integer) -> ("avatar", "owner")), + podPhases = Map(podName -> "Running"), + podMetrics = Map(podName -> Map("cpu" -> "100m", "memory" -> "64Mi")) + ) + + row.computingUnit.getCuid shouldBe 30 + row.isOwner shouldBe true + row.accessPrivilege shouldBe PrivilegeEnum.READ + row.status shouldBe "Running" + row.metrics shouldBe WorkflowComputingUnitMetrics("100m", "64Mi") + row.ownerGoogleAvatar shouldBe "avatar" + row.ownerName shouldBe "owner" + } + + it should "fall back to null owner info when the owner is missing from the map" in { + val row = ComputingUnitHelpers.buildDashboardUnit( + localUnit(cuid = 31, uid = 200), + isOwner = false, + accessPrivilege = PrivilegeEnum.WRITE, + ownerInfo = Map.empty, + podPhases = Map.empty, + podMetrics = Map.empty + ) + + row.ownerGoogleAvatar shouldBe null + row.ownerName shouldBe null + row.status shouldBe "Running" + row.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") + } } From d111e806c613e056ec9b269797590f2996eefd0b Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Fri, 17 Jul 2026 22:02:53 -0400 Subject: [PATCH 3/8] refactor(computing-unit-managing-service): dedup CU listing helpers Simplify the computing-unit listing logic shared by the per-user and admin endpoints: - Add a single ComputingUnitHelpers.isKubernetes predicate plus podPhasesFor/podMetricsFor helpers, collapsing the four copies of the "only hit the cluster when a k8s unit is present" guard. - partitionLiveUnits returns (live, vanished) via a negated predicate instead of partitioning into (vanished, live) and swapping the tuple. - Per-user listComputingUnits maps over the reconciled live units directly (privilege captured by cuid) instead of round-tripping through a Set[cuid], and resolves owner info from the live units to match the admin endpoint. - Extract KubernetesClient.fetchPodMetricsItems so getAllPodMetrics and getPodMetrics share one top() fetch chain. API response behavior is unchanged. --- .../resource/AdminComputingUnitResource.scala | 21 +++---- .../ComputingUnitManagingResource.scala | 60 +++++++++---------- .../service/util/ComputingUnitHelpers.scala | 26 ++++++-- .../service/util/KubernetesClient.scala | 22 +++---- 4 files changed, 62 insertions(+), 67 deletions(-) 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 8c866891b93..173908cdc3e 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 @@ -26,11 +26,11 @@ 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, WorkflowComputingUnitTypeEnum} +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, KubernetesClient} +import org.apache.texera.service.util.ComputingUnitHelpers import org.jooq.DSLContext import scala.jdk.CollectionConverters.CollectionHasAsScala @@ -51,7 +51,7 @@ object AdminComputingUnitResource { * @param units active computing units to render (across every owner) * @param ownerInfo map of owner uid -> (googleAvatar, userName) * @param callerUid the uid of the requesting admin, used to populate `isOwner` - * @param podPhases map of pod name -> phase (see [[KubernetesClient.getAllPodPhases]]) + * @param podPhases map of pod name -> phase (see KubernetesClient.getAllPodPhases) * @param podMetrics map of pod name -> (metric -> value) (see KubernetesClient.getAllPodMetrics) */ def buildDashboardUnits( @@ -102,9 +102,6 @@ class AdminComputingUnitResource { ): List[DashboardWorkflowComputingUnit] = { val ctx = context - def isKubernetes(unit: WorkflowComputingUnit): Boolean = - unit.getType == WorkflowComputingUnitTypeEnum.kubernetes - // Filter to active units in SQL so historically-terminated rows are never loaded. val activeUnits: List[WorkflowComputingUnit] = ctx @@ -115,9 +112,8 @@ class AdminComputingUnitResource { .toList // Pod phases (one namespace-wide `list`) are needed to decide which units are still alive; - // only fetch them when there is a Kubernetes unit to resolve. - val podPhases: Map[String, String] = - if (activeUnits.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty + // only fetched when there is a Kubernetes unit to resolve. + val podPhases = ComputingUnitHelpers.podPhasesFor(activeUnits) // A Kubernetes unit whose pod is gone is stamped terminated and dropped from the response. val liveUnits = ComputingUnitHelpers.reconcileVanishedKubernetesUnits( @@ -126,10 +122,9 @@ class AdminComputingUnitResource { podPhases ) - // Metrics (one namespace-wide `top`) are only rendered for surviving units, so defer this - // call until after reconciliation and skip it when no live Kubernetes unit remains. - val podMetrics: Map[String, Map[String, String]] = - if (liveUnits.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty + // Metrics (one namespace-wide `top`) are only rendered for surviving units, so this is + // deferred until after reconciliation and skipped when no live Kubernetes unit remains. + val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) val userDao = new UserDao(ctx.configuration()) val ownerInfo = ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct) diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala index d1e5ef3931c..871ff1b89d3 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala @@ -493,7 +493,6 @@ class ComputingUnitManagingResource { (List.empty[WorkflowComputingUnit], Map.empty[Integer, PrivilegeEnum]) } - val allUnits = ownedUnits ++ sharedUnits val userDao = new UserDao(ctx.configuration()) // Pair each unit with the caller's privilege (owned units default to WRITE), keeping only @@ -504,46 +503,41 @@ class ComputingUnitManagingResource { 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 (one namespace-wide `list`) decide which Kubernetes units are still alive; - // only fetch them when there is a Kubernetes unit to resolve. - val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit } - val podPhases: Map[String, String] = - if (candidateUnits.exists(_.getType == WorkflowComputingUnitTypeEnum.kubernetes)) - KubernetesClient.getAllPodPhases - else Map.empty + // only fetched when there is a Kubernetes unit to resolve. + val podPhases = ComputingUnitHelpers.podPhasesFor(candidateUnits) // A Kubernetes unit whose pod has vanished (manually deleted or TTL GC-ed by the cluster) // is treated as terminated: its terminateTime is persisted and it is dropped here so // subsequent API calls no longer return it. - val liveCuids = ComputingUnitHelpers - .reconcileVanishedKubernetesUnits(computingUnitDao, candidateUnits, podPhases) - .map(_.getCuid) - .toSet - val liveUnitsWithPrivilege = - unitsWithPrivilege.filter { case (unit, _) => liveCuids.contains(unit.getCuid) } - - // Metrics (one namespace-wide `top`) are only rendered for surviving units, so defer this - // call until after reconciliation and skip it when no live Kubernetes unit remains. - val podMetrics: Map[String, Map[String, String]] = - if (liveUnitsWithPrivilege.exists { - case (unit, _) => unit.getType == WorkflowComputingUnitTypeEnum.kubernetes - }) KubernetesClient.getAllPodMetrics - else Map.empty + val liveUnits = + ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + computingUnitDao, + candidateUnits, + podPhases + ) + + // Metrics (one namespace-wide `top`) are only rendered for surviving units, so this is + // deferred until after reconciliation and skipped when no live Kubernetes unit remains. + val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) val ownerInfoMap = - ComputingUnitHelpers.resolveOwnerInfo(userDao, allUnits.map(_.getUid).distinct) - - liveUnitsWithPrivilege.map { - case (unit, privilege) => - ComputingUnitHelpers.buildDashboardUnit( - unit, - isOwner = unit.getUid.equals(uid), - accessPrivilege = privilege, - ownerInfo = ownerInfoMap, - podPhases = podPhases, - podMetrics = podMetrics - ) + 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 + ) } } } 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 067a99b6336..6d40184009e 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 @@ -137,14 +137,30 @@ object ComputingUnitHelpers { } } + private def isKubernetes(unit: WorkflowComputingUnit): Boolean = + unit.getType == WorkflowComputingUnitTypeEnum.kubernetes + + /** + * Fetch namespace-wide pod phases (one `list`) only when `units` contains a Kubernetes unit; + * otherwise return an empty map without touching the cluster. + */ + def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] = + if (units.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty + + /** + * Fetch namespace-wide pod metrics (one `top`) only when `units` contains a Kubernetes unit; + * otherwise return an empty map without touching the cluster. + */ + def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] = + if (units.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty + /** * A Kubernetes unit is considered "vanished" when its pod is absent from the pre-fetched * `podPhases` map (manually deleted or TTL GC-ed by the cluster). Local/other units always * count as present. */ private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean = - unit.getType == WorkflowComputingUnitTypeEnum.kubernetes && - !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid)) + isKubernetes(unit) && !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid)) /** * Split `units` into `(live, vanished)` using a pre-fetched pod-phase map (see @@ -153,10 +169,8 @@ object ComputingUnitHelpers { def partitionLiveUnits( units: List[WorkflowComputingUnit], podPhases: Map[String, String] - ): (List[WorkflowComputingUnit], List[WorkflowComputingUnit]) = { - val (vanished, live) = units.partition(isVanished(_, podPhases)) - (live, vanished) - } + ): (List[WorkflowComputingUnit], List[WorkflowComputingUnit]) = + units.partition(unit => !isVanished(unit, podPhases)) /** * Reconcile a set of units against the cluster: any Kubernetes unit whose pod has vanished is 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 34390b02f93..c6911446f37 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 @@ -78,31 +78,23 @@ object KubernetesClient { } }.toMap + // One namespace-wide `top` call, returning the raw per-pod metrics items. + private def fetchPodMetricsItems(): Iterable[PodMetrics] = + client.top().pods().metrics(namespace).getItems.asScala + /** * Fetch CPU/memory usage for every pod in the namespace in a single `top` call, keyed by * pod name. Intended for bulk listings so that N units do not each re-fetch the whole * namespace metrics list (which `getPodMetrics` does per call). */ - def getAllPodMetrics: Map[String, Map[String, String]] = { - client - .top() - .pods() - .metrics(namespace) - .getItems - .asScala + def getAllPodMetrics: Map[String, Map[String, String]] = + fetchPodMetricsItems() .map(podMetrics => podMetrics.getMetadata.getName -> containerUsage(podMetrics)) .toMap - } def getPodMetrics(cuid: Int): Map[String, String] = { val targetPodName = generatePodName(cuid) - - client - .top() - .pods() - .metrics(namespace) - .getItems - .asScala + fetchPodMetricsItems() .collectFirst { case podMetrics if podMetrics.getMetadata.getName == targetPodName => containerUsage(podMetrics) From d378186a246e7827afb3d2ccb4c3d8a45990f8e2 Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Fri, 17 Jul 2026 22:56:14 -0400 Subject: [PATCH 4/8] test(computing-unit-managing-service): cover CU listing paths to 100% Bring patch coverage of the admin-listing change to 100% by testing the previously-uncovered database, Kubernetes, and bootstrap paths: - KubernetesClientSpec now injects a stubbed fabric8 client (via a small test seam) to exercise getAllPodPhases/getAllPodMetrics/getPodMetrics, getPodByName/podExists, and ComputingUnitHelpers.podPhasesFor/podMetricsFor. - ComputingUnitHelpersSpec covers resolveOwnerInfo and reconcileVanishedKubernetesUnits against an embedded Postgres (MockTexeraDB), plus the untyped/unknown branches of the bulk status/metrics helpers. - AdminComputingUnitResourceSpec drives listAllComputingUnits end-to-end over the embedded DB (terminated rows excluded, WRITE access, isOwner). - New ComputingUnitManagingResourceSpec drives getComputingUnitInfo, the metrics endpoint, and listComputingUnits over the embedded DB. - ComputingUnitManagingServiceRunSpec boots run() with a mocked Dropwizard environment and verifies the resources (incl. the admin resource) register. Supporting changes: - KubernetesClient: add a package-private setClientForTesting seam so the fabric8 client can be stubbed in tests (not used in production). - ComputingUnitManagingService: open the DB pool in initialize() rather than run() (matching the sibling services) so run() is testable without a DB. - ComputingUnitHelpers: express isKubernetes as a match and avoid tuple destructuring so branch coverage is complete. - build.sbt: depend on DAO/Auth test artifacts for MockTexeraDB and run the suites serially (they mutate JVM-wide singletons). --- build.sbt | 5 + .../ComputingUnitManagingService.scala | 14 +- .../service/util/ComputingUnitHelpers.scala | 16 +- .../service/util/KubernetesClient.scala | 11 +- .../ComputingUnitManagingServiceRunSpec.scala | 26 +++ .../AdminComputingUnitResourceSpec.scala | 60 ++++++- .../ComputingUnitManagingResourceSpec.scala | 101 +++++++++++ .../util/ComputingUnitHelpersSpec.scala | 96 +++++++++- .../service/util/KubernetesClientSpec.scala | 166 +++++++++++++++++- 9 files changed, 475 insertions(+), 20 deletions(-) create mode 100644 computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala diff --git a/build.sbt b/build.sbt index ee6882600d8..940b0bcb101 100644 --- a/build.sbt +++ b/build.sbt @@ -161,7 +161,12 @@ lazy val WorkflowCore = (project in file("common/workflow-core")) 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, diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala index f0dffc89e11..f338821bd7a 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala @@ -48,6 +48,14 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ ) // register scala module to dropwizard default object mapper bootstrap.getObjectMapper.registerModule(DefaultScalaModule) + + // Open the DB connection pool during bootstrap (as the sibling services do), + // so `run` only wires up HTTP resources and stays testable without a database. + SqlServer.initConnection( + StorageConfig.jdbcUrl, + StorageConfig.jdbcUsername, + StorageConfig.jdbcPassword + ) } override def run( configuration: ComputingUnitManagingServiceConfiguration, @@ -59,12 +67,6 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ AuthFeatures.register(environment) - SqlServer.initConnection( - StorageConfig.jdbcUrl, - StorageConfig.jdbcUsername, - StorageConfig.jdbcPassword - ) - environment.jersey().register(new ComputingUnitManagingResource) environment.jersey().register(new ComputingUnitAccessResource) environment.jersey().register(new AdminComputingUnitResource) 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 6d40184009e..827f92691ff 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 @@ -138,7 +138,10 @@ object ComputingUnitHelpers { } private def isKubernetes(unit: WorkflowComputingUnit): Boolean = - unit.getType == WorkflowComputingUnitTypeEnum.kubernetes + unit.getType match { + case WorkflowComputingUnitTypeEnum.kubernetes => true + case _ => false + } /** * Fetch namespace-wide pod phases (one `list`) only when `units` contains a Kubernetes unit; @@ -183,13 +186,14 @@ object ComputingUnitHelpers { units: List[WorkflowComputingUnit], podPhases: Map[String, String] ): List[WorkflowComputingUnit] = { - val (live, vanished) = partitionLiveUnits(units, podPhases) + val partitioned = partitionLiveUnits(units, podPhases) + val vanished = partitioned._2 if (vanished.nonEmpty) { val now = new Timestamp(System.currentTimeMillis()) vanished.foreach(_.setTerminateTime(now)) dao.update(vanished.asJava) } - live + partitioned._1 } /** @@ -206,15 +210,15 @@ object ComputingUnitHelpers { podPhases: Map[String, String], podMetrics: Map[String, Map[String, String]] ): DashboardWorkflowComputingUnit = { - val (avatar, name) = ownerInfo.getOrElse(unit.getUid, (null, null)) + val owner = ownerInfo.getOrElse(unit.getUid, (null, null)) DashboardWorkflowComputingUnit( computingUnit = unit, status = getComputingUnitStatus(unit, podPhases).toString, metrics = getComputingUnitMetrics(unit, podMetrics), isOwner = isOwner, accessPrivilege = accessPrivilege, - ownerGoogleAvatar = avatar, - ownerName = name + ownerGoogleAvatar = owner._1, + ownerName = owner._2 ) } } diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala index c6911446f37..64f05fffef2 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 @@ -29,9 +29,18 @@ import scala.jdk.CollectionConverters._ object KubernetesClient { // Initialize the Kubernetes client - private val client: io.fabric8.kubernetes.client.KubernetesClient = + private var client: io.fabric8.kubernetes.client.KubernetesClient = new KubernetesClientBuilder().build() private val namespace: String = KubernetesConfig.computeUnitPoolNamespace + + /** + * Test seam: swap in a stubbed client so the pod/metrics wrappers can be exercised without a + * live cluster. Not used in production code. + */ + private[util] def setClientForTesting( + testClient: io.fabric8.kubernetes.client.KubernetesClient + ): Unit = + client = testClient private val podNamePrefix = "computing-unit" def generatePodURI(cuid: Int): String = { diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala index 6962c2dc0d5..cf6a175a125 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala @@ -19,6 +19,11 @@ package org.apache.texera.service +import io.dropwizard.core.setup.Environment +import io.dropwizard.jersey.DropwizardResourceConfig +import io.dropwizard.jersey.setup.JerseyEnvironment +import io.dropwizard.jetty.MutableServletContextHandler +import io.dropwizard.jetty.setup.ServletEnvironment import org.apache.texera.auth.RoleAnnotationEnforcer import org.apache.texera.service.resource.{ AdminComputingUnitResource, @@ -26,6 +31,8 @@ import org.apache.texera.service.resource.{ ComputingUnitManagingResource, HealthCheckResource } +import org.mockito.ArgumentMatchers.isA +import org.mockito.Mockito.{mock, verify, when} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -42,4 +49,23 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { ) ) shouldBe empty } + + "ComputingUnitManagingService.run" should "register the admin resource on the Jersey environment" in { + val jersey = mock(classOf[JerseyEnvironment]) + val servlets = mock(classOf[ServletEnvironment]) + val context = mock(classOf[MutableServletContextHandler]) + val env = mock(classOf[Environment]) + when(env.jersey).thenReturn(jersey) + when(env.servlets).thenReturn(servlets) + when(env.getApplicationContext).thenReturn(context) + when(jersey.getResourceConfig).thenReturn(DropwizardResourceConfig.forTesting()) + + val service = new ComputingUnitManagingService + service.run(mock(classOf[ComputingUnitManagingServiceConfiguration]), env) + + verify(jersey).register(isA(classOf[ComputingUnitManagingResource])) + verify(jersey).register(isA(classOf[ComputingUnitAccessResource])) + verify(jersey).register(isA(classOf[AdminComputingUnitResource])) + verify(jersey).setUrlPattern("/api/*") + } } 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 597c50289b1..aaee56f4637 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,13 +20,42 @@ package org.apache.texera.service.resource import jakarta.annotation.security.RolesAllowed -import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} -import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + UserRoleEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit} import org.apache.texera.service.util.KubernetesClient +import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { +import java.sql.Timestamp + +class AdminComputingUnitResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + override protected def beforeAll(): Unit = initializeDBAndReplaceDSLContext() + + override protected def afterAll(): Unit = shutdownDB() + + private def makeUser(uid: Int, name: String): User = { + val u = new User() + u.setUid(uid) + u.setName(name) + u.setEmail(s"user$uid@example.com") + u.setRole(UserRoleEnum.ADMIN) + u.setPassword("password") + u.setGoogleAvatar(s"avatar-$uid") + u + } private def makeUnit( cuid: Int, @@ -161,4 +190,29 @@ class AdminComputingUnitResourceSpec extends AnyFlatSpec with Matchers { result.head.metrics.cpuUsage shouldBe "" result.head.metrics.memoryUsage shouldBe "" } + + "listAllComputingUnits" should "return every non-terminated unit across users, marked WRITE" in { + val userDao = new UserDao(getDSLContext.configuration()) + val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + val admin = makeUser(700, "admin") + userDao.insert(admin) + userDao.insert(makeUser(701, "other")) + unitDao.insert(localUnit(cuid = 700, uid = 700, name = "admin-cu")) + unitDao.insert(localUnit(cuid = 701, uid = 701, name = "other-cu")) + // A terminated unit must be excluded by the SQL filter. + val terminated = localUnit(cuid = 702, uid = 701, name = "terminated-cu") + terminated.setTerminateTime(new Timestamp(0L)) + unitDao.insert(terminated) + + val result = new AdminComputingUnitResource().listAllComputingUnits(new SessionUser(admin)) + + result.map(_.computingUnit.getCuid.intValue()) should contain theSameElementsAs Seq(700, 701) + all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE + all(result.map(_.status)) shouldBe "Running" // local units + val byCuid = result.map(r => r.computingUnit.getCuid.intValue() -> r).toMap + byCuid(700).isOwner shouldBe true // owned by the requesting admin + byCuid(700).ownerName shouldBe "admin" + byCuid(701).isOwner shouldBe false + byCuid(701).ownerName shouldBe "other" + } } diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala new file mode 100644 index 00000000000..0fb6e214afa --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + UserRoleEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit} +import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +// Drives the per-user computing-unit endpoints against the embedded database using +// local units (so no Kubernetes calls are made). +class ComputingUnitManagingResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + private val uid = 800 + private lazy val user: SessionUser = { + val u = new User() + u.setUid(uid) + u.setName("owner") + u.setEmail("owner@example.com") + u.setRole(UserRoleEnum.REGULAR) + u.setPassword("password") + u.setGoogleAvatar("owner-avatar") + new SessionUser(u) + } + + private def localUnit(cuid: Int, name: String): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit() + unit.setCuid(cuid) + unit.setUid(uid) + unit.setName(name) + unit.setType(WorkflowComputingUnitTypeEnum.local) + unit + } + + override protected def beforeAll(): Unit = { + initializeDBAndReplaceDSLContext() + new UserDao(getDSLContext.configuration()).insert(user.getUser) + val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + unitDao.insert(localUnit(800, "cu-a")) + unitDao.insert(localUnit(801, "cu-b")) + } + + override protected def afterAll(): Unit = shutdownDB() + + private val resource = new ComputingUnitManagingResource + + "getComputingUnitInfo" should "return the owner's local unit with WRITE access and Running status" in { + val info = resource.getComputingUnitInfo(800, user) + + info.computingUnit.getCuid shouldBe 800 + info.status shouldBe "Running" + info.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") + info.isOwner shouldBe true + info.accessPrivilege shouldBe PrivilegeEnum.WRITE + info.ownerName shouldBe "owner" + } + + "getComputingUnitMetricsEndpoint" should "return NaN metrics for an owned local unit" in { + resource.getComputingUnitMetricsEndpoint("800", user) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + "listComputingUnits" should "return the caller's owned, non-terminated units" in { + val result = resource.listComputingUnits(user) + + result.map(_.computingUnit.getCuid.intValue()) should contain theSameElementsAs Seq(800, 801) + all(result.map(_.isOwner)) shouldBe true + all(result.map(_.accessPrivilege)) shouldBe PrivilegeEnum.WRITE + all(result.map(_.status)) shouldBe "Running" + } +} diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala index e6fc0124804..97cf94099c9 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala @@ -19,14 +19,40 @@ package org.apache.texera.service.util -import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} -import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + UserRoleEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, WorkflowComputingUnit} import org.apache.texera.service.resource.ComputingUnitManagingResource.WorkflowComputingUnitMetrics import org.apache.texera.service.resource.ComputingUnitState.{Pending, Running} +import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { +class ComputingUnitHelpersSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + override protected def beforeAll(): Unit = initializeDBAndReplaceDSLContext() + + override protected def afterAll(): Unit = shutdownDB() + + private def makeUser(uid: Int, name: String, email: String, avatar: String): User = { + val u = new User() + u.setUid(uid) + u.setName(name) + u.setEmail(email) + u.setRole(UserRoleEnum.REGULAR) + u.setPassword("password") + u.setGoogleAvatar(avatar) + u + } private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit = { val unit = new WorkflowComputingUnit() @@ -132,6 +158,12 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { vanished.map(_.getCuid) shouldBe List(21) } + it should "treat an untyped (null-type) unit as live (never kubernetes)" in { + val (live, vanished) = ComputingUnitHelpers.partitionLiveUnits(List(untypedUnit()), Map.empty) + live should have size 1 + vanished shouldBe empty + } + // ── buildDashboardUnit ─────────────────────────────────────────────── "buildDashboardUnit" should "populate the row from the caller flags and pre-fetched maps" in { @@ -171,4 +203,62 @@ class ComputingUnitHelpersSpec extends AnyFlatSpec with Matchers { row.status shouldBe "Running" row.metrics shouldBe WorkflowComputingUnitMetrics("NaN", "NaN") } + + // ── Bulk variants: unknown (untyped) branch ────────────────────────── + + "getComputingUnitStatus(unit, podPhases)" should "return Pending for an unknown (untyped) unit" in { + ComputingUnitHelpers.getComputingUnitStatus(untypedUnit(), Map.empty) shouldBe Pending + } + + "getComputingUnitMetrics(unit, podMetrics)" should "return NaN for an unknown (untyped) unit" in { + ComputingUnitHelpers.getComputingUnitMetrics(untypedUnit(), Map.empty) shouldBe + WorkflowComputingUnitMetrics("NaN", "NaN") + } + + // ── resolveOwnerInfo (backed by the embedded database) ─────────────── + + "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in { + val userDao = new UserDao(getDSLContext.configuration()) + userDao.insert(makeUser(500, "alice", "alice@example.com", "alice-avatar")) + userDao.insert(makeUser(501, "", "bob@example.com", "")) + + val info = ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq[Integer](500, 501)) + info(500) shouldBe (("alice-avatar", "alice")) + info(501) shouldBe ((null, null)) + } + + it should "return an empty map (and issue no query) for no uids" in { + val userDao = new UserDao(getDSLContext.configuration()) + ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq.empty) shouldBe empty + } + + // ── reconcileVanishedKubernetesUnits (backed by the embedded database) ─ + + "reconcileVanishedKubernetesUnits" should "terminate vanished kubernetes units and return the live ones" in { + val userDao = new UserDao(getDSLContext.configuration()) + userDao.insert(makeUser(600, "carol", "carol@example.com", null)) + val dao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + + val present = kubernetesUnit(600, 600) + present.setName("present") + val gone = kubernetesUnit(601, 600) + gone.setName("gone") + val local = localUnit(602, 600) + local.setName("local") + Seq(present, gone, local).foreach(dao.insert(_)) + + // Only the pod for cuid 600 exists; cuid 601's pod has vanished. + val podPhases = Map(KubernetesClient.generatePodName(600) -> "Running") + val live = + ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + dao, + List(present, gone, local), + podPhases + ) + + live.map(_.getCuid) should contain theSameElementsAs Seq(600, 602) + dao.fetchOneByCuid(601).getTerminateTime should not be null + dao.fetchOneByCuid(600).getTerminateTime shouldBe null + dao.fetchOneByCuid(602).getTerminateTime shouldBe null + } } diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala index 8ce3b107ae4..3354873cf2f 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala @@ -19,10 +19,125 @@ package org.apache.texera.service.util +import io.fabric8.kubernetes.api.model.metrics.v1beta1.{ + ContainerMetricsBuilder, + PodMetricsBuilder, + PodMetricsList, + PodMetricsListBuilder +} +import io.fabric8.kubernetes.api.model.{Pod, PodBuilder, PodList, PodListBuilder, Quantity} +import io.fabric8.kubernetes.client.dsl.{ + MetricAPIGroupDSL, + MixedOperation, + NonNamespaceOperation, + PodMetricOperation, + PodResource +} +import io.fabric8.kubernetes.client.{KubernetesClientBuilder, KubernetesClient => Fabric8Client} +import org.apache.texera.common.config.KubernetesConfig +import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTypeEnum +import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.mockito.Mockito.{mock, when} +import org.scalatest.BeforeAndAfterAll import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -class KubernetesClientSpec extends AnyFlatSpec with Matchers { +import scala.jdk.CollectionConverters._ + +// Exercises the fabric8 wrappers without a cluster by injecting a client whose +// fluent pods()/top() chains are stubbed with Mockito. +class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll { + + private val namespace: String = KubernetesConfig.computeUnitPoolNamespace + + private def unitOfType(cuid: Int, tpe: WorkflowComputingUnitTypeEnum): WorkflowComputingUnit = { + val u = new WorkflowComputingUnit() + u.setCuid(cuid) + u.setType(tpe) + u + } + + // cuid 1 -> Running, cuid 2 -> Pending; a namespace-wide list returns both. + private val podList: PodList = + new PodListBuilder() + .addToItems( + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(1)) + .endMetadata() + .withNewStatus() + .withPhase("Running") + .endStatus() + .build(), + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(2)) + .endMetadata() + .withNewStatus() + .withPhase("Pending") + .endStatus() + .build() + ) + .build() + + // Only cuid 1 reports metrics. + private val metricsList: PodMetricsList = + new PodMetricsListBuilder() + .addToItems( + new PodMetricsBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(1)) + .endMetadata() + .addToContainers( + new ContainerMetricsBuilder() + .withName("main") + .withUsage( + Map("cpu" -> new Quantity("250m"), "memory" -> new Quantity("128Mi")).asJava + ) + .build() + ) + .build() + ) + .build() + + override protected def beforeAll(): Unit = { + val client = mock(classOf[Fabric8Client]) + + val podsMixed = mock(classOf[MixedOperation[_, _, _]]) + .asInstanceOf[MixedOperation[Pod, PodList, PodResource]] + val podsInNamespace = mock(classOf[NonNamespaceOperation[_, _, _]]) + .asInstanceOf[NonNamespaceOperation[Pod, PodList, PodResource]] + when(client.pods()).thenReturn(podsMixed) + when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace) + when(podsInNamespace.list()).thenReturn(podList) + + // withName(cuid 1) -> a present pod; withName(cuid 2) -> no pod. + val presentResource = mock(classOf[PodResource]) + when(presentResource.get()).thenReturn( + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(1)) + .endMetadata() + .build() + ) + val absentResource = mock(classOf[PodResource]) + when(absentResource.get()).thenReturn(null) + when(podsInNamespace.withName(KubernetesClient.generatePodName(1))).thenReturn(presentResource) + when(podsInNamespace.withName(KubernetesClient.generatePodName(2))).thenReturn(absentResource) + + val top = mock(classOf[MetricAPIGroupDSL]) + val podMetricOp = mock(classOf[PodMetricOperation]) + when(client.top()).thenReturn(top) + when(top.pods()).thenReturn(podMetricOp) + when(podMetricOp.metrics(namespace)).thenReturn(metricsList) + + KubernetesClient.setClientForTesting(client) + } + + override protected def afterAll(): Unit = { + // Restore a real client so subsequent suites don't reuse the mock. + KubernetesClient.setClientForTesting(new KubernetesClientBuilder().build()) + } "generatePodName" should "prefix the cuid with computing-unit" in { KubernetesClient.generatePodName(42) shouldBe "computing-unit-42" @@ -31,4 +146,53 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers { it should "handle a cuid of 0" in { KubernetesClient.generatePodName(0) shouldBe "computing-unit-0" } + + "getPodByName" should "return the pod when present and None when absent" in { + KubernetesClient.getPodByName(KubernetesClient.generatePodName(1)) shouldBe defined + KubernetesClient.getPodByName(KubernetesClient.generatePodName(2)) shouldBe None + } + + "podExists" should "reflect whether the pod is present" in { + KubernetesClient.podExists(1) shouldBe true + KubernetesClient.podExists(2) shouldBe false + } + + "getAllPodPhases" should "map every pod name in the namespace to its phase" in { + val phases = KubernetesClient.getAllPodPhases + phases(KubernetesClient.generatePodName(1)) shouldBe "Running" + phases(KubernetesClient.generatePodName(2)) shouldBe "Pending" + } + + "getAllPodMetrics" should "map pod names to their cpu/memory usage" in { + val metrics = KubernetesClient.getAllPodMetrics + metrics(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") + } + + "getPodMetrics" should "return the usage for the pod matching the cuid" in { + KubernetesClient.getPodMetrics(1) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") + } + + it should "return an empty map when no pod matches the cuid" in { + KubernetesClient.getPodMetrics(999) shouldBe empty + } + + "ComputingUnitHelpers.podPhasesFor" should "query the cluster iff a kubernetes unit is present" in { + ComputingUnitHelpers.podPhasesFor( + List(unitOfType(5, WorkflowComputingUnitTypeEnum.local)) + ) shouldBe empty + + ComputingUnitHelpers.podPhasesFor( + List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes)) + )(KubernetesClient.generatePodName(1)) shouldBe "Running" + } + + "ComputingUnitHelpers.podMetricsFor" should "query the cluster iff a kubernetes unit is present" in { + ComputingUnitHelpers.podMetricsFor( + List(unitOfType(5, WorkflowComputingUnitTypeEnum.local)) + ) shouldBe empty + + ComputingUnitHelpers.podMetricsFor( + List(unitOfType(1, WorkflowComputingUnitTypeEnum.kubernetes)) + )(KubernetesClient.generatePodName(1)) shouldBe Map("cpu" -> "250m", "memory" -> "128Mi") + } } From cdcae1c4f2f57ae6b94003d1e311faf07eecddbb Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Fri, 17 Jul 2026 23:13:04 -0400 Subject: [PATCH 5/8] test(computing-unit-managing-service): cover run() resource registration Drive ComputingUnitManagingService.run() in the run-spec so the resource registrations (including the admin resource) are covered. run() opens the real connection pool via SqlServer.initConnection, so the test connects only when Postgres is reachable at the configured JDBC URL (as provisioned in CI) and cancels otherwise, keeping local runs green without a database. Revert the earlier relocation of initConnection to initialize(): keeping it in run() leaves the service's diff minimal (no bootstrap lines become newly uncovered) while the run-spec above covers the registration path. --- .../ComputingUnitManagingService.scala | 14 +++++----- .../ComputingUnitManagingServiceRunSpec.scala | 27 +++++++++++++++++-- 2 files changed, 31 insertions(+), 10 deletions(-) diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala index f338821bd7a..f0dffc89e11 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/ComputingUnitManagingService.scala @@ -48,14 +48,6 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ ) // register scala module to dropwizard default object mapper bootstrap.getObjectMapper.registerModule(DefaultScalaModule) - - // Open the DB connection pool during bootstrap (as the sibling services do), - // so `run` only wires up HTTP resources and stays testable without a database. - SqlServer.initConnection( - StorageConfig.jdbcUrl, - StorageConfig.jdbcUsername, - StorageConfig.jdbcPassword - ) } override def run( configuration: ComputingUnitManagingServiceConfiguration, @@ -67,6 +59,12 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ AuthFeatures.register(environment) + SqlServer.initConnection( + StorageConfig.jdbcUrl, + StorageConfig.jdbcUsername, + StorageConfig.jdbcPassword + ) + environment.jersey().register(new ComputingUnitManagingResource) environment.jersey().register(new ComputingUnitAccessResource) environment.jersey().register(new AdminComputingUnitResource) diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala index cf6a175a125..0835d064d72 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala @@ -25,6 +25,7 @@ import io.dropwizard.jersey.setup.JerseyEnvironment import io.dropwizard.jetty.MutableServletContextHandler import io.dropwizard.jetty.setup.ServletEnvironment import org.apache.texera.auth.RoleAnnotationEnforcer +import org.apache.texera.common.config.StorageConfig import org.apache.texera.service.resource.{ AdminComputingUnitResource, ComputingUnitAccessResource, @@ -36,8 +37,26 @@ import org.mockito.Mockito.{mock, verify, when} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.sql.DriverManager + class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { + // run() opens the real HikariCP pool via SqlServer.initConnection, so it is only + // exercisable where Postgres is provisioned at the configured JDBC URL (as in CI). + private def databaseReachable: Boolean = + try { + DriverManager + .getConnection( + StorageConfig.jdbcUrl, + StorageConfig.jdbcUsername, + StorageConfig.jdbcPassword + ) + .close() + true + } catch { + case _: Throwable => false + } + // Every endpoint this service registers declares @RolesAllowed/@PermitAll/@DenyAll. "ComputingUnitManagingService's registered resources" should "all declare access control" in { RoleAnnotationEnforcer.findUnannotatedEndpoints( @@ -60,8 +79,12 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { when(env.getApplicationContext).thenReturn(context) when(jersey.getResourceConfig).thenReturn(DropwizardResourceConfig.forTesting()) - val service = new ComputingUnitManagingService - service.run(mock(classOf[ComputingUnitManagingServiceConfiguration]), env) + assume( + databaseReachable, + "run() requires a reachable Postgres at the configured JDBC URL (provided in CI)" + ) + new ComputingUnitManagingService() + .run(mock(classOf[ComputingUnitManagingServiceConfiguration]), env) verify(jersey).register(isA(classOf[ComputingUnitManagingResource])) verify(jersey).register(isA(classOf[ComputingUnitAccessResource])) From cf643ca9ce14fb9d1672ae09eb05fee281ad2e41 Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Fri, 17 Jul 2026 23:30:05 -0400 Subject: [PATCH 6/8] test(computing-unit-managing-service): tidy up CU listing test fixtures Small readability cleanups in the new specs (no behavior/coverage change): - KubernetesClientSpec: extract a pod(cuid, phase) helper, collapsing the duplicated PodBuilder blocks (list items + the present-pod stub). - ComputingUnitHelpersSpec: route localUnit/kubernetesUnit through a shared makeUnit(cuid, uid, type) (matching AdminComputingUnitResourceSpec), and hoist the repeated UserDao/WorkflowComputingUnitDao construction into lazy vals. --- .../util/ComputingUnitHelpersSpec.scala | 37 +++++++++--------- .../service/util/KubernetesClientSpec.scala | 39 ++++++------------- 2 files changed, 31 insertions(+), 45 deletions(-) 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 97cf94099c9..43bc8e9a66f 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 @@ -43,6 +43,9 @@ class ComputingUnitHelpersSpec override protected def afterAll(): Unit = shutdownDB() + private lazy val userDao = new UserDao(getDSLContext.configuration()) + private lazy val computingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + private def makeUser(uid: Int, name: String, email: String, avatar: String): User = { val u = new User() u.setUid(uid) @@ -54,21 +57,23 @@ class ComputingUnitHelpersSpec u } - private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit = { + private def makeUnit( + cuid: Int, + uid: Int, + tpe: WorkflowComputingUnitTypeEnum + ): WorkflowComputingUnit = { val unit = new WorkflowComputingUnit() unit.setCuid(cuid) unit.setUid(uid) - unit.setType(WorkflowComputingUnitTypeEnum.local) + unit.setType(tpe) unit } - private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit = { - val unit = new WorkflowComputingUnit() - unit.setCuid(cuid) - unit.setUid(uid) - unit.setType(WorkflowComputingUnitTypeEnum.kubernetes) - unit - } + private def localUnit(cuid: Int = 0, uid: Int = 0): WorkflowComputingUnit = + makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.local) + + private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit = + makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.kubernetes) // WorkflowComputingUnitTypeEnum only defines `local` and `kubernetes`, so an // untyped unit (getType == null) is what exercises the pure "unknown" branch. @@ -218,7 +223,6 @@ class ComputingUnitHelpersSpec // ── resolveOwnerInfo (backed by the embedded database) ─────────────── "resolveOwnerInfo" should "resolve avatar/name and collapse blank values to null" in { - val userDao = new UserDao(getDSLContext.configuration()) userDao.insert(makeUser(500, "alice", "alice@example.com", "alice-avatar")) userDao.insert(makeUser(501, "", "bob@example.com", "")) @@ -228,16 +232,13 @@ class ComputingUnitHelpersSpec } it should "return an empty map (and issue no query) for no uids" in { - val userDao = new UserDao(getDSLContext.configuration()) ComputingUnitHelpers.resolveOwnerInfo(userDao, Seq.empty) shouldBe empty } // ── reconcileVanishedKubernetesUnits (backed by the embedded database) ─ "reconcileVanishedKubernetesUnits" should "terminate vanished kubernetes units and return the live ones" in { - val userDao = new UserDao(getDSLContext.configuration()) userDao.insert(makeUser(600, "carol", "carol@example.com", null)) - val dao = new WorkflowComputingUnitDao(getDSLContext.configuration()) val present = kubernetesUnit(600, 600) present.setName("present") @@ -245,20 +246,20 @@ class ComputingUnitHelpersSpec gone.setName("gone") val local = localUnit(602, 600) local.setName("local") - Seq(present, gone, local).foreach(dao.insert(_)) + Seq(present, gone, local).foreach(computingUnitDao.insert(_)) // Only the pod for cuid 600 exists; cuid 601's pod has vanished. val podPhases = Map(KubernetesClient.generatePodName(600) -> "Running") val live = ComputingUnitHelpers.reconcileVanishedKubernetesUnits( - dao, + computingUnitDao, List(present, gone, local), podPhases ) live.map(_.getCuid) should contain theSameElementsAs Seq(600, 602) - dao.fetchOneByCuid(601).getTerminateTime should not be null - dao.fetchOneByCuid(600).getTerminateTime shouldBe null - dao.fetchOneByCuid(602).getTerminateTime shouldBe null + computingUnitDao.fetchOneByCuid(601).getTerminateTime should not be null + computingUnitDao.fetchOneByCuid(600).getTerminateTime shouldBe null + computingUnitDao.fetchOneByCuid(602).getTerminateTime shouldBe null } } diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala index 3354873cf2f..865ebe6b533 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 @@ -57,28 +57,19 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter u } + private def pod(cuid: Int, phase: String): Pod = + new PodBuilder() + .withNewMetadata() + .withName(KubernetesClient.generatePodName(cuid)) + .endMetadata() + .withNewStatus() + .withPhase(phase) + .endStatus() + .build() + // cuid 1 -> Running, cuid 2 -> Pending; a namespace-wide list returns both. private val podList: PodList = - new PodListBuilder() - .addToItems( - new PodBuilder() - .withNewMetadata() - .withName(KubernetesClient.generatePodName(1)) - .endMetadata() - .withNewStatus() - .withPhase("Running") - .endStatus() - .build(), - new PodBuilder() - .withNewMetadata() - .withName(KubernetesClient.generatePodName(2)) - .endMetadata() - .withNewStatus() - .withPhase("Pending") - .endStatus() - .build() - ) - .build() + new PodListBuilder().addToItems(pod(1, "Running"), pod(2, "Pending")).build() // Only cuid 1 reports metrics. private val metricsList: PodMetricsList = @@ -113,13 +104,7 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter // withName(cuid 1) -> a present pod; withName(cuid 2) -> no pod. val presentResource = mock(classOf[PodResource]) - when(presentResource.get()).thenReturn( - new PodBuilder() - .withNewMetadata() - .withName(KubernetesClient.generatePodName(1)) - .endMetadata() - .build() - ) + when(presentResource.get()).thenReturn(pod(1, "Running")) val absentResource = mock(classOf[PodResource]) when(absentResource.get()).thenReturn(null) when(podsInNamespace.withName(KubernetesClient.generatePodName(1))).thenReturn(presentResource) From 7dbc8cacf9757b5a1f98b8ae76dd91c240e171d9 Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Sat, 18 Jul 2026 00:25:19 -0400 Subject: [PATCH 7/8] docs(computing-unit-managing-service): tighten comments on the CU listing code Trim the scaladocs and inline comments added by this PR to their essentials (keeping the non-obvious "why": O(1) round trips, shared-by-both-endpoints, null-collapse, pure/testable, survivors-only metrics). No code changes. --- .../resource/AdminComputingUnitResource.scala | 37 +++---------- .../ComputingUnitManagingResource.scala | 14 ++--- .../service/util/ComputingUnitHelpers.scala | 52 ++++++------------- .../service/util/KubernetesClient.scala | 23 +++----- .../AdminComputingUnitResourceSpec.scala | 3 +- .../util/ComputingUnitHelpersSpec.scala | 3 +- .../service/util/KubernetesClientSpec.scala | 3 +- 7 files changed, 37 insertions(+), 98 deletions(-) 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 173908cdc3e..646622eb2af 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 @@ -42,17 +42,8 @@ object AdminComputingUnitResource { .createDSLContext() /** - * Assemble the admin dashboard rows for a set of active computing units. The `units` are - * expected to already be the active (non-terminated, pod-still-present) set; this is a pure - * mapping and does no filtering of its own. Every row is marked with WRITE access — admins - * have full control over every unit they can see — and `isOwner` reflects the requesting - * admin. Kubernetes status/metrics are resolved from the pre-fetched maps (no per-unit call). - * - * @param units active computing units to render (across every owner) - * @param ownerInfo map of owner uid -> (googleAvatar, userName) - * @param callerUid the uid of the requesting admin, used to populate `isOwner` - * @param podPhases map of pod name -> phase (see KubernetesClient.getAllPodPhases) - * @param podMetrics map of pod name -> (metric -> value) (see KubernetesClient.getAllPodMetrics) + * 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], @@ -81,18 +72,9 @@ class AdminComputingUnitResource { import AdminComputingUnitResource._ /** - * List every non-terminated computing unit across all users. ADMIN-only. - * - * Mirrors the reconciliation done by the per-user listing endpoint: a Kubernetes unit whose - * pod has vanished (manually deleted or TTL GC-ed by the cluster) is eagerly marked - * terminated in the database and excluded from the response, so ghost units do not - * accumulate in the admin view. - * - * Kubernetes status/metrics are resolved from a single namespace-wide `list`/`top` call each, - * so the number of cluster round trips is constant rather than proportional to the number of - * units. - * - * @return the computing units (owned by any user) that are active and whose pods still exist. + * 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 `list`/`top` each (O(1) round trips). */ @GET @Produces(Array(MediaType.APPLICATION_JSON)) @@ -102,7 +84,7 @@ class AdminComputingUnitResource { ): List[DashboardWorkflowComputingUnit] = { val ctx = context - // Filter to active units in SQL so historically-terminated rows are never loaded. + // Filter to active units in SQL so terminated rows are never loaded. val activeUnits: List[WorkflowComputingUnit] = ctx .selectFrom(WORKFLOW_COMPUTING_UNIT) @@ -111,19 +93,16 @@ class AdminComputingUnitResource { .asScala .toList - // Pod phases (one namespace-wide `list`) are needed to decide which units are still alive; - // only fetched when there is a Kubernetes unit to resolve. + // Pod phases decide which Kubernetes units are still alive. val podPhases = ComputingUnitHelpers.podPhasesFor(activeUnits) - // A Kubernetes unit whose pod is gone is stamped terminated and dropped from the response. val liveUnits = ComputingUnitHelpers.reconcileVanishedKubernetesUnits( new WorkflowComputingUnitDao(ctx.configuration()), activeUnits, podPhases ) - // Metrics (one namespace-wide `top`) are only rendered for surviving units, so this is - // deferred until after reconciliation and skipped when no live Kubernetes unit remains. + // Metrics only for survivors, so fetch after reconciliation. val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) val userDao = new UserDao(ctx.configuration()) diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala index 871ff1b89d3..df67fb3bfe0 100644 --- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala @@ -495,9 +495,8 @@ class ComputingUnitManagingResource { val userDao = new UserDao(ctx.configuration()) - // Pair each unit with the caller's privilege (owned units default to WRITE), keeping only - // one row per cuid. Do this first so we reconcile/render each unit exactly once even when a - // unit is both owned and shared. + // 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)))) @@ -508,13 +507,9 @@ class ComputingUnitManagingResource { }.toMap val candidateUnits = unitsWithPrivilege.map { case (unit, _) => unit } - // Pod phases (one namespace-wide `list`) decide which Kubernetes units are still alive; - // only fetched when there is a Kubernetes unit to resolve. + // Pod phases decide which Kubernetes units are still alive. val podPhases = ComputingUnitHelpers.podPhasesFor(candidateUnits) - // A Kubernetes unit whose pod has vanished (manually deleted or TTL GC-ed by the cluster) - // is treated as terminated: its terminateTime is persisted and it is dropped here so - // subsequent API calls no longer return it. val liveUnits = ComputingUnitHelpers.reconcileVanishedKubernetesUnits( computingUnitDao, @@ -522,8 +517,7 @@ class ComputingUnitManagingResource { podPhases ) - // Metrics (one namespace-wide `top`) are only rendered for surviving units, so this is - // deferred until after reconciliation and skipped when no live Kubernetes unit remains. + // Metrics only for survivors, so fetch after reconciliation. val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) val ownerInfoMap = 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 827f92691ff..e6095723bf6 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 @@ -34,9 +34,8 @@ import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava} object ComputingUnitHelpers { /** - * Resolve owner display info for the given owner uids in a single `fetchByUid` call, - * keyed by uid. Blank avatars/names collapse to `null` so callers can pass them straight - * into the dashboard shape. Returns an empty map when `uids` is empty (no query issued). + * Owner (avatar, name) keyed by uid, from one `fetchByUid`. Blank values collapse to `null`; + * an empty `uids` returns empty without querying. */ def resolveOwnerInfo( userDao: UserDao, @@ -91,10 +90,9 @@ object ComputingUnitHelpers { } /** - * Bulk variant of [[getComputingUnitStatus]]: resolves a unit's status from a pre-fetched - * map of pod name -> phase (see [[KubernetesClient.getAllPodPhases]]) instead of issuing a - * per-unit Kubernetes call. Used by listings that resolve many units at once so the number - * of cluster round trips is O(1) rather than O(units). + * Like [[getComputingUnitStatus]] but reads status from a pre-fetched pod-phase map + * (see [[KubernetesClient.getAllPodPhases]]), so a listing costs O(1) cluster round trips + * rather than one per unit. */ def getComputingUnitStatus( unit: WorkflowComputingUnit, @@ -104,7 +102,7 @@ object ComputingUnitHelpers { case WorkflowComputingUnitTypeEnum.local => Running case WorkflowComputingUnitTypeEnum.kubernetes => - // A missing entry or a null phase both yield `false` here (null != "Running"). + // A missing entry or null phase both count as not-Running. if (podPhases.get(KubernetesClient.generatePodName(unit.getCuid)).contains("Running")) Running else Pending @@ -114,9 +112,8 @@ object ComputingUnitHelpers { } /** - * Bulk variant of [[getComputingUnitMetrics]]: resolves a unit's metrics from a pre-fetched - * map of pod name -> (metric -> value) (see [[KubernetesClient.getAllPodMetrics]]) instead of - * re-fetching the whole namespace metrics list per unit. + * Like [[getComputingUnitMetrics]] but reads from a pre-fetched pod-metrics map + * (see [[KubernetesClient.getAllPodMetrics]]) instead of a per-unit fetch. */ def getComputingUnitMetrics( unit: WorkflowComputingUnit, @@ -143,32 +140,19 @@ object ComputingUnitHelpers { case _ => false } - /** - * Fetch namespace-wide pod phases (one `list`) only when `units` contains a Kubernetes unit; - * otherwise return an empty map without touching the cluster. - */ + /** Namespace-wide pod phases (one `list`), but only when a Kubernetes unit is present. */ def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] = if (units.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty - /** - * Fetch namespace-wide pod metrics (one `top`) only when `units` contains a Kubernetes unit; - * otherwise return an empty map without touching the cluster. - */ + /** Namespace-wide pod metrics (one `top`), but only when a Kubernetes unit is present. */ def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] = if (units.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty - /** - * A Kubernetes unit is considered "vanished" when its pod is absent from the pre-fetched - * `podPhases` map (manually deleted or TTL GC-ed by the cluster). Local/other units always - * count as present. - */ + /** A Kubernetes unit whose pod is absent from `podPhases` (deleted or TTL GC-ed). */ private def isVanished(unit: WorkflowComputingUnit, podPhases: Map[String, String]): Boolean = isKubernetes(unit) && !podPhases.contains(KubernetesClient.generatePodName(unit.getCuid)) - /** - * Split `units` into `(live, vanished)` using a pre-fetched pod-phase map (see - * [[KubernetesClient.getAllPodPhases]]). Pure — does no I/O — so it can be unit-tested. - */ + /** Partition into `(live, vanished)` by `podPhases`. Pure (no I/O), so it is unit-testable. */ def partitionLiveUnits( units: List[WorkflowComputingUnit], podPhases: Map[String, String] @@ -176,10 +160,8 @@ object ComputingUnitHelpers { units.partition(unit => !isVanished(unit, podPhases)) /** - * Reconcile a set of units against the cluster: any Kubernetes unit whose pod has vanished is - * stamped with `terminateTime` and persisted in a single batched update, then the surviving - * (live) units are returned. Shared by the per-user and admin listing endpoints so both agree - * on when a unit is treated as terminated. + * Stamp `terminateTime` on vanished Kubernetes units (one batched update) and return the live + * ones. Shared by both listing endpoints so they agree on when a unit is terminated. */ def reconcileVanishedKubernetesUnits( dao: WorkflowComputingUnitDao, @@ -197,10 +179,8 @@ object ComputingUnitHelpers { } /** - * Assemble a single dashboard row from a unit, its caller-relative ownership/privilege, and - * the pre-fetched owner-info/pod maps. Kubernetes status and metrics are resolved from the - * maps (no per-unit Kubernetes call). Shared by both listing endpoints so the row shape and - * owner-info fallback stay identical. + * Build one dashboard row; status/metrics come from the pre-fetched maps (no per-unit K8s + * call). Shared by both listing endpoints so row shape and owner-info fallback stay identical. */ def buildDashboardUnit( unit: WorkflowComputingUnit, 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 64f05fffef2..129b79a720e 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 @@ -28,15 +28,11 @@ import scala.jdk.CollectionConverters._ object KubernetesClient { - // Initialize the Kubernetes client private var client: io.fabric8.kubernetes.client.KubernetesClient = new KubernetesClientBuilder().build() private val namespace: String = KubernetesConfig.computeUnitPoolNamespace - /** - * Test seam: swap in a stubbed client so the pod/metrics wrappers can be exercised without a - * live cluster. Not used in production code. - */ + /** Test-only seam to swap in a stubbed client (exercises the wrappers without a live cluster). */ private[util] def setClientForTesting( testClient: io.fabric8.kubernetes.client.KubernetesClient ): Unit = @@ -58,15 +54,9 @@ object KubernetesClient { } /** - * Fetch the phase of every pod in the namespace in a single list call, keyed by pod name. - * Intended for bulk listings (e.g. the admin view) so that N units do not trigger N separate - * `getPodByName` round trips. Left unfiltered (rather than label-scoped) to match the - * name-based existence semantics of `getPodByName`/`podExists`: a caller checks - * `contains(generatePodName(cuid))` to decide whether a unit's pod still exists, and the - * `computing-unit-` names never collide with unrelated pods. - * - * A pod that exists but has no status yet maps to a `null` phase; callers can still rely - * on `contains(podName)` to decide whether the pod exists at all. + * Phase of every namespace pod in one `list`, keyed by pod name — so a bulk listing avoids N + * `getPodByName` round trips. Unfiltered to match the name-based existence check callers use + * (`contains(generatePodName(cuid))`); a status-less pod maps to a `null` phase but still appears. */ def getAllPodPhases: Map[String, String] = { client @@ -92,9 +82,8 @@ object KubernetesClient { client.top().pods().metrics(namespace).getItems.asScala /** - * Fetch CPU/memory usage for every pod in the namespace in a single `top` call, keyed by - * pod name. Intended for bulk listings so that N units do not each re-fetch the whole - * namespace metrics list (which `getPodMetrics` does per call). + * CPU/memory of every namespace pod in one `top`, keyed by pod name — the bulk counterpart to + * `getPodMetrics`, which re-fetches the whole list per call. */ def getAllPodMetrics: Map[String, Map[String, String]] = 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 aaee56f4637..0dfa1b4da94 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 @@ -77,8 +77,7 @@ class AdminComputingUnitResourceSpec private def kubernetesUnit(cuid: Int, uid: Int, name: String): WorkflowComputingUnit = makeUnit(cuid, uid, name, WorkflowComputingUnitTypeEnum.kubernetes) - // The class-level @RolesAllowed(Array("ADMIN")) is what makes Jersey's - // RolesAllowedDynamicFeature return 403 for any non-ADMIN (e.g. REGULAR) caller. + // The class-level @RolesAllowed(ADMIN) is what makes Jersey reject non-ADMIN callers. "AdminComputingUnitResource" should "only permit the ADMIN role (non-ADMIN callers are rejected)" in { val annotation = classOf[AdminComputingUnitResource].getAnnotation(classOf[RolesAllowed]) annotation should not be null 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 43bc8e9a66f..73deffb4eb2 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 @@ -75,8 +75,7 @@ class ComputingUnitHelpersSpec private def kubernetesUnit(cuid: Int, uid: Int = 0): WorkflowComputingUnit = makeUnit(cuid, uid, WorkflowComputingUnitTypeEnum.kubernetes) - // WorkflowComputingUnitTypeEnum only defines `local` and `kubernetes`, so an - // untyped unit (getType == null) is what exercises the pure "unknown" branch. + // A null-type unit (the enum has only local/kubernetes) exercises the "unknown" branch. private def untypedUnit(): WorkflowComputingUnit = new WorkflowComputingUnit() "getComputingUnitStatus" should "return Running for a local unit" 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 865ebe6b533..9a8596e6082 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 @@ -44,8 +44,7 @@ import org.scalatest.matchers.should.Matchers import scala.jdk.CollectionConverters._ -// Exercises the fabric8 wrappers without a cluster by injecting a client whose -// fluent pods()/top() chains are stubbed with Mockito. +// Exercises the fabric8 wrappers without a cluster by stubbing the client with Mockito. class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll { private val namespace: String = KubernetesConfig.computeUnitPoolNamespace From 7f5e0ac0a4b63aa05a7b459ae423cb8905095fbf Mon Sep 17 00:00:00 2001 From: Kunwoo Park Date: Sat, 18 Jul 2026 10:10:51 -0400 Subject: [PATCH 8/8] docs(computing-unit-managing-service): drop method names from comments Rephrase the CU-listing comments to describe behavior instead of naming methods (getComputingUnitStatus/getAllPodPhases/getPodMetrics/getPodByName, etc.), so they don't drift when those methods are renamed. Wording only; the non-obvious rationale (O(1) round trips, pre-fetched maps, presence-by- pod-name, DB-gated run test) is preserved. --- .../resource/AdminComputingUnitResource.scala | 2 +- .../service/util/ComputingUnitHelpers.scala | 18 +++++++----------- .../texera/service/util/KubernetesClient.scala | 12 ++++++------ .../ComputingUnitManagingServiceRunSpec.scala | 4 ++-- .../service/util/KubernetesClientSpec.scala | 4 ++-- 5 files changed, 18 insertions(+), 22 deletions(-) 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 646622eb2af..e08cd78b004 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 @@ -74,7 +74,7 @@ class 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 `list`/`top` each (O(1) round trips). + * units don't accumulate. Status/metrics use one namespace-wide call each (O(1) round trips). */ @GET @Produces(Array(MediaType.APPLICATION_JSON)) 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 e6095723bf6..e877d0209ff 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 @@ -34,8 +34,8 @@ import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava} object ComputingUnitHelpers { /** - * Owner (avatar, name) keyed by uid, from one `fetchByUid`. Blank values collapse to `null`; - * an empty `uids` returns empty without querying. + * Owner (avatar, name) keyed by uid, resolved in one query. Blank values collapse to `null`; + * empty `uids` returns empty without querying. */ def resolveOwnerInfo( userDao: UserDao, @@ -90,9 +90,8 @@ object ComputingUnitHelpers { } /** - * Like [[getComputingUnitStatus]] but reads status from a pre-fetched pod-phase map - * (see [[KubernetesClient.getAllPodPhases]]), so a listing costs O(1) cluster round trips - * rather than one per unit. + * Resolves status from a pre-fetched pod-phase map instead of a per-unit cluster call, so a + * listing costs O(1) round trips rather than one per unit. */ def getComputingUnitStatus( unit: WorkflowComputingUnit, @@ -111,10 +110,7 @@ object ComputingUnitHelpers { } } - /** - * Like [[getComputingUnitMetrics]] but reads from a pre-fetched pod-metrics map - * (see [[KubernetesClient.getAllPodMetrics]]) instead of a per-unit fetch. - */ + /** Resolves metrics from a pre-fetched pod-metrics map instead of a per-unit cluster call. */ def getComputingUnitMetrics( unit: WorkflowComputingUnit, podMetrics: Map[String, Map[String, String]] @@ -140,11 +136,11 @@ object ComputingUnitHelpers { case _ => false } - /** Namespace-wide pod phases (one `list`), but only when a Kubernetes unit is present. */ + /** Pod phases for the namespace; skipped (empty) when no Kubernetes unit is present. */ def podPhasesFor(units: Seq[WorkflowComputingUnit]): Map[String, String] = if (units.exists(isKubernetes)) KubernetesClient.getAllPodPhases else Map.empty - /** Namespace-wide pod metrics (one `top`), but only when a Kubernetes unit is present. */ + /** Pod metrics for the namespace; skipped (empty) when no Kubernetes unit is present. */ def podMetricsFor(units: Seq[WorkflowComputingUnit]): Map[String, Map[String, String]] = if (units.exists(isKubernetes)) KubernetesClient.getAllPodMetrics else Map.empty 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 129b79a720e..2f75c3f1562 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 @@ -54,9 +54,9 @@ object KubernetesClient { } /** - * Phase of every namespace pod in one `list`, keyed by pod name — so a bulk listing avoids N - * `getPodByName` round trips. Unfiltered to match the name-based existence check callers use - * (`contains(generatePodName(cuid))`); a status-less pod maps to a `null` phase but still appears. + * Phase of every pod in the namespace, keyed by pod name, in one call — so a bulk listing + * avoids a per-unit lookup. Unfiltered so callers can test a unit's presence by its pod-name + * key; a pod with no status yet maps to a `null` phase but still appears. */ def getAllPodPhases: Map[String, String] = { client @@ -77,13 +77,13 @@ object KubernetesClient { } }.toMap - // One namespace-wide `top` call, returning the raw per-pod metrics items. + // One namespace-wide metrics call, returning the raw per-pod items. private def fetchPodMetricsItems(): Iterable[PodMetrics] = client.top().pods().metrics(namespace).getItems.asScala /** - * CPU/memory of every namespace pod in one `top`, keyed by pod name — the bulk counterpart to - * `getPodMetrics`, which re-fetches the whole list per call. + * CPU/memory of every pod in the namespace, keyed by pod name, in one call — the bulk + * counterpart to the single-unit lookup. */ def getAllPodMetrics: Map[String, Map[String, String]] = fetchPodMetricsItems() diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala index 0835d064d72..c6712e9a815 100644 --- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/ComputingUnitManagingServiceRunSpec.scala @@ -41,8 +41,8 @@ import java.sql.DriverManager class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { - // run() opens the real HikariCP pool via SqlServer.initConnection, so it is only - // exercisable where Postgres is provisioned at the configured JDBC URL (as in CI). + // Booting the service opens a real connection pool, so this path is only exercisable where + // Postgres is provisioned at the configured JDBC URL (as in CI). private def databaseReachable: Boolean = try { DriverManager 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 9a8596e6082..fc98727e7bc 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 @@ -66,7 +66,7 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter .endStatus() .build() - // cuid 1 -> Running, cuid 2 -> Pending; a namespace-wide list returns both. + // cuid 1 -> Running, cuid 2 -> Pending; both returned by the namespace-wide listing. private val podList: PodList = new PodListBuilder().addToItems(pod(1, "Running"), pod(2, "Pending")).build() @@ -101,7 +101,7 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers with BeforeAndAfter when(podsMixed.inNamespace(namespace)).thenReturn(podsInNamespace) when(podsInNamespace.list()).thenReturn(podList) - // withName(cuid 1) -> a present pod; withName(cuid 2) -> no pod. + // cuid 1 -> a present pod; cuid 2 -> no pod. val presentResource = mock(classOf[PodResource]) when(presentResource.get()).thenReturn(pod(1, "Running")) val absentResource = mock(classOf[PodResource])