yrenat commented on code in PR #6046:
URL: https://github.com/apache/texera/pull/6046#discussion_r4058842152
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -61,18 +67,207 @@ import org.apache.texera.service.util.{
KubernetesClient
}
import org.jooq.{DSLContext, EnumType}
+import org.jooq.impl.DSL.{boolOr, max}
+import org.slf4j.LoggerFactory
import play.api.libs.json._
import java.sql.Timestamp
import scala.annotation.unused
import scala.jdk.CollectionConverters.CollectionHasAsScala
+import scala.util.control.NonFatal
object ComputingUnitManagingResource {
+ private[resource] val logger =
LoggerFactory.getLogger(classOf[ComputingUnitManagingResource])
+
private def context: DSLContext =
SqlServer
.getInstance()
.createDSLContext()
+ private[resource] case class IdleComputingUnitCandidate(
+ unit: WorkflowComputingUnit,
+ username: Option[String]
+ )
+
+ /**
+ * The codes persisted in `workflow_executions.status`. These are the
collapsed codes produced
+ * by amber's `Utils.maptoStatusCode`, NOT the ordinals of
`WorkflowAggregatedState`, and this
+ * service cannot depend on the amber module to reuse either. Keep this in
sync with
+ * `Utils.maptoStatusCode`.
+ *
+ * Only the terminal codes are listed, and the sweep treats every other
code as an execution
+ * still in flight. That direction matters: `maptoStatusCode` collapses
PAUSING, RESUMING,
+ * UNKNOWN and TERMINATED alike to -1, so enumerating the non-terminal
codes instead would
+ * leave a paused-or-resuming execution looking idle and get its computing
unit deleted out
+ * from under its owner. An unrecognised code now only costs a unit one
more sweep.
+ */
+ private[resource] object TerminalWorkflowExecutionStatus extends Enumeration
{
+ val Completed: Value = Value(3)
+ val Failed: Value = Value(4)
+ val Killed: Value = Value(5)
+
+ def dbStatuses: Seq[java.lang.Short] =
+ values.toSeq.map(status => Short.box(status.id.toShort))
+ }
+
+ 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)
+
+ /**
+ * Terminates every Kubernetes computing unit whose last execution activity
is older than
+ * `idleTimeoutMinutes`, returning the units terminated so the caller can
log their owners.
+ */
+ def terminateIdleKubernetesComputingUnits(
+ idleTimeoutMinutes: Long
+ ): List[TerminatedComputingUnitInfo] =
+ terminateIdleKubernetesComputingUnits(
+ idleTimeoutMinutes,
+ new Timestamp(System.currentTimeMillis()),
+ KubernetesClient
+ )
+
+ /**
+ * The client is a by-name parameter -- not the global singleton -- so the
sweep is unit-testable
+ * with a stub and a sweep that terminates nothing never forces the
singleton; the public
+ * overload binds the production [[KubernetesClient]]. Same seam as
+ * [[org.apache.texera.service.util.ComputingUnitHelpers.singleUnitStatus]].
+ */
+ private[resource] def terminateIdleKubernetesComputingUnits(
+ idleTimeoutMinutes: Long,
+ now: Timestamp,
+ k8s: => KubernetesClient
+ ): List[TerminatedComputingUnitInfo] = {
+ val cutoff = new Timestamp(now.getTime - idleTimeoutMinutes * 60 * 1000)
+ idleKubernetesComputingUnitCandidates(cutoff).flatMap(candidate =>
+ terminateIdleKubernetesComputingUnitCandidate(candidate, now, k8s)
+ )
+ }
+
+ private[resource] def idleKubernetesComputingUnitCandidates(
+ cutoff: Timestamp
+ ): List[IdleComputingUnitCandidate] = {
+ // All three questions asked per computing unit -- is any execution still
active, when did an
+ // execution last report progress, when did one last start -- are
aggregates over the same rows
+ // grouped by the same key, so one grouped query answers them for every
unit at once. The left
+ // joins keep units that have no executions (both max() are NULL) and
units whose owner row is
+ // gone (name is NULL), matching what a per-unit scan would produce.
+ val latestUpdateTime = max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME)
+ val latestStartTime = max(WORKFLOW_EXECUTIONS.STARTING_TIME)
+ val hasActiveExecution =
+
boolOr(WORKFLOW_EXECUTIONS.STATUS.notIn(TerminalWorkflowExecutionStatus.dbStatuses:
_*))
+
+ withTransaction(context) { ctx =>
+ ctx
+ .select(
+ WORKFLOW_COMPUTING_UNIT.asterisk(),
+ USER.NAME,
+ latestUpdateTime,
+ latestStartTime,
+ hasActiveExecution
+ )
+ .from(WORKFLOW_COMPUTING_UNIT)
+ .leftJoin(WORKFLOW_EXECUTIONS)
+ .on(WORKFLOW_EXECUTIONS.CUID.eq(WORKFLOW_COMPUTING_UNIT.CUID))
+ .leftJoin(USER)
+ .on(USER.UID.eq(WORKFLOW_COMPUTING_UNIT.UID))
+ .where(
+ WORKFLOW_COMPUTING_UNIT.TYPE
+ .eq(WorkflowComputingUnitTypeEnum.kubernetes)
+ .and(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME.isNull)
+ )
+ .groupBy(WORKFLOW_COMPUTING_UNIT.CUID, USER.NAME)
+ .fetch()
+ .asScala
+ .flatMap { record =>
+ val unit =
record.into(WORKFLOW_COMPUTING_UNIT).into(classOf[WorkflowComputingUnit])
+ val lastExecutionTime = lastComputingUnitActivityTime(
+ unit,
+ Option(record.get(latestUpdateTime)),
+ Option(record.get(latestStartTime))
+ )
+
+ // bool_or over zero matching executions yields NULL, which means
"no active execution"
+ val active =
Option(record.get(hasActiveExecution)).exists(_.booleanValue())
+ if (shouldTerminateIdleComputingUnit(active, lastExecutionTime,
cutoff)) {
+ Some(
+ IdleComputingUnitCandidate(
+ unit,
+ Option(record.get(USER.NAME)).filter(_.nonEmpty)
+ )
+ )
+ } else {
+ None
+ }
+ }
+ .toList
+ }
+ }
+
+ private[resource] def terminateIdleKubernetesComputingUnitCandidate(
+ candidate: IdleComputingUnitCandidate,
+ terminationTime: Timestamp,
+ k8s: => KubernetesClient
+ ): Option[TerminatedComputingUnitInfo] = {
+ val unit = candidate.unit
+ val cuid = unit.getCuid
+ val reason = WorkflowComputingUnitTerminationReasonEnum.GARBAGE_COLLECTED
+ try {
+ withTransaction(context) { ctx =>
+ // Stamp the row first and delete the pod second, within one
transaction per unit. The
+ // guards the scan applied are repeated in the WHERE clause, so a unit
a user terminated
+ // in the meantime updates zero rows and is left alone. Deleting an
absent pod is a no-op
+ // and deleting a live one is idempotent, so letting a delete failure
roll the stamp back
+ // only costs a retry next round -- whereas stamping after a failed
delete would leave a
+ // live pod behind a row that says terminated.
+ val marked = ctx
+ .update(WORKFLOW_COMPUTING_UNIT)
+ .set(WORKFLOW_COMPUTING_UNIT.TERMINATE_TIME, terminationTime)
+ .set(WORKFLOW_COMPUTING_UNIT.TERMINATION_REASON, reason)
+ .where(
Review Comment:
Added `.andNot(liveExecutionExists(cuid, cutoff))` to the UPDATE. It checks
`workflow_executions` for any row with a non-terminal status, or with
`starting_time` or `last_update_time` at or after the cutoff. If such a row
shows up after the scan, the UPDATE matches nothing and the pod is left alone.
--
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]
