Ma77Ball commented on code in PR #6046:
URL: https://github.com/apache/texera/pull/6046#discussion_r4066467467
##########
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:
_*))
Review Comment:
`bool_or(STATUS notIn (3,4,5))` treats every non-terminal code as an active
execution, and `Utils.maptoStatusCode` collapses `TERMINATED` (and `UNKNOWN`)
to -1, which `ExecutionStateStore.setStatus` persists. So a unit whose latest
execution ended in `TERMINATED` reports `hasActiveExecution = true` and is
never reclaimed: not "one more sweep" as the docstring above states, but
permanently. That is exactly the abnormally-ended unit an idle sweep should
catch.
This cannot be fixed purely here (the -1 overload lives in amber), so please
either document the gap accurately or track distinguishing `TERMINATED`, and
correct the "only costs a unit one more sweep" line, which holds for genuinely
unrecognised codes but not for `TERMINATED`.
--
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]
