[WIP] Submission client redesign to use a step-based builder pattern

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -621,14 +621,22 @@ object SparkSubmit {
     if (isKubernetesCluster) {
       childMainClass = "org.apache.spark.deploy.kubernetes.submit.Client"
       if (args.isPython) {
+        childArgs += "--py-file"
         childArgs += args.primaryResource
+        childArgs += "--main-class"
         childArgs += "org.apache.spark.deploy.PythonRunner"
+        childArgs += "--other-py-files"
         childArgs += args.pyFiles
       } else {
+        childArgs += "--primary-java-resource"
         childArgs += args.primaryResource
+        childArgs += "--main-class"
         childArgs += args.mainClass
       }
-      childArgs ++= args.childArgs
+      args.childArgs.foreach { arg =>
+        childArgs += "--arg"
+        childArgs += arg
+      }
     }
 
     // Load any properties specified through --conf and the default properties file

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

@@ -16,7 +16,7 @@
  */
 package org.apache.spark.deploy.kubernetes
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, Secret}
+import io.fabric8.kubernetes.api.model.{Container, ContainerBuilder, Pod, PodBuilder, Secret}
 
 import org.apache.spark.deploy.kubernetes.constants._
 
@@ -27,13 +27,13 @@ private[spark] trait InitContainerResourceStagingServerSecretPlugin {
    * from a resource staging server.
    */
   def mountResourceStagingServerSecretIntoInitContainer(
-      initContainer: ContainerBuilder): ContainerBuilder
+      initContainer: Container): Container
 
   /**
    * Configure the pod to attach a Secret volume which hosts secret files allowing the
    * init-container to retrieve dependencies from the resource staging server.
    */
