Ma77Ball commented on code in PR #6046:
URL: https://github.com/apache/texera/pull/6046#discussion_r3607785766
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -69,6 +77,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 =>
Review Comment:
A single DB transaction appears to handle the full scan and deletion of
pods. I recommend deleting pods outside the transaction or committing per unit,
so one failure does not undo the whole batch.
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -173,6 +319,7 @@ object ComputingUnitManagingResource {
@Produces(Array(MediaType.APPLICATION_JSON))
@Path("/computing-unit")
class ComputingUnitManagingResource {
+ private val logger =
LoggerFactory.getLogger(classOf[ComputingUnitManagingService])
Review Comment:
With this setup, the log line in `terminateComputingUnit` will emit under
the service's logger name instead of the resource's (which is the class it is
created under). I recommend using the resource class instead here:
```suggestion
private val logger =
LoggerFactory.getLogger(classOf[ComputingUnitManagingResource])
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
