You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/05 17:30:19 UTC
(texera) branch gh-readonly-queue/main/pr-6854-0f2cd49c39781d683f35e02547da45fbe5adc3c3 created (now b1d51ae243)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6854-0f2cd49c39781d683f35e02547da45fbe5adc3c3
in repository https://gitbox.apache.org/repos/asf/texera.git
at b1d51ae243 feat(computing-unit-managing-service): add admin endpoint to list all computing units (#6854)
This branch includes the following new commits:
new b1d51ae243 feat(computing-unit-managing-service): add admin endpoint to list all computing units (#6854)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: feat(computing-unit-managing-service): add admin endpoint to list all computing units (#6854)
Posted by gi...@apache.org.
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) <14...@users.noreply.github.com>
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"user$uid@example.com")
+ 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"
+ }
+}