Allow driver pod name to be optional.

Author: mccheah

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

@@ -43,7 +43,7 @@ private[spark] case class KubernetesDriverSpecificConf(
  */
 private[spark] case class KubernetesExecutorSpecificConf(
     executorId: String,
-    driverPod: Pod)
+    driverPod: Option[Pod])
   extends KubernetesRoleSpecificConf
 
 /**
@@ -178,7 +178,7 @@ private[spark] object KubernetesConf {
       sparkConf: SparkConf,
       executorId: String,
       appId: String,
-      driverPod: Pod): KubernetesConf[KubernetesExecutorSpecificConf] = {
+      driverPod: Option[Pod]): KubernetesConf[KubernetesExecutorSpecificConf] = {
     val executorCustomLabels = KubernetesUtils.parsePrefixedKeyValuePairs(
       sparkConf, KUBERNETES_EXECUTOR_LABEL_PREFIX)
     require(

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

@@ -18,7 +18,7 @@ package org.apache.spark.deploy.k8s.features
 
 import scala.collection.JavaConverters._
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, ContainerPortBuilder, EnvVar, EnvVarBuilder, EnvVarSourceBuilder, HasMetadata, PodBuilder, QuantityBuilder}
+import io.fabric8.kubernetes.api.model._
 
 import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesExecutorSpecificConf, SparkPod}
@@ -152,19 +152,20 @@ private[spark] class BasicExecutorFeatureStep(
         .build()
     }.getOrElse(executorContainer)
     val driverPod = kubernetesConf.roleSpecificConf.driverPod
+    val ownerReference = driverPod.map(pod =>
+      new OwnerReferenceBuilder()
+        .withController(true)
+        .withApiVersion(pod.getApiVersion)
+        .withKind(pod.getKind)
+        .withName(pod.getMetadata.getName)
+        .withUid(pod.getMetadata.getUid)
+        .build())
     val executorPod = new PodBuilder(pod.pod)
       .editOrNewMetadata()
         .withName(name)
         .withLabels(kubernetesConf.roleLabels.asJava)
         .withAnnotations(kubernetesConf.roleAnnotations.asJava)
-        .withOwnerReferences()
-        .addNewOwnerReference()
-          .withController(true)
-          .withApiVersion(driverPod.getApiVersion)
-          .withKind(driverPod.getKind)
-          .withName(driverPod.getMetadata.getName)
-          .withUid(driverPod.getMetadata.getUid)
-          .endOwnerReference()
+        .addToOwnerReferences(ownerReference.toSeq: _*)
         .endMetadata()
       .editOrNewSpec()
         .withHostname(hostname)

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

@@ -50,14 +50,14 @@ private[spark] class ExecutorPodsAllocator(
 
   private val kubernetesDriverPodName = conf
     .get(KUBERNETES_DRIVER_POD_NAME)
-    .getOrElse(throw new SparkException("Must specify the driver pod name"))
-
-  private val driverPod = Option(kubernetesClient.pods()
-    .withName(kubernetesDriverPodName)
-    .get())
-    .getOrElse(throw new SparkException(
-      s"No pod was found named $kubernetesDriverPodName in the cluster in the" +
-        s" namespace $namespace (this was supposed to be the driver pod.)."))
+
+  private val driverPod = kubernetesDriverPodName
+    .map(name => Option(kubernetesClient.pods()
+      .withName(name)
+      .get())
+      .getOrElse(throw new SparkException(
+        s"No pod was found named $kubernetesDriverPodName in the cluster in the " +
+          s"namespace $namespace (this was supposed to be the driver pod.).")))
 
   // Executor IDs that have been requested from Kubernetes but have not been detected in any
   // snapshot yet. Mapped to the timestamp when they were created.

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

@@ -50,6 +50,12 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
       KUBERNETES_AUTH_CLIENT_MODE_PREFIX
     }
 
+    val apiServerUrl = if (wasSparkSubmitted) {
+      KUBERNETES_MASTER_INTERNAL_URL
+    } else {
+      masterURL.substring("k8s://".length())
+    }
+
     val kubernetesClient = SparkKubernetesClientFactory.createKubernetesClient(
       KUBERNETES_MASTER_INTERNAL_URL,
       Some(sc.conf.get(KUBERNETES_NAMESPACE)),

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala

@@ -81,7 +81,7 @@ class BasicExecutorFeatureStepSuite
     val step = new BasicExecutorFeatureStep(
       KubernetesConf(
         baseConf,
-        KubernetesExecutorSpecificConf("1", DRIVER_POD),
+        KubernetesExecutorSpecificConf("1", Some(DRIVER_POD)),
         RESOURCE_NAME_PREFIX,
         APP_ID,
         LABELS,
@@ -120,7 +120,7 @@ class BasicExecutorFeatureStepSuite
     val step = new BasicExecutorFeatureStep(
       KubernetesConf(
         conf,
-        KubernetesExecutorSpecificConf("1", DRIVER_POD),
+        KubernetesExecutorSpecificConf("1", Some(DRIVER_POD)),
         longPodNamePrefix,
         APP_ID,
         LABELS,
@@ -140,7 +140,7 @@ class BasicExecutorFeatureStepSuite
     val step = new BasicExecutorFeatureStep(
       KubernetesConf(
         conf,
-        KubernetesExecutorSpecificConf("1", DRIVER_POD),
+        KubernetesExecutorSpecificConf("1", Some(DRIVER_POD)),
         RESOURCE_NAME_PREFIX,
         APP_ID,
         LABELS,

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

@@ -190,12 +190,12 @@ private[spark] class KubernetesSuite extends SparkFunSuite
             .withImage(image)
             .withImagePullPolicy("IfNotPresent")
             .withCommand("/opt/spark/bin/run-example")
-            .addToArgs("--master", s"k8s://${kubernetesTestComponents.clientConfig.getMasterUrl}")
+            .addToArgs("--master", s"k8s://https://kubernetes.default.svc")
             .addToArgs("--deploy-mode", "client")
             .addToArgs(
               "--conf",
               s"spark.kubernetes.namespace=${kubernetesTestComponents.namespace}")
-            .addToArgs("--conf", "spark.kubernetes.driver.pod.name=driverPodName")
+            .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")