[SPARK-30821][K8S]Handle executor failure with multiple containers

docs/running-on-kubernetes.md

@@ -1161,6 +1161,14 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
   <td>3.0.0</td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.executor.checkAllContainers</code></td>
+  <td>false</td>
+  <td>
+  Specify whether executor pods should be check all containers (including sidecars) or only the executor container when determining the pod status.
+  </td>
+  <td>3.1.0</td>
+</tr>
 <tr>
   <td><code>spark.kubernetes.submission.connectionTimeout</code></td>
   <td>10000</td>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -417,6 +417,14 @@ private[spark] object Config extends Logging {
       .stringConf
       .createOptional
 
+  val KUBERNETES_EXECUTOR_CHECK_ALL_CONTAINERS =
+    ConfigBuilder("spark.kubernetes.executor.checkAllContainers")
+      .doc("If set to true, all containers in the executor pod will be checked when reporting" +
+        "executor status.")
+      .version("3.1.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val KUBERNETES_DRIVER_LABEL_PREFIX = "spark.kubernetes.driver.label."
   val KUBERNETES_DRIVER_ANNOTATION_PREFIX = "spark.kubernetes.driver.annotation."
   val KUBERNETES_DRIVER_SERVICE_ANNOTATION_PREFIX = "spark.kubernetes.driver.service.annotation."

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshot.scala

@@ -18,6 +18,7 @@ package org.apache.spark.scheduler.cluster.k8s
 
 import java.util.Locale
 
+import io.fabric8.kubernetes.api.model.ContainerStateTerminated
 import io.fabric8.kubernetes.api.model.Pod
 
 import org.apache.spark.deploy.k8s.Constants._
@@ -37,13 +38,18 @@ private[spark] case class ExecutorPodsSnapshot(executorPods: Map[Long, ExecutorP
 }
 
 object ExecutorPodsSnapshot extends Logging {
+  private var shouldCheckAllContainers: Boolean = _
 
   def apply(executorPods: Seq[Pod]): ExecutorPodsSnapshot = {
     ExecutorPodsSnapshot(toStatesByExecutorId(executorPods))
   }
 
   def apply(): ExecutorPodsSnapshot = ExecutorPodsSnapshot(Map.empty[Long, ExecutorPodState])
 
+  def setShouldCheckAllContainers(watchAllContainers: Boolean): Unit = {
+    shouldCheckAllContainers = watchAllContainers
+  }
+
   private def toStatesByExecutorId(executorPods: Seq[Pod]): Map[Long, ExecutorPodState] = {
     executorPods.map { pod =>
       (pod.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL).toLong, toState(pod))
@@ -59,7 +65,15 @@ object ExecutorPodsSnapshot extends Logging {
         case "pending" =>
           PodPending(pod)
         case "running" =>
-          PodRunning(pod)
+          if (shouldCheckAllContainers &&
+            "Never" == pod.getSpec.getRestartPolicy &&
+            pod.getStatus.getContainerStatuses.stream
+              .map[ContainerStateTerminated](cs => cs.getState.getTerminated)
+              .anyMatch(t => t != null && t.getExitCode != 0)) {
+            PodFailed(pod)
+          } else {
+            PodRunning(pod)
+          }
         case "failed" =>
           PodFailed(pod)
         case "succeeded" =>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala

@@ -95,10 +95,13 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val schedulerExecutorService = ThreadUtils.newDaemonSingleThreadScheduledExecutor(
       "kubernetes-executor-maintenance")
 
+    ExecutorPodsSnapshot.setShouldCheckAllContainers(
+      sc.conf.get(KUBERNETES_EXECUTOR_CHECK_ALL_CONTAINERS))
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(
         "kubernetes-executor-snapshots-subscribers", 2)
     val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(subscribersExecutor)
+
     val removedExecutorsCache = CacheBuilder.newBuilder()
       .expireAfterWrite(3, TimeUnit.MINUTES)
       .build[java.lang.Long, java.lang.Long]()

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/DeterministicExecutorPodsSnapshotsStore.scala

@@ -21,6 +21,8 @@ import scala.collection.mutable
 
 class DeterministicExecutorPodsSnapshotsStore extends ExecutorPodsSnapshotsStore {
 
+  ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
+
   private val snapshotsBuffer = mutable.Buffer.empty[ExecutorPodsSnapshot]
   private val subscribers = mutable.Buffer.empty[Seq[ExecutorPodsSnapshot] => Unit]
 

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorLifecycleTestUtils.scala

@@ -71,6 +71,33 @@ object ExecutorLifecycleTestUtils {
       .build()
   }
 
+  /**
+   * [SPARK-30821]
+   * This creates a pod with one container in running state and one container in failed
+   * state (terminated with non-zero exit code). This pod is used for unit-testing the
+   * spark.kubernetes.executor.checkAllContainers Spark Conf.
+   */
+  def runningExecutorWithFailedContainer(executorId: Long): Pod = {
+    new PodBuilder(podWithAttachedContainerForId(executorId))
+      .editOrNewStatus()
+        .withPhase("running")
+        .addNewContainerStatus()
+          .withNewState()
+            .withNewTerminated()
+              .withExitCode(1)
+            .endTerminated()
+          .endState()
+        .endContainerStatus()
+        .addNewContainerStatus()
+          .withNewState()
+            .withNewRunning()
+            .endRunning()
+          .endState()
+        .endContainerStatus()
+      .endStatus()
+      .build()
+  }
+
   def succeededExecutor(executorId: Long): Pod = {
     new PodBuilder(podWithAttachedContainerForId(executorId))
       .editOrNewStatus()
@@ -112,7 +139,10 @@ object ExecutorLifecycleTestUtils {
         .addToLabels(SPARK_APP_ID_LABEL, TEST_SPARK_APP_ID)
         .addToLabels(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
         .addToLabels(SPARK_EXECUTOR_ID_LABEL, executorId.toString)
-        .endMetadata()
+      .endMetadata()
+      .editOrNewSpec()
+        .withRestartPolicy("Never")
+      .endSpec()
       .build()
     val container = new ContainerBuilder()
       .withName("spark-executor")

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSuite.scala

@@ -16,31 +16,55 @@
  */
 package org.apache.spark.scheduler.cluster.k8s
 
+import io.fabric8.kubernetes.api.model.Pod
+
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils._
 
 class ExecutorPodsSnapshotSuite extends SparkFunSuite {
 
+  def testCase(pod: Pod, state: Pod => ExecutorPodState): (Pod, ExecutorPodState) =
+    (pod, state(pod))
+
+  def doTest(testCases: Seq[(Pod, ExecutorPodState)]): Unit = {
+    val snapshot = ExecutorPodsSnapshot(testCases.map(_._1))
+    for (((_, state), i) <- testCases.zipWithIndex) {
+      assertResult(state.getClass.getName, s"executor ID $i") {
+        snapshot.executorPods(i).getClass.getName
+      }
+    }
+  }
+
   test("States are interpreted correctly from pod metadata.") {
-    val pods = Seq(
-      pendingExecutor(0),
-      runningExecutor(1),
-      succeededExecutor(2),
-      failedExecutorWithoutDeletion(3),
-      deletedExecutor(4),
-      unknownExecutor(5))
-    val snapshot = ExecutorPodsSnapshot(pods)
-    assert(snapshot.executorPods ===
-      Map(
-        0L -> PodPending(pods(0)),
-        1L -> PodRunning(pods(1)),
-        2L -> PodSucceeded(pods(2)),
-        3L -> PodFailed(pods(3)),
-        4L -> PodDeleted(pods(4)),
-        5L -> PodUnknown(pods(5))))
+    ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
+    val testCases = Seq(
+      testCase(pendingExecutor(0), PodPending),
+      testCase(runningExecutor(1), PodRunning),
+      testCase(succeededExecutor(2), PodSucceeded),
+      testCase(failedExecutorWithoutDeletion(3), PodFailed),
+      testCase(deletedExecutor(4), PodDeleted),
+      testCase(unknownExecutor(5), PodUnknown)
+    )
+    doTest(testCases)
+  }
+
+  test("SPARK-30821: States are interpreted correctly from pod metadata"
+    + " when configured to check all containers.") {
+    ExecutorPodsSnapshot.setShouldCheckAllContainers(true)
+    val testCases = Seq(
+      testCase(pendingExecutor(0), PodPending),
+      testCase(runningExecutor(1), PodRunning),
+      testCase(runningExecutorWithFailedContainer(2), PodFailed),
+      testCase(succeededExecutor(3), PodSucceeded),
+      testCase(failedExecutorWithoutDeletion(4), PodFailed),
+      testCase(deletedExecutor(5), PodDeleted),
+      testCase(unknownExecutor(6), PodUnknown)
+    )
+    doTest(testCases)
   }
 
   test("Updates add new pods for non-matching ids and edit existing pods for matching ids") {
+    ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
     val originalPods = Seq(
       pendingExecutor(0),
       runningExecutor(1))

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreSuite.scala

@@ -35,6 +35,7 @@ class ExecutorPodsSnapshotsStoreSuite extends SparkFunSuite with BeforeAndAfter
   before {
     eventBufferScheduler = new DeterministicScheduler()
     eventQueueUnderTest = new ExecutorPodsSnapshotsStoreImpl(eventBufferScheduler)
+    ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
   }
 
   test("Subscribers get notified of events periodically.") {