-  def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder
+  def addResourceStagingServerSecretVolumeToPod(basePod: Pod): Pod
 }
 
 private[spark] class InitContainerResourceStagingServerSecretPluginImpl(
@@ -42,21 +42,25 @@ private[spark] class InitContainerResourceStagingServerSecretPluginImpl(
     extends InitContainerResourceStagingServerSecretPlugin {
 
   override def mountResourceStagingServerSecretIntoInitContainer(
-      initContainer: ContainerBuilder): ContainerBuilder = {
-    initContainer.addNewVolumeMount()
-      .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
-      .withMountPath(initContainerSecretMountPath)
-      .endVolumeMount()
+      initContainer: Container): Container = {
+    new ContainerBuilder(initContainer)
+      .addNewVolumeMount()
+        .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
+        .withMountPath(initContainerSecretMountPath)
+        .endVolumeMount()
+      .build()
   }
 
-  override def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder = {
-    basePod.editSpec()
-      .addNewVolume()
-        .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
-        .withNewSecret()
-          .withSecretName(initContainerSecretName)
-          .endSecret()
-        .endVolume()
-      .endSpec()
+  override def addResourceStagingServerSecretVolumeToPod(basePod: Pod): Pod = {
+    new PodBuilder(basePod)
+      .editSpec()
+        .addNewVolume()
+          .withName(INIT_CONTAINER_SECRET_VOLUME_NAME)
+          .withNewSecret()
+            .withSecretName(initContainerSecretName)
+            .endSecret()
+          .endVolume()
+        .endSpec()
+      .build()
   }
 }

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

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.deploy.kubernetes
+
+import io.fabric8.kubernetes.api.model.{Container, Pod}
+
+private[spark] case class PodWithDetachedInitContainer(
+    pod: Pod,
+    initContainer: Container,
+    mainContainer: Container)

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

@@ -19,19 +19,25 @@ package org.apache.spark.deploy.kubernetes
 import io.fabric8.kubernetes.api.model.{ContainerBuilder, EmptyDirVolumeSource, PodBuilder, VolumeMount, VolumeMountBuilder}
 
 import org.apache.spark.deploy.kubernetes.constants._
-import org.apache.spark.deploy.kubernetes.submit.{ContainerNameEqualityPredicate, InitContainerUtil}
 
+/**
+ * This is separated out from the init-container steps API because this component can be reused to
+ * set up the init-container for executors as well.
+ */
 private[spark] trait SparkPodInitContainerBootstrap {
   /**
    * Bootstraps an init-container that downloads dependencies to be used by a main container.
    * Note that this primarily assumes that the init-container's configuration is being provided
    * by a ConfigMap that was installed by some other component; that is, the implementation
    * here makes no assumptions about how the init-container is specifically configured. For
    * example, this class is unaware if the init-container is fetching remote dependencies or if
-   * it is fetching dependencies from a resource staging server.
+   * it is fetching dependencies from a resource staging server. Additionally, the container itself
+   * is not actually attached to the pod, but the init container is returned so it can be attached
+   * by InitContainerUtil after the caller has decided to make any changes to it.
    */
   def bootstrapInitContainerAndVolumes(
-      mainContainerName: String, originalPodSpec: PodBuilder): PodBuilder
+      originalPodWithUnattachedInitContainer: PodWithDetachedInitContainer)
+      : PodWithDetachedInitContainer
 }
 
 private[spark] class SparkPodInitContainerBootstrapImpl(
@@ -41,13 +47,12 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
     filesDownloadPath: String,
     downloadTimeoutMinutes: Long,
     initContainerConfigMapName: String,
-    initContainerConfigMapKey: String,
-    resourceStagingServerSecretPlugin: Option[InitContainerResourceStagingServerSecretPlugin])
+    initContainerConfigMapKey: String)
   extends SparkPodInitContainerBootstrap {
 
   override def bootstrapInitContainerAndVolumes(
-      mainContainerName: String,
-      originalPodSpec: PodBuilder): PodBuilder = {
+      originalPodWithUnattachedInitContainer: PodWithDetachedInitContainer)
+      : PodWithDetachedInitContainer = {
     val sharedVolumeMounts = Seq[VolumeMount](
       new VolumeMountBuilder()
         .withName(INIT_CONTAINER_DOWNLOAD_JARS_VOLUME_NAME)
@@ -58,7 +63,7 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
         .withMountPath(filesDownloadPath)
         .build())
 
-    val initContainer = new ContainerBuilder()
+    val initContainer = new ContainerBuilder(originalPodWithUnattachedInitContainer.initContainer)
       .withName(s"spark-init")
       .withImage(initContainerImage)
       .withImagePullPolicy(dockerImagePullPolicy)
@@ -68,11 +73,8 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
         .endVolumeMount()
       .addToVolumeMounts(sharedVolumeMounts: _*)
       .addToArgs(INIT_CONTAINER_PROPERTIES_FILE_PATH)
-    val resolvedInitContainer = resourceStagingServerSecretPlugin.map { plugin =>
-      plugin.mountResourceStagingServerSecretIntoInitContainer(initContainer)
-    }.getOrElse(initContainer).build()
-    val podWithBasicVolumes = InitContainerUtil.appendInitContainer(
-        originalPodSpec, resolvedInitContainer)
+      .build()
+    val podWithBasicVolumes = new PodBuilder(originalPodWithUnattachedInitContainer.pod)
       .editSpec()
         .addNewVolume()
           .withName(INIT_CONTAINER_PROPERTIES_FILE_VOLUME)
@@ -92,17 +94,20 @@ private[spark] class SparkPodInitContainerBootstrapImpl(
           .withName(INIT_CONTAINER_DOWNLOAD_FILES_VOLUME_NAME)
           .withEmptyDir(new EmptyDirVolumeSource())
           .endVolume()
-        .editMatchingContainer(new ContainerNameEqualityPredicate(mainContainerName))
-          .addToVolumeMounts(sharedVolumeMounts: _*)
-          .addNewEnv()
-            .withName(ENV_MOUNTED_FILES_DIR)
-            .withValue(filesDownloadPath)
-            .endEnv()
-          .endContainer()
         .endSpec()
-    resourceStagingServerSecretPlugin.map { plugin =>
-      plugin.addResourceStagingServerSecretVolumeToPod(podWithBasicVolumes)
-    }.getOrElse(podWithBasicVolumes)
+      .build()
+    val mainContainerWithMountedFiles = new ContainerBuilder(
+      originalPodWithUnattachedInitContainer.mainContainer)
+        .addToVolumeMounts(sharedVolumeMounts: _*)
+        .addNewEnv()
+          .withName(ENV_MOUNTED_FILES_DIR)
+          .withValue(filesDownloadPath)
+          .endEnv()
+        .build()
+    PodWithDetachedInitContainer(
+      podWithBasicVolumes,
+      initContainer,
+      mainContainerWithMountedFiles)
   }
 
 }

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

@@ -418,6 +418,14 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
+  private[spark] val INIT_CONTAINER_REMOTE_PYSPARK_FILES =
+    ConfigBuilder("spark.kubernetes.initcontainer.remotePyFiles")
+      .doc("Comma-separated list of Python file URIs to download in the init-container. This is" +
+        " calculated given the list of python files sent to spark-submit.")
+      .internal()
+      .stringConf
+      .createOptional
+
   private[spark] val INIT_CONTAINER_DOCKER_IMAGE =
     ConfigBuilder("spark.kubernetes.initcontainer.docker.image")
       .doc("Image for the driver and executor's init-container that downloads dependencies.")