aicam commented on code in PR #6046:
URL: https://github.com/apache/texera/pull/6046#discussion_r3846121035
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -61,18 +63,200 @@ 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
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] 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)
+
+ idleKubernetesComputingUnitCandidates(cutoff).flatMap(candidate =>
+ terminateIdleKubernetesComputingUnitCandidate(candidate, now,
podOperations)
+ )
+ }
+
+ private[resource] def idleKubernetesComputingUnitCandidates(
+ cutoff: Timestamp
+ ): List[IdleComputingUnitCandidate] = {
+ val activeStatuses = Seq(Short.box(0), Short.box(1), Short.box(2))
Review Comment:
Correctness here is actually fine, but it took real digging to confirm,
which is the problem.
These look wrong against `WorkflowAggregatedState` (where 3/4/5 are
PAUSING/PAUSED/RESUMING). They're right only because the column stores the
*collapsed* codes from `Utils.maptoStatusCode` — 0=UNINITIALIZED/READY,
1=RUNNING, 2=PAUSED, 3=COMPLETED, 4=FAILED, 5=KILLED — so `{0,1,2}` is the
correct non-terminal set.
Nothing links these literals to that mapping. Add a state to
`maptoStatusCode` and this silently keeps the old semantics. Worth noting the
line just above does the typed thing
(`TYPE.eq(WorkflowComputingUnitTypeEnum.kubernetes)`) only because `type` is a
real Postgres enum and jOOQ generated one for it; `status` is `SMALLINT` so we
get no help.
`WorkflowAggregatedState` can't be used directly — its ordinals aren't
what's persisted, and it's generated in the amber module, which this service
doesn't depend on. Minimal fix: a named `val` plus a comment pointing at
`Utils.maptoStatusCode`. Better: a small `ExecutionStatus` enum in `common/dao`
(on the classpath of both amber and this service) exposing a `nonTerminal` set,
with `maptoStatusCode` delegating to it, so the encoding is written down once.
That would also settle the `Byte` vs `Short` mismatch — `maptoStatusCode`
returns `Byte` while the jOOQ field is `Short`, which only compiles today via
Scala's silent numeric widening.
Separate but related: this predicate has no staleness bound. Nothing
rewrites `workflow_executions.status` when a coordinator dies ungracefully, so
the row sits at 0/1/2 indefinitely and that CU becomes permanently exempt from
the sweep — precisely the orphaned-pod case this PR exists to reclaim. Consider
anding a recency bound on `coalesce(last_update_time, starting_time)`.
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala:
##########
@@ -61,18 +63,200 @@ 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
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] 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)
+
+ idleKubernetesComputingUnitCandidates(cutoff).flatMap(candidate =>
+ terminateIdleKubernetesComputingUnitCandidate(candidate, now,
podOperations)
+ )
+ }
+
+ private[resource] def idleKubernetesComputingUnitCandidates(
+ cutoff: Timestamp
+ ): List[IdleComputingUnitCandidate] = {
+ val activeStatuses = Seq(Short.box(0), Short.box(1), Short.box(2))
+
+ withTransaction(context) { ctx =>
+ 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)
Review Comment:
This is three round trips per non-terminated Kubernetes CU, plus
`fetchOneByUid` per qualifying candidate below — so **3N + C + 1** queries per
scan, all inside one `withTransaction` that pins one of the 10 Hikari
connections (`common/dao/.../SqlServer.scala:51`) for the entire duration.
Two things make it more expensive than it reads:
1. **The two `max()` queries have byte-identical `FROM`/`WHERE`** and differ
only in the projected column. They collapse to `select(max(LAST_UPDATE_TIME),
max(STARTING_TIME))` with no other change — a one-line fix that takes this to
2N.
2. **`workflow_executions.cuid` has no index.** It's a plain FK column added
in `sql/updates/07.sql`, and Postgres only indexes the *referenced* side of an
FK, never the referencing side. So each of the 3N queries is a sequential scan
of the executions table. Worth `CREATE INDEX ON workflow_executions (cuid)`
regardless of the rest — it helps every other query on that column too.
All three questions are aggregates over the same rows grouped by the same
key, so the whole loop collapses to a single query:
```scala
ctx.select(
WORKFLOW_COMPUTING_UNIT.asterisk(),
USER.NAME,
max(WORKFLOW_EXECUTIONS.LAST_UPDATE_TIME),
max(WORKFLOW_EXECUTIONS.STARTING_TIME),
boolOr(WORKFLOW_EXECUTIONS.STATUS.in(activeStatuses: _*))
)
.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)
```
`LEFT JOIN` keeps CUs with zero executions (both `max()`es come back `NULL`,
`boolOr` `NULL` → false), matching what the per-unit code produces today. I'd
keep `shouldTerminateIdleComputingUnit` in Scala rather than pushing the
predicate into `HAVING` — the pure-function testability is worth more than the
last bit of filtering.
I realize the current shape is partly a consequence of
`lastComputingUnitActivityTime` taking two pre-fetched `Option[Timestamp]`s,
which is a good testability instinct but forces the caller to fetch them
separately. Passing the aggregated row in would keep that function just as
testable.
At 10 CUs none of this matters. At a few hundred against a large executions
table it's a long-held transaction on a recurring timer — and the scan duration
is also what widens the window for the snapshot to go stale before the pod
delete runs further down.
--
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]
