eugenegujing commented on code in PR #7944:
URL: https://github.com/apache/texera/pull/7944#discussion_r3877429918
##########
computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala:
##########
@@ -55,42 +62,140 @@ object ComputingUnitHelpers {
}
def getComputingUnitStatus(unit: WorkflowComputingUnit): ComputingUnitState =
- singleUnitStatus(unit, KubernetesClient)
+ getComputingUnitStatusWithReason(unit)._1
+
+ /** Single-unit status plus the owner-facing reason (see
[[kubernetesStatusAndReason]]). */
+ def getComputingUnitStatusWithReason(
+ unit: WorkflowComputingUnit
+ ): (ComputingUnitState, Option[String]) =
+ singleUnitStatusAndReason(unit, KubernetesClient)
/**
* Single-unit status via a per-unit pod lookup (a targeted GET, cheaper
than listing the whole
* namespace). The client is a by-name parameter — not the global singleton
— so the kubernetes
* branch is unit-testable with a stub and the local/unknown branches never
force the singleton;
- * the public overload binds the production [[KubernetesClient]]. (Metrics
has no analogous seam:
+ * the public overloads bind the production [[KubernetesClient]]. (Metrics
has no analogous seam:
* its per-unit lookup already fans out to the whole namespace and the bulk
(unit, podMetrics)
* overload already covers the cpu/memory resolution, so nothing there is
worth pinning.)
*/
- private[util] def singleUnitStatus(
+ private[util] def singleUnitStatusAndReason(
unit: WorkflowComputingUnit,
k8s: => KubernetesClient
- ): ComputingUnitState = {
+ ): (ComputingUnitState, Option[String]) = {
unit.getType match {
// Local CUs are always “running”
case WorkflowComputingUnitTypeEnum.local =>
- Running
+ (Running, None)
- // Kubernetes CUs – only explicit “Running” counts as running
+ // Kubernetes CUs – resolved from the pod's status snapshot
case WorkflowComputingUnitTypeEnum.kubernetes =>
- // Guard the pod status the same way the bulk getAllPodPhases does: a
pod with no
- // status yet has a null getStatus, so map through Option to avoid an
NPE.
val client = k8s
- val phaseOpt = client
- .getPodByName(client.generatePodName(unit.getCuid))
- .flatMap(pod => Option(pod.getStatus).map(_.getPhase))
-
- if (phaseOpt.contains("Running")) Running else Pending
+ kubernetesStatusAndReason(
+ client
+ .getPodByName(client.generatePodName(unit.getCuid))
+ .map(PodStatusSnapshot.fromPod)
+ )
// Any other (unknown) type is treated as pending
case _ =>
- Pending
+ (Pending, None)
+ }
+ }
+
+ // Owner-facing wording for each failure mode. Deliberately actionable
prose, never a raw
+ // Kubernetes dump; buildDashboardUnit withholds these from non-owners
entirely.
+ private val ImagePullWaitingReasons = Set("ImagePullBackOff",
"ErrImagePull", "InvalidImageName")
+ private val EvictedDiskReason =
+ "The computing unit was evicted because it ran out of local disk storage.
Consider " +
+ "storing less data on the unit's local file system, or recreate it with
more storage."
+ private val ImagePullReason =
+ "The computing unit's image could not be pulled. Please recreate the unit
or contact " +
+ "an administrator."
+ private val CrashLoopOomReason =
+ "The computing unit keeps crashing because it runs out of memory. Please
terminate it " +
+ "and recreate it with a higher memory limit."
+ private val GenericFailedReason =
+ "The computing unit stopped unexpectedly. Please terminate and recreate
it, or contact " +
+ "an administrator."
+ private val UnknownStateReason =
+ "The state of the computing unit cannot be determined (its node may be
unreachable)."
+ private val UnschedulableReason =
+ "The computing unit is waiting for cluster resources to become available."
+
+ private def evictedReason(podMessage: Option[String]): String = {
+ val mentionsDisk =
+ podMessage.exists { message =>
+ val lower = message.toLowerCase
+ lower.contains("ephemeral") || lower.contains("disk")
+ }
+ if (mentionsDisk) EvictedDiskReason
+ else {
+ // First sentence of the cluster's message, capped so the tooltip stays
readable.
+ val shortReason = podMessage
+ .map(_.takeWhile(_ != '.').trim)
+ .filter(_.nonEmpty)
+ .map(sentence => if (sentence.length > 120) sentence.take(120).trim +
"..." else sentence)
+ .getOrElse("Evicted")
+ s"The computing unit was evicted by the cluster ($shortReason). Consider
recreating it."
}
}
+ private def crashLoopReason(restartCount: Int): String =
+ s"The computing unit is repeatedly crashing (restarted $restartCount
times). Please " +
+ "terminate and recreate it, or contact an administrator."
+
+ private def recoveredOomWarning(restartCount: Int): String =
+ s"The last run was terminated because the computing unit ran out of memory
(restarted " +
+ s"$restartCount times). Consider recreating the unit with a higher
memory limit before " +
+ "running the same workload."
+
+ /**
+ * Pure (snapshot -> state, reason) mapping, mirroring the Kubernetes pod
lifecycle. An absent
+ * pod stays Pending — exactly today's behavior — because the vanish
reconciliation, not this
+ * mapping, is what retires units whose pods are gone.
+ *
+ * Note the restartPolicy-Always subtlety: an OOM-killed container restarts
in place with the
+ * pod phase still "Running", so OOM kills and crash loops are read from
the container-level
+ * fields, and a waiting-state failure takes precedence over the
recovered-OOM warning.
+ */
+ private[util] def kubernetesStatusAndReason(
+ snapshotOpt: Option[PodStatusSnapshot]
+ ): (ComputingUnitState, Option[String]) =
+ snapshotOpt match {
+ case None => (Pending, None)
+ case Some(snapshot) =>
+ val phase = snapshot.phase.getOrElse("")
+ val imagePullFailed =
+
snapshot.containers.exists(_.waitingReason.exists(ImagePullWaitingReasons.contains))
+ val crashLooping =
snapshot.containers.find(_.waitingReason.contains("CrashLoopBackOff"))
+ val oomKilled =
snapshot.containers.find(_.lastTerminatedReason.contains("OOMKilled"))
+
+ if (snapshot.terminating)
+ (Terminating, None)
+ else if (phase == "Failed" && snapshot.podReason.contains("Evicted"))
+ (Failed, Some(evictedReason(snapshot.podMessage)))
+ else if (imagePullFailed)
+ (Failed, Some(ImagePullReason))
+ else if (crashLooping.isDefined) {
+ val container = crashLooping.get
+ if (container.lastTerminatedReason.contains("OOMKilled"))
+ (Failed, Some(CrashLoopOomReason))
+ else
+ (Failed, Some(crashLoopReason(container.restartCount)))
+ } else if (phase == "Failed")
+ (Failed, Some(GenericFailedReason))
+ else if (phase == "Unknown")
+ (Unknown, Some(UnknownStateReason))
+ else if (phase == "Pending" && snapshot.unschedulable)
+ (Pending, Some(UnschedulableReason))
+ else if (phase == "Running" && oomKilled.isDefined)
+ (Running, Some(recoveredOomWarning(oomKilled.get.restartCount)))
+ else if (phase == "Running")
+ (Running, None)
Review Comment:
Fixed in a4f24d0ce6ca4176baefc6bd74129279a04bb72a
--
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]
