kunwp1 commented on code in PR #6046:
URL: https://github.com/apache/texera/pull/6046#discussion_r4057919798
##########
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:
If a user starts a run between the scan and this update, it might end up
deleting the pod out from a live execution. The window doesn't look tiny as
well. Can you add a guard in the `WHERE` clause to handle such case?
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -678,8 +880,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 computing unit: cuid=${unit.getCuid},
name=${unit.getName}, uid=${unit.getUid}, username=${user.getName},
reason=${terminationReason.getLiteral}"
Review Comment:
The username here seems like it's an acting user who is deleting the
computing unit. So if an admin terminates the CU, the uid becomes the owner of
the CU and the username becomes the admin's name. I think these two information
needs to be consistent in a single log.
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -61,18 +67,213 @@ 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 val logger =
LoggerFactory.getLogger(classOf[ComputingUnitManagingResource])
+
private def context: DSLContext =
SqlServer
.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] final class IdleComputingUnitCandidate(
+ val unit: WorkflowComputingUnit,
+ val username: Option[String]
+ )
+
+ private[resource] object WorkflowExecutionStatus extends Enumeration {
+ val UninitializedOrReady: Value = Value(0)
+ val Running: Value = Value(1)
+ val Paused: Value = Value(2)
+
+ def toDbStatus(status: Value): java.lang.Short =
Short.box(status.id.toShort)
+ }
+
+ 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] =
+ runIdleKubernetesComputingUnitCleanup(
+ new IdleComputingUnitCleanupConfig(
+ KubernetesConfig.kubernetesComputingUnitEnabled,
+ KubernetesConfig.computingUnitIdleTimeoutMinutes
+ ),
+ () => new Timestamp(System.currentTimeMillis()),
+ DefaultKubernetesPodOperations
+ )
+
+ private[resource] def runIdleKubernetesComputingUnitCleanup(
+ 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)
+
+ idleKubernetesComputingUnitCandidates(cutoff).flatMap(candidate =>
+ terminateIdleKubernetesComputingUnitCandidate(candidate, now,
podOperations)
+ )
+ }
+
+ private[resource] def idleKubernetesComputingUnitCandidates(
+ cutoff: Timestamp
+ ): List[IdleComputingUnitCandidate] = {
+ val activeStatuses = Seq(
+ WorkflowExecutionStatus.UninitializedOrReady,
+ WorkflowExecutionStatus.Running,
+ WorkflowExecutionStatus.Paused
+ ).map(WorkflowExecutionStatus.toDbStatus)
+
+ // 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.in(activeStatuses: _*))
Review Comment:
I reopened this conversation again. I don't think
`reconcileVanishedKubernetesUnits` covers this case because it requires someone
to trigger computing unit listing endpoint. It's not called automatically. The
case that worries me is a live pod whose execution row is stuck at status 1
(coordinator died in place, OOM inside the container). bool_or(status NOT IN
(3,4,5)) has no time bound, so that kind of computing units doesn't get
reclaimed at all. Can you create a follow-up issue that there is a known issue
from this PR and needs to be fixed?
--
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]
