[SPARK-29865][K8S] Ensure client-mode executors have same name prefix

This basically does what BasicDriverFeatureStep already does to achieve the
same thing in cluster mode; but since that class (or any other feature) is
not invoked in client mode, it needs to be done elsewhere.

I also modified the client mode integration test to check the executor name
prefix; while there I had to fix the minikube backend to parse the output
from newer minikube versions (I have 1.5.2).

Closes #26488 from vanzin/SPARK-29865.

Authored-by: Marcelo Vanzin <vanzin@cloudera.com>
Signed-off-by: Erik Erlandson <eerlands@redhat.com>

Author: Marcelo Vanzin

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

@@ -23,7 +23,7 @@ import com.google.common.cache.CacheBuilder
 import io.fabric8.kubernetes.client.Config
 
 import org.apache.spark.SparkContext
-import org.apache.spark.deploy.k8s.{KubernetesUtils, SparkKubernetesClientFactory}
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesUtils, SparkKubernetesClientFactory}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.internal.Logging
@@ -61,6 +61,17 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
         None)
     }
 
+    // If KUBERNETES_EXECUTOR_POD_NAME_PREFIX is not set, initialize it so that all executors have
+    // the same prefix. This is needed for client mode, where the feature steps code that sets this
+    // configuration is not used.
+    //
+    // If/when feature steps are executed in client mode, they should instead take care of this,
+    // and this code should be removed.
+    if (!sc.conf.contains(KUBERNETES_EXECUTOR_POD_NAME_PREFIX)) {
+      sc.conf.set(KUBERNETES_EXECUTOR_POD_NAME_PREFIX,
+        KubernetesConf.getResourceNamePrefix(sc.conf.get("spark.app.name")))
+    }
+
     val kubernetesClient = SparkKubernetesClientFactory.createKubernetesClient(
       apiServerUri,
       Some(sc.conf.get(KUBERNETES_NAMESPACE)),

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ClientModeTestsSuite.scala

@@ -27,6 +27,7 @@ private[spark] trait ClientModeTestsSuite { k8sSuite: KubernetesSuite =>
     val labels = Map("spark-app-selector" -> driverPodName)
     val driverPort = 7077
     val blockManagerPort = 10000
+    val executorLabel = "spark-client-it"
     val driverService = testBackend
       .getKubernetesClient
       .services()
@@ -78,10 +79,11 @@ private[spark] trait ClientModeTestsSuite { k8sSuite: KubernetesSuite =>
               "/var/run/secrets/kubernetes.io/serviceaccount/token")
             .addToArgs("--conf", "spark.kubernetes.authenticate.caCertFile=" +
               "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
-            .addToArgs("--conf", s"spark.kubernetes.driver.pod.name=$driverPodName")
             .addToArgs("--conf", "spark.executor.memory=500m")
             .addToArgs("--conf", "spark.executor.cores=1")
-            .addToArgs("--conf", "spark.executor.instances=1")
+            .addToArgs("--conf", "spark.executor.instances=2")
+            .addToArgs("--conf", "spark.kubernetes.executor.deleteOnTermination=false")
+            .addToArgs("--conf", s"spark.kubernetes.executor.label.$executorLabel=$executorLabel")
             .addToArgs("--conf",
               s"spark.driver.host=" +
                 s"${driverService.getMetadata.getName}.${kubernetesTestComponents.namespace}.svc")
@@ -99,13 +101,34 @@ private[spark] trait ClientModeTestsSuite { k8sSuite: KubernetesSuite =>
           .getLog
           .contains("Pi is roughly 3"), "The application did not complete.")
       }
+
+      val executors = kubernetesTestComponents
+        .kubernetesClient
+        .pods()
+        .inNamespace(kubernetesTestComponents.namespace)
+        .withLabel(executorLabel, executorLabel)
+        .list()
+        .getItems()
+      assert(executors.size === 2)
+      val prefixes = executors.asScala.map { pod =>
+        val name = pod.getMetadata().getName()
+        name.substring(0, name.lastIndexOf("-"))
+      }.toSet
+      assert(prefixes.size === 1, s"Executor prefixes did not match: $prefixes")
     } finally {
       // Have to delete the service manually since it doesn't have an owner reference
       kubernetesTestComponents
         .kubernetesClient
         .services()
         .inNamespace(kubernetesTestComponents.namespace)
         .delete(driverService)
+      // Delete all executors, since the test explicitly asks them not to be deleted by the app.
+      kubernetesTestComponents
+        .kubernetesClient
+        .pods()
+        .inNamespace(kubernetesTestComponents.namespace)
+        .withLabel(executorLabel, executorLabel)
+        .delete()
     }
   }
 

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala

@@ -30,6 +30,7 @@ private[spark] object Minikube extends Logging {
   private val KUBELET_PREFIX = "kubelet:"
   private val APISERVER_PREFIX = "apiserver:"
   private val KUBECTL_PREFIX = "kubectl:"
+  private val KUBECONFIG_PREFIX = "kubeconfig:"
   private val MINIKUBE_VM_PREFIX = "minikubeVM: "
   private val MINIKUBE_PREFIX = "minikube: "
   private val MINIKUBE_PATH = ".minikube"
@@ -86,18 +87,23 @@ private[spark] object Minikube extends Logging {
     val kubeletString = statusString.find(_.contains(s"$KUBELET_PREFIX "))
     val apiserverString = statusString.find(_.contains(s"$APISERVER_PREFIX "))
     val kubectlString = statusString.find(_.contains(s"$KUBECTL_PREFIX "))
+    val kubeconfigString = statusString.find(_.contains(s"$KUBECONFIG_PREFIX "))
+    val hasConfigStatus = kubectlString.isDefined || kubeconfigString.isDefined
 
-    if (hostString.isEmpty || kubeletString.isEmpty
-      || apiserverString.isEmpty || kubectlString.isEmpty) {
+    if (hostString.isEmpty || kubeletString.isEmpty || apiserverString.isEmpty ||
+        !hasConfigStatus) {
       MinikubeStatus.NONE
     } else {
       val status1 = hostString.get.replaceFirst(s"$HOST_PREFIX ", "")
       val status2 = kubeletString.get.replaceFirst(s"$KUBELET_PREFIX ", "")
       val status3 = apiserverString.get.replaceFirst(s"$APISERVER_PREFIX ", "")
-      val status4 = kubectlString.get.replaceFirst(s"$KUBECTL_PREFIX ", "")
-      if (!status4.contains("Correctly Configured:")) {
-        MinikubeStatus.NONE
+      val isConfigured = if (kubectlString.isDefined) {
+        val cfgStatus = kubectlString.get.replaceFirst(s"$KUBECTL_PREFIX ", "")
+        cfgStatus.contains("Correctly Configured:")
       } else {
+        kubeconfigString.get.replaceFirst(s"$KUBECONFIG_PREFIX ", "") == "Configured"
+      }
+      if (isConfigured) {
         val stats = List(status1, status2, status3)
           .map(MinikubeStatus.unapply)
           .map(_.getOrElse(throw new IllegalStateException(s"Unknown status $statusString")))
@@ -106,6 +112,8 @@ private[spark] object Minikube extends Logging {
         } else {
           MinikubeStatus.RUNNING
         }
+      } else {
+        MinikubeStatus.NONE
       }
     }
   }