This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6854-0f2cd49c39781d683f35e02547da45fbe5adc3c3 in repository https://gitbox.apache.org/repos/asf/texera.git
commit b1d51ae2438d7d5e78ce5d3a226eec8c87c21e9d Author: Kunwoo (Chris) <[email protected]> AuthorDate: Wed Aug 5 13:29:57 2026 -0400 feat(computing-unit-managing-service): add admin endpoint to list all computing units (#6854) ### What changes were proposed in this PR? Adds an ADMIN-only endpoint that returns every non-terminated computing unit across all users. Needed for the admin Computing Units dashboard. The endpoint queries all units via `WorkflowComputingUnitDao`, excludes rows with a non-null `terminate_time`, joins `UserDao` for each owner's name and avatar, and reuses the shared `ComputingUnitHelpers` (introduced in the preceding refactor PR) to reconcile vanished pods and resolve status/metrics. The new resource is registered in `ComputingUnitManagingService`, and the startup access-control coverage check is extended to include it. `accessPrivilege` is reported as `WRITE` for admin rows (an admin can manage any unit it can see), and `isOwner` reflects whether the requesting admin happens to own the unit. Note that the mutating endpoints gate on ownership alone and have no ADMIN bypass yet, so a client must not present these rows as writable until they do — recorded in the resource's scaladoc. > 📌 **#6853 has landed and this is rebased onto it.** The diff is now just the admin resource + registration + specs, in a single commit. ### Any related issues, documentation, discussions? Closes #6477. Part of #6476 (Admin Computing Units Dashboard). Originates from discussion #6322. ### How was this PR tested? Added `AdminComputingUnitResourceSpec` (drives `listAllComputingUnits` end-to-end over the embedded DB: terminated rows excluded, `WRITE` access, `isOwner`, an empty table, and a caller owning none of the listed units) and extended `ComputingUnitManagingServiceRunSpec` to verify the admin resource registers along with the auth features that make `@RolesAllowed` enforceable. Verified on the rebased commit by CI's `platform (computing-unit-managing-service)` job, which runs `ComputingUnitManagingService/test` and the scalafmt checks: 79 tests, 0 failed, 0 canceled (Postgres is reachable there, so the DB-gated `run()` registration test executes rather than cancelling). ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code, Claude Opus 4.8 --- .../service/ComputingUnitManagingService.scala | 2 + .../resource/AdminComputingUnitResource.scala | 111 +++++++++++++++ .../ComputingUnitManagingServiceRunSpec.scala | 65 +++++++++ .../resource/AdminComputingUnitResourceSpec.scala | 158 +++++++++++++++++++++ 4 files changed, 336 insertions(+) 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 db63bbf2eb..f0dffc89e1 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 0000000000..45d137a4b4 --- /dev/null +++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/AdminComputingUnitResource.scala @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import io.dropwizard.auth.Auth +import jakarta.annotation.security.RolesAllowed +import jakarta.ws.rs.{GET, Path, Produces} +import jakarta.ws.rs.core.MediaType +import org.apache.texera.auth.SessionUser +import org.apache.texera.dao.SqlServer +import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_COMPUTING_UNIT +import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.service.resource.ComputingUnitManagingResource.DashboardWorkflowComputingUnit +import org.apache.texera.service.util.ComputingUnitHelpers +import org.jooq.DSLContext + +import scala.jdk.CollectionConverters.CollectionHasAsScala + +object AdminComputingUnitResource { + private def context: DSLContext = + SqlServer + .getInstance() + .createDSLContext() +} + +// Path is /computing-unit/admin, not the /admin/<domain> shape other admin resources use: the +// gateway only routes /api/computing-unit here, so /api/admin/... never reaches this service. +// Jersey matches the literal /admin/list ahead of ComputingUnitManagingResource's @Path("/{cuid}"), +// so don't add another two-segment literal path under /computing-unit without re-checking this. +@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). + * + * TODO: rows report WRITE, but the mutating endpoints on [[ComputingUnitManagingResource]] + * still gate on ownership with no ADMIN bypass, so an admin acting on a unit it does not own is + * rejected. Add that bypass; until then a client must not present these rows as writable. + */ + @GET + @Path("/list") + def listAllComputingUnits( + @Auth user: SessionUser + ): List[DashboardWorkflowComputingUnit] = { + val ctx = context + + // Filter active units in SQL. Explicit `IS NULL`, not the DAO's fetchByTerminateTime(null), + // which renders `terminate_time IN (null)` and matches nothing. + val activeUnits = + ctx + .selectFrom(WORKFLOW_COMPUTING_UNIT) + .where(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME.isNull) + .fetchInto(classOf[WorkflowComputingUnit]) + .asScala + .toList + + val podPhases = ComputingUnitHelpers.podPhasesFor(activeUnits) + + // Wrap only the reconcile write, not the Kubernetes round trips above: keeps a pooled + // connection from being held open across them, while the transaction makes the batchUpdate + // all-or-nothing (autocommit would retire only some vanished units on a mid-batch failure). + + val liveUnits = SqlServer.withTransaction(ctx) { txCtx => + ComputingUnitHelpers.reconcileVanishedKubernetesUnits( + new WorkflowComputingUnitDao(txCtx.configuration()), + activeUnits, + podPhases + ) + } + + // Metrics only for survivors, so fetch after reconciliation. + val podMetrics = ComputingUnitHelpers.podMetricsFor(liveUnits) + + val userDao = new UserDao(ctx.configuration()) + val ownerInfo = ComputingUnitHelpers.resolveOwnerInfo(userDao, liveUnits.map(_.getUid).distinct) + + liveUnits.map { unit => + ComputingUnitHelpers.buildDashboardUnit( + unit, + isOwner = unit.getUid.equals(user.getUid), + accessPrivilege = PrivilegeEnum.WRITE, + ownerInfo = ownerInfo, + podPhases = podPhases, + podMetrics = podMetrics + ) + } + } +} 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 d2162d48c7..e4694e2d20 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,25 +19,90 @@ package org.apache.texera.service +import io.dropwizard.auth.AuthDynamicFeature +import io.dropwizard.core.setup.Environment +import io.dropwizard.jersey.DropwizardResourceConfig +import io.dropwizard.jersey.setup.JerseyEnvironment +import io.dropwizard.jetty.MutableServletContextHandler import org.apache.texera.auth.RoleAnnotationEnforcer +import org.apache.texera.common.config.StorageConfig +import org.apache.texera.dao.SqlServer import org.apache.texera.service.resource.{ + AdminComputingUnitResource, ComputingUnitAccessResource, ComputingUnitManagingResource, HealthCheckResource } +import org.glassfish.jersey.server.filter.RolesAllowedDynamicFeature +import org.mockito.ArgumentMatchers.isA +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 { + // 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 + .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( Seq( classOf[ComputingUnitManagingResource], classOf[ComputingUnitAccessResource], + classOf[AdminComputingUnitResource], classOf[HealthCheckResource] ) ) shouldBe empty } + + "ComputingUnitManagingService.run" should "register the admin resource on the Jersey environment" in { + assume( + databaseReachable, + "run() requires a reachable Postgres at the configured JDBC URL (provided in CI)" + ) + + val jersey = mock(classOf[JerseyEnvironment]) + val context = mock(classOf[MutableServletContextHandler]) + val env = mock(classOf[Environment]) + when(env.jersey).thenReturn(jersey) + when(env.getApplicationContext).thenReturn(context) + when(jersey.getResourceConfig).thenReturn(DropwizardResourceConfig.forTesting()) + + try { + new ComputingUnitManagingService() + .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/*") + // Without these two, Jersey never enforces AdminComputingUnitResource's + // @RolesAllowed(ADMIN) and the cross-user listing is readable by any authenticated user. + verify(jersey).register(isA(classOf[AuthDynamicFeature])) + verify(jersey).register(classOf[RolesAllowedDynamicFeature]) + } finally { + // run() calls SqlServer.initConnection, which opens a real HikariCP pool against the + // configured JDBC URL. Close it so the pool's threads/connections don't outlive this suite + // in the shared forked JVM. Swallowed because run() may have failed before initConnection, + // and a throw here would mask that failure. + try SqlServer.getInstance().close() + catch { case _: Throwable => () } + } + } } 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 0000000000..a721986b69 --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala @@ -0,0 +1,158 @@ +/* + * 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.auth.SessionUser +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_COMPUTING_UNIT +import org.apache.texera.dao.jooq.generated.enums.{ + 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.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.sql.Timestamp + +class AdminComputingUnitResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with BeforeAndAfterEach + with MockTexeraDB { + + override protected def beforeAll(): Unit = { + super.beforeAll() + initializeDBAndReplaceDSLContext() + } + + // MockTexeraDB does not truncate between tests, and this endpoint lists every unit in the + // database, so each case must start from a known-empty table — otherwise the exact-match + // assertions below depend on the order ScalaTest happens to run the cases in. Safe to wipe + // unqualified: MockTexeraDB provisions a fresh database per suite, so these are the only rows. + override protected def beforeEach(): Unit = { + super.beforeEach() + getDSLContext.deleteFrom(WORKFLOW_COMPUTING_UNIT).execute() + } + + override protected def afterAll(): Unit = + try shutdownDB() + finally super.afterAll() + + private def makeUser(uid: Int, name: String): User = { + val u = new User() + u.setUid(uid) + u.setName(name) + u.setEmail(s"[email protected]") + u.setRole(UserRoleEnum.ADMIN) + u.setPassword("password") + u.setGoogleAvatar(s"avatar-$uid") + u + } + + // Only local units are used here: a kubernetes unit would make the listing reach for the real + // KubernetesClient singleton, and its stub seam is private[util]. + private def localUnit(cuid: Int, uid: Int, name: String): WorkflowComputingUnit = { + val u = new WorkflowComputingUnit() + u.setCuid(cuid) + u.setUid(uid) + u.setName(name) + u.setType(WorkflowComputingUnitTypeEnum.local) + u + } + + // The class-level @RolesAllowed(ADMIN) is what makes Jersey reject non-ADMIN callers; that this + // spec calls the method directly is exactly why the annotation has to be asserted here. The + // enforcement half (Jersey actually honouring it) is pinned by + // ComputingUnitManagingServiceRunSpec, which verifies RolesAllowedDynamicFeature is registered. + "AdminComputingUnitResource" should "declare @RolesAllowed(ADMIN) at the class level" in { + val annotation = classOf[AdminComputingUnitResource].getAnnotation(classOf[RolesAllowed]) + annotation should not be null + annotation.value.toSeq shouldBe Seq("ADMIN") + } + + "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 + // isOwner tracks the caller; owner name/avatar are joined from the user table. + byCuid(700).isOwner shouldBe true + byCuid(700).ownerName shouldBe "admin" + byCuid(700).ownerGoogleAvatar shouldBe "avatar-700" + byCuid(701).isOwner shouldBe false + byCuid(701).ownerName shouldBe "other" + } + + it should "return an empty list when every unit is terminated" in { + val userDao = new UserDao(getDSLContext.configuration()) + val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + val admin = makeUser(710, "lonely-admin") + userDao.insert(admin) + val terminated = localUnit(cuid = 710, uid = 710, name = "only-terminated-cu") + terminated.setTerminateTime(new Timestamp(0L)) + unitDao.insert(terminated) + + new AdminComputingUnitResource().listAllComputingUnits(new SessionUser(admin)) shouldBe empty + } + + it should "return an empty list when no unit exists at all" in { + val admin = makeUser(720, "empty-admin") + new UserDao(getDSLContext.configuration()).insert(admin) + + new AdminComputingUnitResource().listAllComputingUnits(new SessionUser(admin)) shouldBe empty + } + + // A caller whose own user row was deleted (or who is an admin from another realm) still sees + // every unit, but owns none of them — isOwner must not accidentally default to true. + it should "mark no row as owned when the caller owns none of the units" in { + val userDao = new UserDao(getDSLContext.configuration()) + val unitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + val admin = makeUser(730, "outsider-admin") + userDao.insert(admin) + userDao.insert(makeUser(731, "owner")) + unitDao.insert(localUnit(cuid = 731, uid = 731, name = "someone-elses-cu")) + + val result = new AdminComputingUnitResource().listAllComputingUnits(new SessionUser(admin)) + + result should have size 1 + result.head.isOwner shouldBe false + result.head.ownerName shouldBe "owner" + } +}
