fix(kubernetes): terminate idle computing units

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}
 

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")

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.{
@@ -32,9 +32,44 @@ 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
 
 class ComputingUnitManagingService extends Application[ComputingUnitManagingServiceConfiguration] {
+  private val logger = LoggerFactory.getLogger(classOf[ComputingUnitManagingService])
+
+  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(
+      scheduleWithFixedDelay,
+      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]
@@ -59,11 +94,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)
@@ -74,12 +105,83 @@ class ComputingUnitManagingService extends Application[ComputingUnitManagingServ
       "ComputingUnitManagingService"
     )
 
+    registerIdleComputingUnitCleanup(environment)
+
     // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL
     RequestLoggingFilter.register(environment.getApplicationContext)
   }
 }
 
 object ComputingUnitManagingService {
+  private[service] def shouldScheduleIdleComputingUnitCleanup(
+      kubernetesComputingUnitEnabled: Boolean,
+      idleTimeoutMinutes: Long
+  ): 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,
+      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,
+      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", "."))

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

@@ -42,9 +42,11 @@ 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,
   UserRoleEnum,
+  WorkflowComputingUnitTerminationReasonEnum,
   WorkflowComputingUnitTypeEnum
 }
 import org.apache.texera.dao.jooq.generated.tables.daos.{
@@ -53,6 +55,7 @@ import org.apache.texera.dao.jooq.generated.tables.daos.{
   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.util.{
   ComputingUnitHelpers,
@@ -61,6 +64,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
@@ -73,6 +78,136 @@ object ComputingUnitManagingResource {
       .getInstance()
       .createDSLContext()
 
+  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[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(
+      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(
+      new 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 = 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 =>
+      val cuDao = new WorkflowComputingUnitDao(ctx.configuration())
+      val userDao = new UserDao(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
+        .flatMap { 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(max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME))
+            .from(WORKFLOW_EXECUTIONS)
+            .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid))
+            .fetchOne(0, classOf[Timestamp])
+          val latestStartTime = ctx
+            .select(max(WORKFLOW_EXECUTIONS.STARTING_TIME))
+            .from(WORKFLOW_EXECUTIONS)
+            .where(WORKFLOW_EXECUTIONS.CUID.eq(cuid))
+            .fetchOne(0, classOf[Timestamp])
+          val lastExecutionTime = lastComputingUnitActivityTime(
+            unit,
+            Option(latestUpdateTime),
+            Option(latestStartTime)
+          )
+
+          if (shouldTerminateIdleComputingUnit(hasActiveExecution, lastExecutionTime, cutoff)) {
+            if (podOperations.podExists(cuid)) {
+              podOperations.deletePod(cuid)
+            }
+            unit.setTerminateTime(now)
+            unit.setTerminationReason(WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED)
+            cuDao.update(unit)
+            val owner = Option(userDao.fetchOneByUid(unit.getUid))
+            Some(
+              new TerminatedComputingUnitInfo(
+                cuid = unit.getCuid,
+                name = unit.getName,
+                uid = unit.getUid,
+                username = owner.flatMap(u => Option(u.getName).filter(_.nonEmpty)),
+                reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED
+              )
+            )
+          } else {
+            None
+          }
+        }
+        .toList
+    }
+  }
+
   private def icebergEnvironmentVariables: Map[String, Any] = {
     val base = Map[String, Any](
       EnvironmentalVariable.ENV_ICEBERG_CATALOG_TYPE -> StorageConfig.icebergCatalogType
@@ -131,6 +266,14 @@ object ComputingUnitManagingResource {
         .get
     )
 
+  final class TerminatedComputingUnitInfo(
+      val cuid: Integer,
+      val name: String,
+      val uid: Integer,
+      val username: Option[String],
+      val reason: WorkflowComputingUnitTerminationReasonEnum
+  )
+
   case class WorkflowComputingUnitCreationParams(
       name: String,
       unitType: String,
@@ -177,6 +320,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())
@@ -467,6 +611,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())
@@ -621,8 +771,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()
   }