diff --git a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala index fec93ca4e..35fc168bb 100644 --- a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala +++ b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala @@ -575,11 +575,10 @@ private[utils] case class KubernetesAppReport(driver: Option[Pod], executors: Se if (!livyConf.getBoolean(LivyConf.KUBERNETES_SPARK_SIDECAR_ENABLED)) { return phase } - if (phase != KubernetesApplicationState.RUNNING) { - return phase - } - // if the POD is still running, check spark container termination status - // default to pod phase if container state is indeterminate. + // When sidecars are enabled, prefer spark container's exit status over + // pod phase since a sidecar failure can mark the pod as "failed" even + // if the driver exited successfully. Default to pod phase if the spark + // container cannot be found (e.g. misconfigured container name). getTerminalState(podStatus).getOrElse(phase) } @@ -588,7 +587,7 @@ private[utils] case class KubernetesAppReport(driver: Option[Pod], executors: Se def getTerminalState(podStatus: PodStatus): Option[String] = { import scala.collection.JavaConverters._ val sparkContainerName = livyConf.get(LivyConf.KUBERNETES_SPARK_CONTAINER_NAME) - for (c <- podStatus.getContainerStatuses.asScala) { + for (c <- Option(podStatus.getContainerStatuses).map(_.asScala).getOrElse(Nil)) { if (c.getName == sparkContainerName && c.getState.getTerminated != null) { val exitCode = c.getState.getTerminated.getExitCode if (exitCode == 0) { diff --git a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala b/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala index 759cbed10..57f5d4e40 100644 --- a/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala +++ b/server/src/test/scala/org/apache/livy/utils/SparkKubernetesAppSpec.scala @@ -91,6 +91,58 @@ class SparkKubernetesAppSpec extends AnyFunSpec with LivyBaseUnitTestSuite with } } + def podWithSidecar(phase: String, terminatedExitCode: Option[Int]): Pod = { + val builder = new PodBuilder() + .withNewStatus() + .withPhase(phase) + .endStatus() + .withNewSpec().endSpec() + .withNewMetadata().endMetadata() + terminatedExitCode match { + case Some(exitCode) => + builder.editStatus() + .addNewContainerStatus() + .withName("spark-container") + .withNewState() + .withNewTerminated() + .withExitCode(exitCode) + .endTerminated() + .endState() + .endContainerStatus() + .endStatus() + case None => () + } + builder.build() + } + + it("should prefer spark container exit status over pod phase when sidecars are enabled") { + def state(pod: Pod, sidecarEnabled: Boolean): String = + KubernetesAppReport( + Some(pod), Seq.empty, IndexedSeq.empty, None, + new LivyConf(false).set(LivyConf.KUBERNETES_SPARK_SIDECAR_ENABLED, sidecarEnabled) + ).getApplicationState + + // A sidecar failure marks the pod "Failed" even though the spark container exited 0. + assertResult("succeeded") { + state(podWithSidecar("Failed", Some(0)), sidecarEnabled = true) + } + assertResult("failed") { + state(podWithSidecar("Running", Some(1)), sidecarEnabled = true) + } + // Spark container not yet terminated (or not found): fall back to pod phase. + assertResult("running") { + state(podWithSidecar("Running", None), sidecarEnabled = true) + } + // getContainerStatuses is null while the pod is pending: no NPE, fall back to phase. + assertResult("pending") { + state(podWithSidecar("Pending", None), sidecarEnabled = true) + } + // Sidecars disabled: pod phase is authoritative even if the spark container failed. + assertResult("failed") { + state(podWithSidecar("Failed", Some(0)), sidecarEnabled = false) + } + } + def livyConf(lokiEnabled: Boolean): LivyConf = new LivyConf(false) .set(LivyConf.KUBERNETES_GRAFANA_LOKI_ENABLED, lokiEnabled)