From 797849d57174980f9529b319640fa5034b322193 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Wed, 1 Jul 2026 14:22:24 +0800 Subject: [PATCH 01/10] fix(kubernetes): terminate idle computing units --- .../config/src/main/resources/kubernetes.conf | 7 ++ .../common/config/KubernetesConfig.scala | 6 ++ .../ComputingUnitManagingService.scala | 29 +++++++- .../ComputingUnitManagingResource.scala | 68 +++++++++++++++++++ 4 files changed, 109 insertions(+), 1 deletion(-) diff --git a/common/config/src/main/resources/kubernetes.conf b/common/config/src/main/resources/kubernetes.conf index e85924e570c..b45c6ffbfeb 100644 --- a/common/config/src/main/resources/kubernetes.conf +++ b/common/config/src/main/resources/kubernetes.conf @@ -41,6 +41,13 @@ kubernetes { max-num-of-running-computing-units-per-user = 10 max-num-of-running-computing-units-per-user = ${?MAX_NUM_OF_RUNNING_COMPUTING_UNITS_PER_USER} + # Terminate Kubernetes CUs whose latest workflow execution is older than this. + computing-unit-idle-timeout-minutes = 1440 + computing-unit-idle-timeout-minutes = ${?KUBERNETES_COMPUTING_UNIT_IDLE_TIMEOUT_MINUTES} + + computing-unit-idle-check-interval-minutes = 60 + computing-unit-idle-check-interval-minutes = ${?KUBERNETES_COMPUTING_UNIT_IDLE_CHECK_INTERVAL_MINUTES} + computing-unit-cpu-limit-options = "1,2,4" computing-unit-cpu-limit-options = ${?KUBERNETES_COMPUTING_UNIT_CPU_LIMIT_OPTIONS} diff --git a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala index f6294767365..65203666399 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/KubernetesConfig.scala @@ -38,6 +38,12 @@ object KubernetesConfig { val maxNumOfRunningComputingUnitsPerUser: Int = conf.getInt("kubernetes.max-num-of-running-computing-units-per-user") + val computingUnitIdleTimeoutMinutes: Long = + conf.getLong("kubernetes.computing-unit-idle-timeout-minutes") + + val computingUnitIdleCheckIntervalMinutes: Long = + conf.getLong("kubernetes.computing-unit-idle-check-interval-minutes") + val cpuLimitOptions: List[String] = conf .getString("kubernetes.computing-unit-cpu-limit-options") 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..7c753559a8b 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 @@ -23,7 +23,7 @@ import com.fasterxml.jackson.module.scala.DefaultScalaModule import io.dropwizard.configuration.{EnvironmentVariableSubstitutor, SubstitutingSourceProvider} import io.dropwizard.core.Application import io.dropwizard.core.setup.{Bootstrap, Environment} -import org.apache.texera.common.config.StorageConfig +import org.apache.texera.common.config.{KubernetesConfig, StorageConfig} import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer} import org.apache.texera.dao.SqlServer import org.apache.texera.service.resource.{ @@ -31,9 +31,12 @@ import org.apache.texera.service.resource.{ ComputingUnitManagingResource, HealthCheckResource } +import org.slf4j.LoggerFactory import java.nio.file.Path +import java.util.concurrent.TimeUnit class ComputingUnitManagingService extends Application[ComputingUnitManagingServiceConfiguration] { + private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) override def initialize( bootstrap: Bootstrap[ComputingUnitManagingServiceConfiguration] @@ -72,6 +75,30 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ "ComputingUnitManagingService" ) + if ( + KubernetesConfig.kubernetesComputingUnitEnabled && + KubernetesConfig.computingUnitIdleTimeoutMinutes > 0 + ) { + environment.lifecycle + .scheduledExecutorService("idle-computing-unit-terminator") + .threads(1) + .build() + .scheduleWithFixedDelay( + () => + try { + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits() + if (terminated > 0) { + logger.info(s"Terminated $terminated idle Kubernetes computing unit(s)") + } + } catch { + case t: Throwable => logger.warn("Failed to terminate idle Kubernetes computing units", t) + }, + KubernetesConfig.computingUnitIdleCheckIntervalMinutes, + KubernetesConfig.computingUnitIdleCheckIntervalMinutes, + TimeUnit.MINUTES + ) + } + // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL RequestLoggingFilter.register(environment.getApplicationContext) } 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..fe3f21deac9 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 @@ -42,6 +42,7 @@ import org.apache.texera.common.config.{ } import org.apache.texera.dao.SqlServer import org.apache.texera.dao.SqlServer.withTransaction +import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW_COMPUTING_UNIT, WORKFLOW_EXECUTIONS} import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} import org.apache.texera.dao.jooq.generated.tables.daos.{ ComputingUnitUserAccessDao, @@ -69,6 +70,73 @@ object ComputingUnitManagingResource { .getInstance() .createDSLContext() + def terminateIdleKubernetesComputingUnits(): Int = { + if ( + !KubernetesConfig.kubernetesComputingUnitEnabled || + KubernetesConfig.computingUnitIdleTimeoutMinutes <= 0 + ) { + return 0 + } + + val now = new Timestamp(System.currentTimeMillis()) + val cutoff = new Timestamp( + now.getTime - KubernetesConfig.computingUnitIdleTimeoutMinutes * 60 * 1000 + ) + val activeStatuses = Seq(Short.box(0), Short.box(1), Short.box(2)) + + withTransaction(context) { ctx => + val cuDao = new WorkflowComputingUnitDao(ctx.configuration()) + ctx + .selectFrom(WORKFLOW_COMPUTING_UNIT) + .where( + WORKFLOW_COMPUTING_UNIT.TYPE + .eq(WorkflowComputingUnitTypeEnum.kubernetes) + .and(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME.isNull) + ) + .fetchInto(classOf[WorkflowComputingUnit]) + .asScala + .count { unit => + val cuid = unit.getCuid + val hasActiveExecution = ctx.fetchExists( + ctx + .selectOne() + .from(WORKFLOW_EXECUTIONS) + .where( + WORKFLOW_EXECUTIONS.CUID + .eq(cuid) + .and(WORKFLOW_EXECUTIONS.STATUS.in(activeStatuses: _*)) + ) + ) + val latestUpdateTime = ctx + .select(org.jooq.impl.DSL.max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME)) + .from(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) + .fetchOne(0, classOf[Timestamp]) + val latestStartTime = ctx + .select(org.jooq.impl.DSL.max(WORKFLOW_EXECUTIONS.STARTING_TIME)) + .from(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) + .fetchOne(0, classOf[Timestamp]) + val lastExecutionTime = Seq( + Option(latestUpdateTime), + Option(latestStartTime), + Some(unit.getCreationTime) + ).flatten.maxBy(_.getTime) + + if (!hasActiveExecution && lastExecutionTime.before(cutoff)) { + if (KubernetesClient.podExists(cuid)) { + KubernetesClient.deletePod(cuid) + } + unit.setTerminateTime(now) + cuDao.update(unit) + true + } else { + false + } + } + } + } + private def icebergEnvironmentVariables: Map[String, Any] = { val base = Map[String, Any]( EnvironmentalVariable.ENV_ICEBERG_CATALOG_TYPE -> StorageConfig.icebergCatalogType From a11601fa8a23a680fd46debdcd2e0d48cefda3b7 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Sun, 5 Jul 2026 19:35:35 +0800 Subject: [PATCH 02/10] 1). add CU terminate reason. 2). return info about garbage CUs that have been closed. 3). use DSL.max instead --- .../ComputingUnitManagingService.scala | 15 ++++-- .../ComputingUnitManagingResource.scala | 50 ++++++++++++++++--- sql/texera_ddl.sql | 2 + sql/updates/28.sql | 39 +++++++++++++++ 4 files changed, 95 insertions(+), 11 deletions(-) create mode 100644 sql/updates/28.sql 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 7c753559a8b..24423d49eb3 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 @@ -87,11 +87,20 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ () => try { val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits() - if (terminated > 0) { - logger.info(s"Terminated $terminated idle Kubernetes computing unit(s)") + if (terminated.nonEmpty) { + val terminatedDetails = terminated + .map(unit => + s"cuid=${unit.cuid}, name=${unit.name}, uid=${unit.uid}, username=${unit.username + .getOrElse("unknown")}, reason=${unit.reason.getLiteral}" + ) + .mkString("; ") + logger.info( + s"Terminated ${terminated.size} idle Kubernetes computing unit(s): $terminatedDetails" + ) } } catch { - case t: Throwable => logger.warn("Failed to terminate idle Kubernetes computing units", t) + case t: Throwable => + logger.warn("Failed to terminate idle Kubernetes computing units", t) }, KubernetesConfig.computingUnitIdleCheckIntervalMinutes, KubernetesConfig.computingUnitIdleCheckIntervalMinutes, 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 fe3f21deac9..d89ccd0105d 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 @@ -43,13 +43,18 @@ import org.apache.texera.common.config.{ import org.apache.texera.dao.SqlServer import org.apache.texera.dao.SqlServer.withTransaction import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW_COMPUTING_UNIT, WORKFLOW_EXECUTIONS} -import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, WorkflowComputingUnitTypeEnum} +import org.apache.texera.dao.jooq.generated.enums.{ + PrivilegeEnum, + WorkflowComputingUnitTerminationReasonEnum, + WorkflowComputingUnitTypeEnum +} import org.apache.texera.dao.jooq.generated.tables.daos.{ ComputingUnitUserAccessDao, UserDao, WorkflowComputingUnitDao } import org.apache.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit +import org.apache.texera.service.ComputingUnitManagingService import org.apache.texera.service.resource.ComputingUnitManagingResource._ import org.apache.texera.service.resource.ComputingUnitState._ import org.apache.texera.service.util.{ @@ -58,6 +63,8 @@ import org.apache.texera.service.util.{ KubernetesClient } import org.jooq.{DSLContext, EnumType} +import org.jooq.impl.DSL.max +import org.slf4j.LoggerFactory import play.api.libs.json._ import java.sql.Timestamp @@ -70,12 +77,12 @@ object ComputingUnitManagingResource { .getInstance() .createDSLContext() - def terminateIdleKubernetesComputingUnits(): Int = { + def terminateIdleKubernetesComputingUnits(): List[TerminatedComputingUnitInfo] = { if ( !KubernetesConfig.kubernetesComputingUnitEnabled || KubernetesConfig.computingUnitIdleTimeoutMinutes <= 0 ) { - return 0 + return List.empty } val now = new Timestamp(System.currentTimeMillis()) @@ -86,6 +93,7 @@ object ComputingUnitManagingResource { withTransaction(context) { ctx => val cuDao = new WorkflowComputingUnitDao(ctx.configuration()) + val userDao = new UserDao(ctx.configuration()) ctx .selectFrom(WORKFLOW_COMPUTING_UNIT) .where( @@ -95,7 +103,7 @@ object ComputingUnitManagingResource { ) .fetchInto(classOf[WorkflowComputingUnit]) .asScala - .count { unit => + .flatMap { unit => val cuid = unit.getCuid val hasActiveExecution = ctx.fetchExists( ctx @@ -108,12 +116,12 @@ object ComputingUnitManagingResource { ) ) val latestUpdateTime = ctx - .select(org.jooq.impl.DSL.max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME)) + .select(max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME)) .from(WORKFLOW_EXECUTIONS) .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) .fetchOne(0, classOf[Timestamp]) val latestStartTime = ctx - .select(org.jooq.impl.DSL.max(WORKFLOW_EXECUTIONS.STARTING_TIME)) + .select(max(WORKFLOW_EXECUTIONS.STARTING_TIME)) .from(WORKFLOW_EXECUTIONS) .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) .fetchOne(0, classOf[Timestamp]) @@ -128,12 +136,23 @@ object ComputingUnitManagingResource { KubernetesClient.deletePod(cuid) } unit.setTerminateTime(now) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) cuDao.update(unit) - true + val owner = Option(userDao.fetchOneByUid(unit.getUid)) + Some( + TerminatedComputingUnitInfo( + cuid = unit.getCuid, + name = unit.getName, + uid = unit.getUid, + username = owner.flatMap(u => Option(u.getName).filter(_.nonEmpty)), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ) } else { - false + None } } + .toList } } @@ -195,6 +214,14 @@ object ComputingUnitManagingResource { .get ) + case class TerminatedComputingUnitInfo( + cuid: Integer, + name: String, + uid: Integer, + username: Option[String], + reason: WorkflowComputingUnitTerminationReasonEnum + ) + case class WorkflowComputingUnitCreationParams( name: String, unitType: String, @@ -241,6 +268,7 @@ object ComputingUnitManagingResource { @Produces(Array(MediaType.APPLICATION_JSON)) @Path("/computing-unit") class ComputingUnitManagingResource { + private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) private def getComputingUnitByCuid(ctx: DSLContext, cuid: Int): WorkflowComputingUnit = { val wcDao = new WorkflowComputingUnitDao(ctx.configuration()) @@ -621,6 +649,7 @@ class ComputingUnitManagingResource { !KubernetesClient.podExists(unit.getCuid) ) { unit.setTerminateTime(new Timestamp(System.currentTimeMillis())) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) computingUnitDao.update(unit) } } @@ -736,8 +765,13 @@ class ComputingUnitManagingResource { KubernetesClient.deletePod(cuid) } + val terminationReason = WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED unit.setTerminateTime(new Timestamp(System.currentTimeMillis())) + unit.setTerminationReason(terminationReason) cuDao.update(unit) + logger.info( + s"Terminated 1 Kubernetes computing unit(s): cuid=${unit.getCuid}, name=${unit.getName}, uid=${unit.getUid}, username=${user.getName}, reason=${terminationReason.getLiteral}" + ) } Response.ok().build() } diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql index 8202614932f..b59cc6ad2f3 100644 --- a/sql/texera_ddl.sql +++ b/sql/texera_ddl.sql @@ -91,6 +91,7 @@ CREATE TYPE user_role_enum AS ENUM ('INACTIVE', 'RESTRICTED', 'REGULAR', 'ADMIN' CREATE TYPE action_enum AS ENUM ('like', 'unlike', 'view', 'clone'); CREATE TYPE privilege_enum AS ENUM ('NONE', 'READ', 'WRITE'); CREATE TYPE workflow_computing_unit_type_enum AS ENUM ('local', 'kubernetes'); +CREATE TYPE workflow_computing_unit_termination_reason_enum AS ENUM ('USER_REQUESTED', 'GARBAGE_COLLECTED'); -- ============================================ -- 5. Create tables @@ -227,6 +228,7 @@ CREATE TABLE IF NOT EXISTS workflow_computing_unit cuid SERIAL PRIMARY KEY, creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, terminate_time TIMESTAMP DEFAULT NULL, + termination_reason workflow_computing_unit_termination_reason_enum DEFAULT NULL, type workflow_computing_unit_type_enum, uri TEXT NOT NULL DEFAULT '', resource TEXT DEFAULT '', diff --git a/sql/updates/28.sql b/sql/updates/28.sql new file mode 100644 index 00000000000..31a4995c833 --- /dev/null +++ b/sql/updates/28.sql @@ -0,0 +1,39 @@ +-- 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. + +\c texera_db + +SET search_path TO texera_db; + +BEGIN; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type WHERE typname = 'workflow_computing_unit_termination_reason_enum' + ) THEN + CREATE TYPE workflow_computing_unit_termination_reason_enum AS ENUM ( + 'USER_REQUESTED', + 'GARBAGE_COLLECTED' + ); + END IF; +END $$; + +ALTER TABLE workflow_computing_unit + ADD COLUMN IF NOT EXISTS termination_reason workflow_computing_unit_termination_reason_enum DEFAULT NULL; + +COMMIT; From 6d94489f60361a3f65a0785a8a76fbff9f16a737 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Wed, 8 Jul 2026 11:34:25 +0800 Subject: [PATCH 03/10] add test computing-unit-managing-service --- build.sbt | 2 + .../ComputingUnitManagingService.scala | 12 +- .../ComputingUnitManagingResource.scala | 74 +++-- .../ComputingUnitManagingServiceRunSpec.scala | 15 + .../ComputingUnitManagingResourceSpec.scala | 272 ++++++++++++++++++ 5 files changed, 357 insertions(+), 18 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 c11e638d75d..7999b3de259 100644 --- a/build.sbt +++ b/build.sbt @@ -111,6 +111,8 @@ lazy val WorkflowCore = (project in file("common/workflow-core")) .dependsOn(DAO % "test->test") // test scope dependency lazy val ComputingUnitManagingService = (project in file("computing-unit-managing-service")) .dependsOn(WorkflowCore, Auth, Config, Resource) + .configs(Test) + .dependsOn(DAO % "test->test") .settings(commonModuleSettings) .settings( dependencyOverrides ++= Seq( 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 24423d49eb3..25f8f8b4213 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 @@ -76,8 +76,10 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ ) if ( - KubernetesConfig.kubernetesComputingUnitEnabled && - KubernetesConfig.computingUnitIdleTimeoutMinutes > 0 + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + KubernetesConfig.kubernetesComputingUnitEnabled, + KubernetesConfig.computingUnitIdleTimeoutMinutes + ) ) { environment.lifecycle .scheduledExecutorService("idle-computing-unit-terminator") @@ -114,6 +116,12 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ } object ComputingUnitManagingService { + private[service] def shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled: Boolean, + idleTimeoutMinutes: Long + ): Boolean = + kubernetesComputingUnitEnabled && idleTimeoutMinutes > 0 + def main(args: Array[String]): Unit = { val configFilePath = Path .of(sys.env.getOrElse("TEXERA_HOME", ".")) 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 d89ccd0105d..b08a4831618 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 @@ -77,18 +77,60 @@ object ComputingUnitManagingResource { .getInstance() .createDSLContext() - def terminateIdleKubernetesComputingUnits(): List[TerminatedComputingUnitInfo] = { - if ( - !KubernetesConfig.kubernetesComputingUnitEnabled || - KubernetesConfig.computingUnitIdleTimeoutMinutes <= 0 - ) { + private[resource] case class IdleComputingUnitCleanupConfig( + enabled: Boolean, + idleTimeoutMinutes: Long + ) + + private[resource] trait KubernetesPodOperations { + def podExists(cuid: Int): Boolean + def deletePod(cuid: Int): Unit + } + + private object DefaultKubernetesPodOperations extends KubernetesPodOperations { + override def podExists(cuid: Int): Boolean = KubernetesClient.podExists(cuid) + override def deletePod(cuid: Int): Unit = KubernetesClient.deletePod(cuid) + } + + private[resource] def lastComputingUnitActivityTime( + unit: WorkflowComputingUnit, + latestUpdateTime: Option[Timestamp], + latestStartTime: Option[Timestamp] + ): Timestamp = + Seq( + latestUpdateTime, + latestStartTime, + Option(unit.getCreationTime) + ).flatten.maxBy(_.getTime) + + private[resource] def shouldTerminateIdleComputingUnit( + hasActiveExecution: Boolean, + lastExecutionTime: Timestamp, + cutoff: Timestamp + ): Boolean = + !hasActiveExecution && lastExecutionTime.before(cutoff) + + def terminateIdleKubernetesComputingUnits(): List[TerminatedComputingUnitInfo] = + terminateIdleKubernetesComputingUnits( + IdleComputingUnitCleanupConfig( + KubernetesConfig.kubernetesComputingUnitEnabled, + KubernetesConfig.computingUnitIdleTimeoutMinutes + ), + () => new Timestamp(System.currentTimeMillis()), + DefaultKubernetesPodOperations + ) + + private[resource] def terminateIdleKubernetesComputingUnits( + cleanupConfig: IdleComputingUnitCleanupConfig, + currentTime: () => Timestamp, + podOperations: KubernetesPodOperations + ): List[TerminatedComputingUnitInfo] = { + if (!cleanupConfig.enabled || cleanupConfig.idleTimeoutMinutes <= 0) { return List.empty } - val now = new Timestamp(System.currentTimeMillis()) - val cutoff = new Timestamp( - now.getTime - KubernetesConfig.computingUnitIdleTimeoutMinutes * 60 * 1000 - ) + val now = currentTime() + val cutoff = new Timestamp(now.getTime - cleanupConfig.idleTimeoutMinutes * 60 * 1000) val activeStatuses = Seq(Short.box(0), Short.box(1), Short.box(2)) withTransaction(context) { ctx => @@ -125,15 +167,15 @@ object ComputingUnitManagingResource { .from(WORKFLOW_EXECUTIONS) .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid)) .fetchOne(0, classOf[Timestamp]) - val lastExecutionTime = Seq( + val lastExecutionTime = lastComputingUnitActivityTime( + unit, Option(latestUpdateTime), - Option(latestStartTime), - Some(unit.getCreationTime) - ).flatten.maxBy(_.getTime) + Option(latestStartTime) + ) - if (!hasActiveExecution && lastExecutionTime.before(cutoff)) { - if (KubernetesClient.podExists(cuid)) { - KubernetesClient.deletePod(cuid) + if (shouldTerminateIdleComputingUnit(hasActiveExecution, lastExecutionTime, cutoff)) { + if (podOperations.podExists(cuid)) { + podOperations.deletePod(cuid) } unit.setTerminateTime(now) unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) 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..b8cb0c8a12b 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 @@ -40,4 +40,19 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { ) ) shouldBe empty } + + "ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup" should "enable scheduling only when Kubernetes cleanup is enabled with a positive timeout" in { + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 1 + ) shouldBe true + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = false, + idleTimeoutMinutes = 1 + ) shouldBe false + ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 0 + ) shouldBe false + } } 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..0a4cff2a78d --- /dev/null +++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala @@ -0,0 +1,272 @@ +/* + * 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.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW, WORKFLOW_COMPUTING_UNIT, WORKFLOW_EXECUTIONS, WORKFLOW_VERSION, USER => USER_TABLE} +import org.apache.texera.dao.jooq.generated.enums.{WorkflowComputingUnitTerminationReasonEnum, WorkflowComputingUnitTypeEnum} +import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao, WorkflowDao, WorkflowExecutionsDao, WorkflowVersionDao} +import org.apache.texera.dao.jooq.generated.tables.pojos.{User, Workflow, WorkflowComputingUnit, WorkflowExecutions, WorkflowVersion} +import org.apache.texera.service.resource.ComputingUnitManagingResource.{IdleComputingUnitCleanupConfig, KubernetesPodOperations} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} + +import java.sql.Timestamp +import java.util.UUID +import java.util.concurrent.TimeUnit + +class ComputingUnitManagingResourceSpec + extends AnyFlatSpec + with Matchers + with BeforeAndAfterAll + with BeforeAndAfterEach + with MockTexeraDB { + + private val testUserId = 810000 + scala.util.Random.nextInt(10000) + private val testWorkflowId = 820000 + scala.util.Random.nextInt(10000) + private val now = new Timestamp(TimeUnit.DAYS.toMillis(20)) + private val cleanupConfig = IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + + private var userDao: UserDao = _ + private var workflowDao: WorkflowDao = _ + private var workflowVersionDao: WorkflowVersionDao = _ + private var workflowComputingUnitDao: WorkflowComputingUnitDao = _ + private var workflowExecutionsDao: WorkflowExecutionsDao = _ + private var testVersion: WorkflowVersion = _ + + override protected def beforeAll(): Unit = { + initializeDBAndReplaceDSLContext() + } + + override protected def beforeEach(): Unit = { + userDao = new UserDao(getDSLContext.configuration()) + workflowDao = new WorkflowDao(getDSLContext.configuration()) + workflowVersionDao = new WorkflowVersionDao(getDSLContext.configuration()) + workflowComputingUnitDao = new WorkflowComputingUnitDao(getDSLContext.configuration()) + workflowExecutionsDao = new WorkflowExecutionsDao(getDSLContext.configuration()) + + cleanupTestData() + + val user = new User + user.setUid(testUserId) + user.setName("idle-cu-owner") + user.setEmail(s"idle-cu-${UUID.randomUUID()}@example.com") + user.setPassword("password") + userDao.insert(user) + + val workflow = new Workflow + workflow.setWid(testWorkflowId) + workflow.setName("idle-cu-workflow") + workflow.setContent("{}") + workflow.setCreationTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflow.setLastModifiedTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflowDao.insert(workflow) + + testVersion = new WorkflowVersion + testVersion.setWid(testWorkflowId) + testVersion.setContent("{}") + testVersion.setCreationTime(new Timestamp(now.getTime - TimeUnit.DAYS.toMillis(2))) + workflowVersionDao.insert(testVersion) + } + + override protected def afterEach(): Unit = { + cleanupTestData() + } + + override protected def afterAll(): Unit = { + shutdownDB() + } + + private def cleanupTestData(): Unit = { + getDSLContext + .deleteFrom(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.UID.eq(testUserId)) + .execute() + getDSLContext + .deleteFrom(WORKFLOW_COMPUTING_UNIT) + .where(WORKFLOW_COMPUTING_UNIT.UID.eq(testUserId)) + .execute() + getDSLContext + .deleteFrom(WORKFLOW_VERSION) + .where(WORKFLOW_VERSION.WID.eq(testWorkflowId)) + .execute() + getDSLContext.deleteFrom(WORKFLOW).where(WORKFLOW.WID.eq(testWorkflowId)).execute() + getDSLContext.deleteFrom(USER_TABLE).where(USER_TABLE.UID.eq(testUserId)).execute() + } + + private def timestampMinutesBefore(minutes: Long): Timestamp = + new Timestamp(now.getTime - TimeUnit.MINUTES.toMillis(minutes)) + + private def insertComputingUnit( + name: String, + unitType: WorkflowComputingUnitTypeEnum = WorkflowComputingUnitTypeEnum.kubernetes, + creationMinutesBefore: Long = 120, + terminated: Boolean = false + ): WorkflowComputingUnit = { + val unit = new WorkflowComputingUnit + unit.setUid(testUserId) + unit.setName(name) + unit.setCreationTime(timestampMinutesBefore(creationMinutesBefore)) + unit.setType(unitType) + unit.setUri("kubernetes://test") + unit.setResource("{}") + if (terminated) { + unit.setTerminateTime(timestampMinutesBefore(10)) + unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED) + } + workflowComputingUnitDao.insert(unit) + unit + } + + private def insertExecution( + unit: WorkflowComputingUnit, + status: Short, + startingMinutesBefore: Long, + lastUpdateMinutesBefore: Option[Long] = None + ): Unit = { + val execution = new WorkflowExecutions + execution.setVid(testVersion.getVid) + execution.setUid(testUserId) + execution.setCuid(unit.getCuid) + execution.setStatus(status) + execution.setStartingTime(timestampMinutesBefore(startingMinutesBefore)) + lastUpdateMinutesBefore.foreach(minutes => execution.setLastUpdateTime(timestampMinutesBefore(minutes))) + execution.setBookmarked(false) + execution.setName("execution-" + UUID.randomUUID().toString.substring(0, 8)) + execution.setEnvironmentVersion("test-env") + workflowExecutionsDao.insert(execution) + } + + private class RecordingPodOperations(existingPods: Set[Int]) extends KubernetesPodOperations { + var deletedPods: List[Int] = List.empty + + override def podExists(cuid: Int): Boolean = existingPods.contains(cuid) + + override def deletePod(cuid: Int): Unit = { + deletedPods = deletedPods :+ cuid + } + } + + "terminateIdleKubernetesComputingUnits" should "return empty without scanning when cleanup is disabled" in { + val podOperations = new RecordingPodOperations(Set.empty) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig.copy(enabled = false), + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + } + + it should "return empty when the idle timeout is disabled" in { + val podOperations = new RecordingPodOperations(Set.empty) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig.copy(idleTimeoutMinutes = 0), + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + } + + it should "garbage collect only inactive Kubernetes computing units past the timeout" in { + val stale = insertComputingUnit("stale") + val active = insertComputingUnit("active") + val recent = insertComputingUnit("recent") + val local = insertComputingUnit("local", WorkflowComputingUnitTypeEnum.local) + val alreadyTerminated = insertComputingUnit("already-terminated", terminated = true) + + insertExecution(active, status = 1, startingMinutesBefore = 180) + insertExecution(recent, status = 3, startingMinutesBefore = 180, lastUpdateMinutesBefore = Some(5)) + insertExecution(local, status = 3, startingMinutesBefore = 180) + + val podOperations = new RecordingPodOperations(Set(stale.getCuid, active.getCuid)) + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(stale.getCuid) + terminated.head.username shouldBe Some("idle-cu-owner") + terminated.head.reason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + podOperations.deletedPods shouldBe List(stale.getCuid) + + val staleAfterCleanup = workflowComputingUnitDao.fetchOneByCuid(stale.getCuid) + staleAfterCleanup.getTerminateTime shouldBe now + staleAfterCleanup.getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + + workflowComputingUnitDao.fetchOneByCuid(active.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(recent.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(alreadyTerminated.getCuid).getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED + } + + it should "mark an idle unit terminated even when the pod is already absent" in { + val stale = insertComputingUnit("stale-missing-pod") + val podOperations = new RecordingPodOperations(Set.empty) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(stale.getCuid) + podOperations.deletedPods shouldBe empty + workflowComputingUnitDao + .fetchOneByCuid(stale.getCuid) + .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + } + + "lastComputingUnitActivityTime" should "prefer the latest execution timestamp over creation time" in { + val unit = new WorkflowComputingUnit + unit.setCreationTime(timestampMinutesBefore(120)) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = Some(timestampMinutesBefore(10)), + latestStartTime = Some(timestampMinutesBefore(30)) + ) shouldBe timestampMinutesBefore(10) + } + + "shouldTerminateIdleComputingUnit" should "require both no active execution and activity before cutoff" in { + val cutoff = timestampMinutesBefore(60) + + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = false, + lastExecutionTime = timestampMinutesBefore(61), + cutoff = cutoff + ) shouldBe true + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = true, + lastExecutionTime = timestampMinutesBefore(61), + cutoff = cutoff + ) shouldBe false + ComputingUnitManagingResource.shouldTerminateIdleComputingUnit( + hasActiveExecution = false, + lastExecutionTime = timestampMinutesBefore(60), + cutoff = cutoff + ) shouldBe false + } +} From 789339c0a35382d37c4aa0e95bbea1fa9aac9b24 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Wed, 8 Jul 2026 11:51:24 +0800 Subject: [PATCH 04/10] fix formatting issues --- .../ComputingUnitManagingResourceSpec.scala | 52 +++++++++++++++---- 1 file changed, 43 insertions(+), 9 deletions(-) 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 index 0a4cff2a78d..8265bb26385 100644 --- 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 @@ -20,11 +20,35 @@ package org.apache.texera.service.resource import org.apache.texera.dao.MockTexeraDB -import org.apache.texera.dao.jooq.generated.Tables.{WORKFLOW, WORKFLOW_COMPUTING_UNIT, WORKFLOW_EXECUTIONS, WORKFLOW_VERSION, USER => USER_TABLE} -import org.apache.texera.dao.jooq.generated.enums.{WorkflowComputingUnitTerminationReasonEnum, WorkflowComputingUnitTypeEnum} -import org.apache.texera.dao.jooq.generated.tables.daos.{UserDao, WorkflowComputingUnitDao, WorkflowDao, WorkflowExecutionsDao, WorkflowVersionDao} -import org.apache.texera.dao.jooq.generated.tables.pojos.{User, Workflow, WorkflowComputingUnit, WorkflowExecutions, WorkflowVersion} -import org.apache.texera.service.resource.ComputingUnitManagingResource.{IdleComputingUnitCleanupConfig, KubernetesPodOperations} +import org.apache.texera.dao.jooq.generated.Tables.{ + WORKFLOW, + WORKFLOW_COMPUTING_UNIT, + WORKFLOW_EXECUTIONS, + WORKFLOW_VERSION, + USER => USER_TABLE +} +import org.apache.texera.dao.jooq.generated.enums.{ + WorkflowComputingUnitTerminationReasonEnum, + WorkflowComputingUnitTypeEnum +} +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowComputingUnitDao, + WorkflowDao, + WorkflowExecutionsDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowComputingUnit, + WorkflowExecutions, + WorkflowVersion +} +import org.apache.texera.service.resource.ComputingUnitManagingResource.{ + IdleComputingUnitCleanupConfig, + KubernetesPodOperations +} import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} @@ -43,7 +67,8 @@ class ComputingUnitManagingResourceSpec private val testUserId = 810000 + scala.util.Random.nextInt(10000) private val testWorkflowId = 820000 + scala.util.Random.nextInt(10000) private val now = new Timestamp(TimeUnit.DAYS.toMillis(20)) - private val cleanupConfig = IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + private val cleanupConfig = + IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) private var userDao: UserDao = _ private var workflowDao: WorkflowDao = _ @@ -148,7 +173,9 @@ class ComputingUnitManagingResourceSpec execution.setCuid(unit.getCuid) execution.setStatus(status) execution.setStartingTime(timestampMinutesBefore(startingMinutesBefore)) - lastUpdateMinutesBefore.foreach(minutes => execution.setLastUpdateTime(timestampMinutesBefore(minutes))) + lastUpdateMinutesBefore.foreach(minutes => + execution.setLastUpdateTime(timestampMinutesBefore(minutes)) + ) execution.setBookmarked(false) execution.setName("execution-" + UUID.randomUUID().toString.substring(0, 8)) execution.setEnvironmentVersion("test-env") @@ -197,7 +224,12 @@ class ComputingUnitManagingResourceSpec val alreadyTerminated = insertComputingUnit("already-terminated", terminated = true) insertExecution(active, status = 1, startingMinutesBefore = 180) - insertExecution(recent, status = 3, startingMinutesBefore = 180, lastUpdateMinutesBefore = Some(5)) + insertExecution( + recent, + status = 3, + startingMinutesBefore = 180, + lastUpdateMinutesBefore = Some(5) + ) insertExecution(local, status = 3, startingMinutesBefore = 180) val podOperations = new RecordingPodOperations(Set(stale.getCuid, active.getCuid)) @@ -219,7 +251,9 @@ class ComputingUnitManagingResourceSpec workflowComputingUnitDao.fetchOneByCuid(active.getCuid).getTerminateTime shouldBe null workflowComputingUnitDao.fetchOneByCuid(recent.getCuid).getTerminateTime shouldBe null workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null - workflowComputingUnitDao.fetchOneByCuid(alreadyTerminated.getCuid).getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED + workflowComputingUnitDao + .fetchOneByCuid(alreadyTerminated.getCuid) + .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED } it should "mark an idle unit terminated even when the pod is already absent" in { From a1e2ba4a61f01dc5ef7d2e2daead6c484c77bc39 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Wed, 8 Jul 2026 21:56:28 +0800 Subject: [PATCH 05/10] test(computing-unit-managing-service), improve testing coverage --- .../ComputingUnitManagingService.scala | 47 ++++++++----- .../ComputingUnitManagingServiceRunSpec.scala | 67 +++++++++++++++++++ .../ComputingUnitManagingResourceSpec.scala | 37 ++++++++++ 3 files changed, 134 insertions(+), 17 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 25f8f8b4213..c8dce7ae6e1 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 @@ -31,6 +31,7 @@ import org.apache.texera.service.resource.{ ComputingUnitManagingResource, HealthCheckResource } +import org.apache.texera.service.resource.ComputingUnitManagingResource.TerminatedComputingUnitInfo import org.slf4j.LoggerFactory import java.nio.file.Path import java.util.concurrent.TimeUnit @@ -87,23 +88,12 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ .build() .scheduleWithFixedDelay( () => - try { - val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits() - if (terminated.nonEmpty) { - val terminatedDetails = terminated - .map(unit => - s"cuid=${unit.cuid}, name=${unit.name}, uid=${unit.uid}, username=${unit.username - .getOrElse("unknown")}, reason=${unit.reason.getLiteral}" - ) - .mkString("; ") - logger.info( - s"Terminated ${terminated.size} idle Kubernetes computing unit(s): $terminatedDetails" - ) - } - } catch { - case t: Throwable => - logger.warn("Failed to terminate idle Kubernetes computing units", t) - }, + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), + message => logger.info(message), + throwable => + logger.warn("Failed to terminate idle Kubernetes computing units", throwable) + ), KubernetesConfig.computingUnitIdleCheckIntervalMinutes, KubernetesConfig.computingUnitIdleCheckIntervalMinutes, TimeUnit.MINUTES @@ -122,6 +112,29 @@ object ComputingUnitManagingService { ): Boolean = kubernetesComputingUnitEnabled && idleTimeoutMinutes > 0 + private[service] def runIdleComputingUnitCleanup( + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + try { + val terminated = terminateIdleComputingUnits() + if (terminated.nonEmpty) { + val terminatedDetails = terminated + .map(unit => + s"cuid=${unit.cuid}, name=${unit.name}, uid=${unit.uid}, username=${unit.username + .getOrElse("unknown")}, reason=${unit.reason.getLiteral}" + ) + .mkString("; ") + logTerminatedUnits( + s"Terminated ${terminated.size} idle Kubernetes computing unit(s): $terminatedDetails" + ) + } + } catch { + case t: Throwable => + logCleanupFailure(t) + } + def main(args: Array[String]): Unit = { val configFilePath = Path .of(sys.env.getOrElse("TEXERA_HOME", ".")) 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 b8cb0c8a12b..f30cd5ed5a2 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 @@ -20,11 +20,13 @@ package org.apache.texera.service import org.apache.texera.auth.RoleAnnotationEnforcer +import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTerminationReasonEnum import org.apache.texera.service.resource.{ ComputingUnitAccessResource, ComputingUnitManagingResource, HealthCheckResource } +import org.apache.texera.service.resource.ComputingUnitManagingResource.TerminatedComputingUnitInfo import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -41,6 +43,71 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { ) shouldBe empty } + "ComputingUnitManagingService.runIdleComputingUnitCleanup" should "not log when no idle computing units are terminated" in { + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => List.empty, + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + + infoMessages shouldBe empty + failures shouldBe empty + } + + it should "log terminated idle computing unit details" in { + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => + List( + TerminatedComputingUnitInfo( + cuid = 1, + name = "idle-a", + uid = 10, + username = Some("alice"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ), + TerminatedComputingUnitInfo( + cuid = 2, + name = "idle-b", + uid = 20, + username = None, + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ), + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + + infoMessages shouldBe List( + "Terminated 2 idle Kubernetes computing unit(s): " + + "cuid=1, name=idle-a, uid=10, username=alice, reason=GARBAGE_COLLECTED; " + + "cuid=2, name=idle-b, uid=20, username=unknown, reason=GARBAGE_COLLECTED" + ) + failures shouldBe empty + } + + it should "log cleanup failures without throwing" in { + val failure = new RuntimeException("cleanup failed") + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + noException shouldBe thrownBy { + ComputingUnitManagingService.runIdleComputingUnitCleanup( + () => throw failure, + message => infoMessages = infoMessages :+ message, + throwable => failures = failures :+ throwable + ) + } + + infoMessages shouldBe empty + failures shouldBe List(failure) + } + "ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup" should "enable scheduling only when Kubernetes cleanup is enabled with a positive timeout" in { ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( kubernetesComputingUnitEnabled = true, 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 index 8265bb26385..293e0f013aa 100644 --- 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 @@ -19,6 +19,7 @@ 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.Tables.{ WORKFLOW, @@ -182,6 +183,16 @@ class ComputingUnitManagingResourceSpec workflowExecutionsDao.insert(execution) } + private def sessionUser( + uid: Integer = testUserId, + name: String = "idle-cu-owner" + ): SessionUser = { + val user = new User + user.setUid(uid) + user.setName(name) + new SessionUser(user) + } + private class RecordingPodOperations(existingPods: Set[Int]) extends KubernetesPodOperations { var deletedPods: List[Int] = List.empty @@ -273,6 +284,32 @@ class ComputingUnitManagingResourceSpec .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED } + "terminateComputingUnit" should "mark manual termination as user requested" in { + val local = insertComputingUnit("manual-local", WorkflowComputingUnitTypeEnum.local) + + val response = new ComputingUnitManagingResource().terminateComputingUnit( + local.getCuid, + sessionUser() + ) + + response.getStatus shouldBe 200 + val terminated = workflowComputingUnitDao.fetchOneByCuid(local.getCuid) + terminated.getTerminateTime should not be null + terminated.getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.USER_REQUESTED + } + + it should "reject manual termination from a non-owner" in { + val local = insertComputingUnit("manual-local-non-owner", WorkflowComputingUnitTypeEnum.local) + + val response = new ComputingUnitManagingResource().terminateComputingUnit( + local.getCuid, + sessionUser(uid = testUserId + 1) + ) + + response.getStatus shouldBe 400 + workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null + } + "lastComputingUnitActivityTime" should "prefer the latest execution timestamp over creation time" in { val unit = new WorkflowComputingUnit unit.setCreationTime(timestampMinutesBefore(120)) From 1159e93c6b6078e7b05692a981410435cf6a05cb Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Thu, 9 Jul 2026 09:26:58 +0800 Subject: [PATCH 06/10] add more test coverage --- .../ComputingUnitManagingService.scala | 47 ++++++++++++------- .../ComputingUnitManagingServiceRunSpec.scala | 39 +++++++++++++++ 2 files changed, 70 insertions(+), 16 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 c8dce7ae6e1..f163c477ce9 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 @@ -82,22 +82,18 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ KubernetesConfig.computingUnitIdleTimeoutMinutes ) ) { - environment.lifecycle - .scheduledExecutorService("idle-computing-unit-terminator") - .threads(1) - .build() - .scheduleWithFixedDelay( - () => - ComputingUnitManagingService.runIdleComputingUnitCleanup( - () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), - message => logger.info(message), - throwable => - logger.warn("Failed to terminate idle Kubernetes computing units", throwable) - ), - KubernetesConfig.computingUnitIdleCheckIntervalMinutes, - KubernetesConfig.computingUnitIdleCheckIntervalMinutes, - TimeUnit.MINUTES - ) + ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( + (command, initialDelay, delay, unit) => + environment.lifecycle + .scheduledExecutorService("idle-computing-unit-terminator") + .threads(1) + .build() + .scheduleWithFixedDelay(command, initialDelay, delay, unit), + KubernetesConfig.computingUnitIdleCheckIntervalMinutes, + () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), + message => logger.info(message), + throwable => logger.warn("Failed to terminate idle Kubernetes computing units", throwable) + ) } // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL @@ -112,6 +108,25 @@ object ComputingUnitManagingService { ): Boolean = kubernetesComputingUnitEnabled && idleTimeoutMinutes > 0 + private[service] def scheduleIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit, + idleCheckIntervalMinutes: Long, + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + scheduleWithFixedDelay( + () => + runIdleComputingUnitCleanup( + terminateIdleComputingUnits, + logTerminatedUnits, + logCleanupFailure + ), + idleCheckIntervalMinutes, + idleCheckIntervalMinutes, + TimeUnit.MINUTES + ) + private[service] def runIdleComputingUnitCleanup( terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], logTerminatedUnits: String => Unit, 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 f30cd5ed5a2..42f8372768e 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 @@ -30,6 +30,8 @@ import org.apache.texera.service.resource.ComputingUnitManagingResource.Terminat import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.util.concurrent.TimeUnit + class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { // Every endpoint this service registers declares @RolesAllowed/@PermitAll/@DenyAll. @@ -43,6 +45,43 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { ) shouldBe empty } + "ComputingUnitManagingService.scheduleIdleComputingUnitCleanup" should "schedule cleanup with the configured fixed delay" in { + var scheduledCommand: Runnable = null + var scheduledInitialDelay: Long = -1 + var scheduledDelay: Long = -1 + var scheduledTimeUnit: TimeUnit = null + var cleanupInvocations = 0 + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( + (command, initialDelay, delay, unit) => { + scheduledCommand = command + scheduledInitialDelay = initialDelay + scheduledDelay = delay + scheduledTimeUnit = unit + }, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => { + cleanupInvocations += 1 + List.empty + }, + logTerminatedUnits = message => infoMessages = infoMessages :+ message, + logCleanupFailure = throwable => failures = failures :+ throwable + ) + + scheduledCommand should not be null + scheduledInitialDelay shouldBe 15 + scheduledDelay shouldBe 15 + scheduledTimeUnit shouldBe TimeUnit.MINUTES + + scheduledCommand.run() + + cleanupInvocations shouldBe 1 + infoMessages shouldBe empty + failures shouldBe empty + } + "ComputingUnitManagingService.runIdleComputingUnitCleanup" should "not log when no idle computing units are terminated" in { var infoMessages = List.empty[String] var failures = List.empty[Throwable] From 1c33c1b0ae0d249d3e6e98bdd4a47dd1101cfebd Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Thu, 9 Jul 2026 11:32:36 +0800 Subject: [PATCH 07/10] add more test coverage --- .../ComputingUnitManagingService.scala | 70 +++++++++------ .../ComputingUnitManagingResource.scala | 8 +- .../ComputingUnitManagingServiceRunSpec.scala | 79 ++++++++++++++++- .../ComputingUnitManagingResourceSpec.scala | 86 +++++++++++++++++-- 4 files changed, 209 insertions(+), 34 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 f163c477ce9..d7066393d4f 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 @@ -39,6 +39,29 @@ import java.util.concurrent.TimeUnit class ComputingUnitManagingService extends Application[ComputingUnitManagingServiceConfiguration] { private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) + private[service] def initSqlServer(): Unit = + SqlServer.initConnection( + StorageConfig.jdbcUrl, + StorageConfig.jdbcUsername, + StorageConfig.jdbcPassword + ) + + private[service] def registerIdleComputingUnitCleanup(environment: Environment): Unit = + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + (command, initialDelay, delay, unit) => + environment.lifecycle + .scheduledExecutorService("idle-computing-unit-terminator") + .threads(1) + .build() + .scheduleWithFixedDelay(command, initialDelay, delay, unit), + KubernetesConfig.kubernetesComputingUnitEnabled, + KubernetesConfig.computingUnitIdleTimeoutMinutes, + KubernetesConfig.computingUnitIdleCheckIntervalMinutes, + () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), + message => logger.info(message), + throwable => logger.warn("Failed to terminate idle Kubernetes computing units", throwable) + ) + override def initialize( bootstrap: Bootstrap[ComputingUnitManagingServiceConfiguration] ): Unit = { @@ -62,11 +85,7 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ AuthFeatures.register(environment) - SqlServer.initConnection( - StorageConfig.jdbcUrl, - StorageConfig.jdbcUsername, - StorageConfig.jdbcPassword - ) + initSqlServer() environment.jersey().register(new ComputingUnitManagingResource) environment.jersey().register(new ComputingUnitAccessResource) @@ -76,25 +95,7 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ "ComputingUnitManagingService" ) - if ( - ComputingUnitManagingService.shouldScheduleIdleComputingUnitCleanup( - KubernetesConfig.kubernetesComputingUnitEnabled, - KubernetesConfig.computingUnitIdleTimeoutMinutes - ) - ) { - ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( - (command, initialDelay, delay, unit) => - environment.lifecycle - .scheduledExecutorService("idle-computing-unit-terminator") - .threads(1) - .build() - .scheduleWithFixedDelay(command, initialDelay, delay, unit), - KubernetesConfig.computingUnitIdleCheckIntervalMinutes, - () => ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits(), - message => logger.info(message), - throwable => logger.warn("Failed to terminate idle Kubernetes computing units", throwable) - ) - } + registerIdleComputingUnitCleanup(environment) // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL RequestLoggingFilter.register(environment.getApplicationContext) @@ -108,6 +109,27 @@ object ComputingUnitManagingService { ): Boolean = kubernetesComputingUnitEnabled && idleTimeoutMinutes > 0 + private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit, + kubernetesComputingUnitEnabled: Boolean, + idleTimeoutMinutes: Long, + idleCheckIntervalMinutes: Long, + terminateIdleComputingUnits: () => List[TerminatedComputingUnitInfo], + logTerminatedUnits: String => Unit, + logCleanupFailure: Throwable => Unit + ): Unit = + if ( + shouldScheduleIdleComputingUnitCleanup(kubernetesComputingUnitEnabled, idleTimeoutMinutes) + ) { + scheduleIdleComputingUnitCleanup( + scheduleWithFixedDelay, + idleCheckIntervalMinutes, + terminateIdleComputingUnits, + logTerminatedUnits, + logCleanupFailure + ) + } + private[service] def scheduleIdleComputingUnitCleanup( scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit, idleCheckIntervalMinutes: Long, 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 b08a4831618..15cc1e5d8e9 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 @@ -83,13 +83,13 @@ object ComputingUnitManagingResource { ) private[resource] trait KubernetesPodOperations { - def podExists(cuid: Int): Boolean - def deletePod(cuid: Int): Unit + val podExists: Int => Boolean + val deletePod: Int => Unit } private object DefaultKubernetesPodOperations extends KubernetesPodOperations { - override def podExists(cuid: Int): Boolean = KubernetesClient.podExists(cuid) - override def deletePod(cuid: Int): Unit = KubernetesClient.deletePod(cuid) + override val podExists: Int => Boolean = KubernetesClient.podExists + override val deletePod: Int => Unit = KubernetesClient.deletePod } private[resource] def lastComputingUnitActivityTime( 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 42f8372768e..20e917d980b 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,7 +19,9 @@ package org.apache.texera.service +import io.dropwizard.core.setup.Environment import org.apache.texera.auth.RoleAnnotationEnforcer +import org.apache.texera.dao.MockTexeraDB import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTerminationReasonEnum import org.apache.texera.service.resource.{ ComputingUnitAccessResource, @@ -32,7 +34,21 @@ import org.scalatest.matchers.should.Matchers import java.util.concurrent.TimeUnit -class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { +class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers with MockTexeraDB { + + private class TestComputingUnitManagingService extends ComputingUnitManagingService { + override private[service] def initSqlServer(): Unit = initializeDBAndReplaceDSLContext() + } + + "ComputingUnitManagingService.run" should "register resources without requiring the configured SQL server" in { + noException shouldBe thrownBy { + new TestComputingUnitManagingService().run( + new ComputingUnitManagingServiceConfiguration, + new Environment("test-computing-unit-managing-service") + ) + } + shutdownDB() + } // Every endpoint this service registers declares @RolesAllowed/@PermitAll/@DenyAll. "ComputingUnitManagingService's registered resources" should "all declare access control" in { @@ -45,6 +61,32 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { ) shouldBe empty } + "ComputingUnitManagingService.registerIdleComputingUnitCleanup" should "schedule cleanup only when Kubernetes idle cleanup is enabled" in { + var scheduled = false + + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + (_, _, _, _) => scheduled = true, + kubernetesComputingUnitEnabled = false, + idleTimeoutMinutes = 60, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => List.empty, + logTerminatedUnits = _ => (), + logCleanupFailure = _ => () + ) + scheduled shouldBe false + + ComputingUnitManagingService.registerIdleComputingUnitCleanup( + (_, _, _, _) => scheduled = true, + kubernetesComputingUnitEnabled = true, + idleTimeoutMinutes = 60, + idleCheckIntervalMinutes = 15, + terminateIdleComputingUnits = () => List.empty, + logTerminatedUnits = _ => (), + logCleanupFailure = _ => () + ) + scheduled shouldBe true + } + "ComputingUnitManagingService.scheduleIdleComputingUnitCleanup" should "schedule cleanup with the configured fixed delay" in { var scheduledCommand: Runnable = null var scheduledInitialDelay: Long = -1 @@ -82,6 +124,41 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers { failures shouldBe empty } + it should "run the scheduled cleanup command and log terminated units" in { + var scheduledCommand: Runnable = null + var cleanupInvocations = 0 + var infoMessages = List.empty[String] + var failures = List.empty[Throwable] + + ComputingUnitManagingService.scheduleIdleComputingUnitCleanup( + (command, _, _, _) => scheduledCommand = command, + idleCheckIntervalMinutes = 5, + terminateIdleComputingUnits = () => { + cleanupInvocations += 1 + List( + TerminatedComputingUnitInfo( + cuid = 3, + name = "scheduled-idle", + uid = 30, + username = Some("carol"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ) + }, + logTerminatedUnits = message => infoMessages = infoMessages :+ message, + logCleanupFailure = throwable => failures = failures :+ throwable + ) + + scheduledCommand.run() + + cleanupInvocations shouldBe 1 + infoMessages shouldBe List( + "Terminated 1 idle Kubernetes computing unit(s): " + + "cuid=3, name=scheduled-idle, uid=30, username=carol, reason=GARBAGE_COLLECTED" + ) + failures shouldBe empty + } + "ComputingUnitManagingService.runIdleComputingUnitCleanup" should "not log when no idle computing units are terminated" in { var infoMessages = List.empty[String] var failures = List.empty[Throwable] 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 index 293e0f013aa..991a7c2b1be 100644 --- 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 @@ -196,14 +196,15 @@ class ComputingUnitManagingResourceSpec private class RecordingPodOperations(existingPods: Set[Int]) extends KubernetesPodOperations { var deletedPods: List[Int] = List.empty - override def podExists(cuid: Int): Boolean = existingPods.contains(cuid) + override val podExists: Int => Boolean = cuid => existingPods.contains(cuid) + override val deletePod: Int => Unit = cuid => deletedPods = deletedPods :+ cuid + } - override def deletePod(cuid: Int): Unit = { - deletedPods = deletedPods :+ cuid - } + "terminateIdleKubernetesComputingUnits" should "use the default disabled cleanup configuration" in { + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits() shouldBe empty } - "terminateIdleKubernetesComputingUnits" should "return empty without scanning when cleanup is disabled" in { + it should "return empty without scanning when cleanup is disabled" in { val podOperations = new RecordingPodOperations(Set.empty) ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( @@ -284,6 +285,64 @@ class ComputingUnitManagingResourceSpec .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED } + it should "keep units with active status 0 or 2 running" in { + val activeQueued = insertComputingUnit("active-status-0") + val activeRunning = insertComputingUnit("active-status-2") + insertExecution(activeQueued, status = 0, startingMinutesBefore = 180) + insertExecution(activeRunning, status = 2, startingMinutesBefore = 180) + val podOperations = new RecordingPodOperations(Set(activeQueued.getCuid, activeRunning.getCuid)) + + ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) shouldBe empty + + podOperations.deletedPods shouldBe empty + workflowComputingUnitDao.fetchOneByCuid(activeQueued.getCuid).getTerminateTime shouldBe null + workflowComputingUnitDao.fetchOneByCuid(activeRunning.getCuid).getTerminateTime shouldBe null + } + + it should "terminate a unit whose latest completed execution activity is past the timeout" in { + val staleWithExecution = insertComputingUnit("stale-completed-execution") + insertExecution( + staleWithExecution, + status = 3, + startingMinutesBefore = 180, + lastUpdateMinutesBefore = Some(90) + ) + val podOperations = new RecordingPodOperations(Set(staleWithExecution.getCuid)) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated.map(_.cuid) shouldBe List(staleWithExecution.getCuid) + podOperations.deletedPods shouldBe List(staleWithExecution.getCuid) + workflowComputingUnitDao + .fetchOneByCuid(staleWithExecution.getCuid) + .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + } + + it should "omit an empty owner name from terminated unit info" in { + val user = userDao.fetchOneByUid(testUserId) + user.setName("") + userDao.update(user) + val stale = insertComputingUnit("stale-empty-owner") + val podOperations = new RecordingPodOperations(Set(stale.getCuid)) + + val terminated = ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits( + cleanupConfig, + () => now, + podOperations + ) + + terminated should have size 1 + terminated.head.username shouldBe None + } + "terminateComputingUnit" should "mark manual termination as user requested" in { val local = insertComputingUnit("manual-local", WorkflowComputingUnitTypeEnum.local) @@ -321,6 +380,23 @@ class ComputingUnitManagingResourceSpec ) shouldBe timestampMinutesBefore(10) } + it should "fall back to start time and then creation time" in { + val unit = new WorkflowComputingUnit + unit.setCreationTime(timestampMinutesBefore(120)) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = None, + latestStartTime = Some(timestampMinutesBefore(30)) + ) shouldBe timestampMinutesBefore(30) + + ComputingUnitManagingResource.lastComputingUnitActivityTime( + unit, + latestUpdateTime = None, + latestStartTime = None + ) shouldBe timestampMinutesBefore(120) + } + "shouldTerminateIdleComputingUnit" should "require both no active execution and activity before cutoff" in { val cutoff = timestampMinutesBefore(60) From 82a57ee86461a8d6c25d10a5343d75f7a12cfa0c Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Thu, 9 Jul 2026 12:58:30 +0800 Subject: [PATCH 08/10] add more coverage --- .../ComputingUnitManagingService.scala | 24 +++++--- .../ComputingUnitManagingResource.scala | 10 +++- .../ComputingUnitManagingServiceRunSpec.scala | 44 +++++++++++++- .../ComputingUnitManagingResourceSpec.scala | 60 ++++++++++++++++++- 4 files changed, 125 insertions(+), 13 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 d7066393d4f..e59cf3c8760 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 @@ -39,21 +39,29 @@ import java.util.concurrent.TimeUnit class ComputingUnitManagingService extends Application[ComputingUnitManagingServiceConfiguration] { private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService]) - private[service] def initSqlServer(): Unit = - SqlServer.initConnection( + private[service] def initSqlServer( + connect: (String, String, String) => Unit = SqlServer.initConnection + ): Unit = + connect( StorageConfig.jdbcUrl, StorageConfig.jdbcUsername, StorageConfig.jdbcPassword ) private[service] def registerIdleComputingUnitCleanup(environment: Environment): Unit = + registerIdleComputingUnitCleanup((command, initialDelay, delay, unit) => + environment.lifecycle + .scheduledExecutorService("idle-computing-unit-terminator") + .threads(1) + .build() + .scheduleWithFixedDelay(command, initialDelay, delay, unit) + ) + + private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit + ): Unit = ComputingUnitManagingService.registerIdleComputingUnitCleanup( - (command, initialDelay, delay, unit) => - environment.lifecycle - .scheduledExecutorService("idle-computing-unit-terminator") - .threads(1) - .build() - .scheduleWithFixedDelay(command, initialDelay, delay, unit), + scheduleWithFixedDelay, KubernetesConfig.kubernetesComputingUnitEnabled, KubernetesConfig.computingUnitIdleTimeoutMinutes, KubernetesConfig.computingUnitIdleCheckIntervalMinutes, 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 15cc1e5d8e9..667a16061df 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 @@ -636,6 +636,12 @@ class ComputingUnitManagingResource { @Path("") def listComputingUnits( @Auth user: SessionUser + ): List[DashboardWorkflowComputingUnit] = + listComputingUnits(user, DefaultKubernetesPodOperations) + + private[resource] def listComputingUnits( + user: SessionUser, + podOperations: KubernetesPodOperations ): List[DashboardWorkflowComputingUnit] = { withTransaction(context) { ctx => val computingUnitDao = new WorkflowComputingUnitDao(ctx.configuration()) @@ -688,7 +694,7 @@ class ComputingUnitManagingResource { allUnits.foreach { unit => if ( unit.getType == WorkflowComputingUnitTypeEnum.kubernetes && - !KubernetesClient.podExists(unit.getCuid) + !podOperations.podExists(unit.getCuid) ) { unit.setTerminateTime(new Timestamp(System.currentTimeMillis())) unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED) @@ -708,7 +714,7 @@ class ComputingUnitManagingResource { case (unit, _) => unit.getType match { case WorkflowComputingUnitTypeEnum.kubernetes => - KubernetesClient.podExists(unit.getCuid) + podOperations.podExists(unit.getCuid) case _ => true } } 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 20e917d980b..621e9945f34 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 @@ -20,8 +20,10 @@ package org.apache.texera.service import io.dropwizard.core.setup.Environment +import io.dropwizard.lifecycle.setup.{LifecycleEnvironment, ScheduledExecutorServiceBuilder} import org.apache.texera.auth.RoleAnnotationEnforcer import org.apache.texera.dao.MockTexeraDB +import org.mockito.Mockito.{mock, verify, when} import org.apache.texera.dao.jooq.generated.enums.WorkflowComputingUnitTerminationReasonEnum import org.apache.texera.service.resource.{ ComputingUnitAccessResource, @@ -32,12 +34,24 @@ import org.apache.texera.service.resource.ComputingUnitManagingResource.Terminat import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.util.concurrent.TimeUnit +import java.util.concurrent.{ScheduledExecutorService, TimeUnit} class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers with MockTexeraDB { private class TestComputingUnitManagingService extends ComputingUnitManagingService { - override private[service] def initSqlServer(): Unit = initializeDBAndReplaceDSLContext() + override private[service] def initSqlServer( + connect: (String, String, String) => Unit + ): Unit = initializeDBAndReplaceDSLContext() + } + + "ComputingUnitManagingService.initSqlServer" should "connect with the configured storage settings" in { + var connectionArgs: Option[(String, String, String)] = None + + new ComputingUnitManagingService().initSqlServer { (jdbcUrl, username, password) => + connectionArgs = Some((jdbcUrl, username, password)) + } + + connectionArgs shouldBe defined } "ComputingUnitManagingService.run" should "register resources without requiring the configured SQL server" in { @@ -61,6 +75,32 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers with ) shouldBe empty } + "ComputingUnitManagingService.registerIdleComputingUnitCleanup" should "wire the Dropwizard scheduled executor" in { + val environment = mock(classOf[Environment]) + val lifecycle = mock(classOf[LifecycleEnvironment]) + val builder = mock(classOf[ScheduledExecutorServiceBuilder]) + val executor = mock(classOf[ScheduledExecutorService]) + val command = new Runnable { + override def run(): Unit = () + } + + when(environment.lifecycle()).thenReturn(lifecycle) + when(lifecycle.scheduledExecutorService("idle-computing-unit-terminator")).thenReturn(builder) + when(builder.threads(1)).thenReturn(builder) + when(builder.build()).thenReturn(executor) + + val service = new ComputingUnitManagingService { + override private[service] def registerIdleComputingUnitCleanup( + scheduleWithFixedDelay: (Runnable, Long, Long, TimeUnit) => Unit + ): Unit = + scheduleWithFixedDelay(command, 7, 11, TimeUnit.SECONDS) + } + + service.registerIdleComputingUnitCleanup(environment) + + verify(executor).scheduleWithFixedDelay(command, 7, 11, TimeUnit.SECONDS) + } + "ComputingUnitManagingService.registerIdleComputingUnitCleanup" should "schedule cleanup only when Kubernetes idle cleanup is enabled" in { var scheduled = false 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 index 991a7c2b1be..b76612102c1 100644 --- 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 @@ -48,7 +48,8 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ } import org.apache.texera.service.resource.ComputingUnitManagingResource.{ IdleComputingUnitCleanupConfig, - KubernetesPodOperations + KubernetesPodOperations, + TerminatedComputingUnitInfo } import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -200,6 +201,45 @@ class ComputingUnitManagingResourceSpec override val deletePod: Int => Unit = cuid => deletedPods = deletedPods :+ cuid } + "cleanup value objects" should "support generated case class operations" in { + val config = IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + config.copy(enabled = false) shouldBe IdleComputingUnitCleanupConfig( + enabled = false, + idleTimeoutMinutes = 60 + ) + IdleComputingUnitCleanupConfig.unapply(config) shouldBe Some((true, 60)) + config.productIterator.toList shouldBe List(true, 60) + + val terminated = TerminatedComputingUnitInfo( + cuid = 1, + name = "case-class-unit", + uid = testUserId, + username = Some("idle-cu-owner"), + reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + terminated.copy(username = None).username shouldBe None + TerminatedComputingUnitInfo.unapply(terminated) shouldBe Some( + ( + 1, + "case-class-unit", + testUserId, + Some("idle-cu-owner"), + WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + ) + terminated.productIterator.toList shouldBe List( + 1, + "case-class-unit", + testUserId, + Some("idle-cu-owner"), + WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + ) + } + + "listComputingUnits" should "return empty through the public default pod operations wrapper" in { + new ComputingUnitManagingResource().listComputingUnits(sessionUser()) shouldBe empty + } + "terminateIdleKubernetesComputingUnits" should "use the default disabled cleanup configuration" in { ComputingUnitManagingResource.terminateIdleKubernetesComputingUnits() shouldBe empty } @@ -285,6 +325,24 @@ class ComputingUnitManagingResourceSpec .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED } + it should "garbage collect disappeared Kubernetes units while listing computing units" in { + val disappeared = insertComputingUnit("disappeared-pod") + val local = insertComputingUnit("local-running", WorkflowComputingUnitTypeEnum.local) + val podOperations = new RecordingPodOperations(Set.empty) + + val listed = new ComputingUnitManagingResource().listComputingUnits( + sessionUser(), + podOperations + ) + + listed.map(_.computingUnit.getCuid) shouldBe List(local.getCuid) + workflowComputingUnitDao.fetchOneByCuid(disappeared.getCuid).getTerminateTime should not be null + workflowComputingUnitDao + .fetchOneByCuid(disappeared.getCuid) + .getTerminationReason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED + workflowComputingUnitDao.fetchOneByCuid(local.getCuid).getTerminateTime shouldBe null + } + it should "keep units with active status 0 or 2 running" in { val activeQueued = insertComputingUnit("active-status-0") val activeRunning = insertComputingUnit("active-status-2") From e5512847b33a24deafa1fe9112631766a12a74d3 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Thu, 9 Jul 2026 15:36:57 +0800 Subject: [PATCH 09/10] add test coverage --- .../ComputingUnitManagingResource.scala | 39 ++++++----- .../ComputingUnitManagingServiceRunSpec.scala | 6 +- .../ComputingUnitManagingResourceSpec.scala | 65 ++++++++++--------- 3 files changed, 63 insertions(+), 47 deletions(-) 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 667a16061df..2a8d74290bb 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 @@ -77,19 +77,28 @@ object ComputingUnitManagingResource { .getInstance() .createDSLContext() - private[resource] case class IdleComputingUnitCleanupConfig( - enabled: Boolean, - idleTimeoutMinutes: Long - ) + private[resource] final class IdleComputingUnitCleanupConfig( + val enabled: Boolean, + val idleTimeoutMinutes: Long + ) { + def copy( + enabled: Boolean = this.enabled, + idleTimeoutMinutes: Long = this.idleTimeoutMinutes + ): IdleComputingUnitCleanupConfig = + new IdleComputingUnitCleanupConfig(enabled, idleTimeoutMinutes) + } private[resource] trait KubernetesPodOperations { val podExists: Int => Boolean val deletePod: Int => Unit } - private object DefaultKubernetesPodOperations extends KubernetesPodOperations { - override val podExists: Int => Boolean = KubernetesClient.podExists - override val deletePod: Int => Unit = KubernetesClient.deletePod + private[resource] object DefaultKubernetesPodOperations extends KubernetesPodOperations { + private[resource] var podExistsDelegate: Int => Boolean = KubernetesClient.podExists + private[resource] var deletePodDelegate: Int => Unit = KubernetesClient.deletePod + + override val podExists: Int => Boolean = cuid => podExistsDelegate(cuid) + override val deletePod: Int => Unit = cuid => deletePodDelegate(cuid) } private[resource] def lastComputingUnitActivityTime( @@ -112,7 +121,7 @@ object ComputingUnitManagingResource { def terminateIdleKubernetesComputingUnits(): List[TerminatedComputingUnitInfo] = terminateIdleKubernetesComputingUnits( - IdleComputingUnitCleanupConfig( + new IdleComputingUnitCleanupConfig( KubernetesConfig.kubernetesComputingUnitEnabled, KubernetesConfig.computingUnitIdleTimeoutMinutes ), @@ -182,7 +191,7 @@ object ComputingUnitManagingResource { cuDao.update(unit) val owner = Option(userDao.fetchOneByUid(unit.getUid)) Some( - TerminatedComputingUnitInfo( + new TerminatedComputingUnitInfo( cuid = unit.getCuid, name = unit.getName, uid = unit.getUid, @@ -256,12 +265,12 @@ object ComputingUnitManagingResource { .get ) - case class TerminatedComputingUnitInfo( - cuid: Integer, - name: String, - uid: Integer, - username: Option[String], - reason: WorkflowComputingUnitTerminationReasonEnum + final class TerminatedComputingUnitInfo( + val cuid: Integer, + val name: String, + val uid: Integer, + val username: Option[String], + val reason: WorkflowComputingUnitTerminationReasonEnum ) case class WorkflowComputingUnitCreationParams( 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 621e9945f34..8656a46852b 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 @@ -176,7 +176,7 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers with terminateIdleComputingUnits = () => { cleanupInvocations += 1 List( - TerminatedComputingUnitInfo( + new TerminatedComputingUnitInfo( cuid = 3, name = "scheduled-idle", uid = 30, @@ -220,14 +220,14 @@ class ComputingUnitManagingServiceRunSpec extends AnyFlatSpec with Matchers with ComputingUnitManagingService.runIdleComputingUnitCleanup( () => List( - TerminatedComputingUnitInfo( + new TerminatedComputingUnitInfo( cuid = 1, name = "idle-a", uid = 10, username = Some("alice"), reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED ), - TerminatedComputingUnitInfo( + new TerminatedComputingUnitInfo( cuid = 2, name = "idle-b", uid = 20, 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 index b76612102c1..d6769008bcd 100644 --- 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 @@ -47,6 +47,7 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ WorkflowVersion } import org.apache.texera.service.resource.ComputingUnitManagingResource.{ + DefaultKubernetesPodOperations, IdleComputingUnitCleanupConfig, KubernetesPodOperations, TerminatedComputingUnitInfo @@ -70,7 +71,7 @@ class ComputingUnitManagingResourceSpec private val testWorkflowId = 820000 + scala.util.Random.nextInt(10000) private val now = new Timestamp(TimeUnit.DAYS.toMillis(20)) private val cleanupConfig = - IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + new IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) private var userDao: UserDao = _ private var workflowDao: WorkflowDao = _ @@ -201,43 +202,49 @@ class ComputingUnitManagingResourceSpec override val deletePod: Int => Unit = cuid => deletedPods = deletedPods :+ cuid } - "cleanup value objects" should "support generated case class operations" in { - val config = IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) - config.copy(enabled = false) shouldBe IdleComputingUnitCleanupConfig( - enabled = false, - idleTimeoutMinutes = 60 - ) - IdleComputingUnitCleanupConfig.unapply(config) shouldBe Some((true, 60)) - config.productIterator.toList shouldBe List(true, 60) + "cleanup value objects" should "expose cleanup configuration and termination info fields" in { + val config = new IdleComputingUnitCleanupConfig(enabled = true, idleTimeoutMinutes = 60) + config.enabled shouldBe true + config.idleTimeoutMinutes shouldBe 60 + + val disabledConfig = config.copy(enabled = false) + disabledConfig.enabled shouldBe false + disabledConfig.idleTimeoutMinutes shouldBe 60 + + val shorterTimeoutConfig = config.copy(idleTimeoutMinutes = 30) + shorterTimeoutConfig.enabled shouldBe true + shorterTimeoutConfig.idleTimeoutMinutes shouldBe 30 - val terminated = TerminatedComputingUnitInfo( + val terminated = new TerminatedComputingUnitInfo( cuid = 1, - name = "case-class-unit", + name = "plain-unit-info", uid = testUserId, username = Some("idle-cu-owner"), reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED ) - terminated.copy(username = None).username shouldBe None - TerminatedComputingUnitInfo.unapply(terminated) shouldBe Some( - ( - 1, - "case-class-unit", - testUserId, - Some("idle-cu-owner"), - WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED - ) - ) - terminated.productIterator.toList shouldBe List( - 1, - "case-class-unit", - testUserId, - Some("idle-cu-owner"), - WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED - ) + terminated.cuid shouldBe 1 + terminated.name shouldBe "plain-unit-info" + terminated.uid shouldBe testUserId + terminated.username shouldBe Some("idle-cu-owner") + terminated.reason shouldBe WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED } "listComputingUnits" should "return empty through the public default pod operations wrapper" in { - new ComputingUnitManagingResource().listComputingUnits(sessionUser()) shouldBe empty + val originalPodExists = DefaultKubernetesPodOperations.podExistsDelegate + val originalDeletePod = DefaultKubernetesPodOperations.deletePodDelegate + var deletedPods = List.empty[Int] + try { + DefaultKubernetesPodOperations.podExistsDelegate = _ => false + DefaultKubernetesPodOperations.deletePodDelegate = cuid => deletedPods = deletedPods :+ cuid + + DefaultKubernetesPodOperations.podExists(123) shouldBe false + DefaultKubernetesPodOperations.deletePod(123) + deletedPods shouldBe List(123) + new ComputingUnitManagingResource().listComputingUnits(sessionUser()) shouldBe empty + } finally { + DefaultKubernetesPodOperations.podExistsDelegate = originalPodExists + DefaultKubernetesPodOperations.deletePodDelegate = originalDeletePod + } } "terminateIdleKubernetesComputingUnits" should "use the default disabled cleanup configuration" in { From a624db6fd3b7fcf9c929e6a2654a0c209e33a9e1 Mon Sep 17 00:00:00 2001 From: zaoduyuan Date: Tue, 18 Aug 2026 15:46:15 +0800 Subject: [PATCH 10/10] empty commit, just to trigger the PR check