Move local file mounting to new architecture.

Author: mccheah

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

@@ -70,6 +70,7 @@ private[spark] object Constants {
 
   val ENV_SPARK_CONF_DIR = "SPARK_CONF_DIR"
   // Spark app configs for containers
+  val MOUNTED_FILES_SECRET_DIR = "/var/data/spark-submitted-files"
   val SPARK_CONF_VOLUME = "spark-conf-volume"
   val SPARK_CONF_DIR_INTERNAL = "/opt/spark/conf"
   val SPARK_CONF_FILE_NAME = "spark.properties"

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

@@ -51,6 +51,7 @@ private[spark] case class KubernetesConf[T <: KubernetesRoleSpecificConf](
     roleSpecificConf: T,
     appResourceNamePrefix: String,
     appId: String,
+    mountLocalFilesSecretName: Option[String],
     roleLabels: Map[String, String],
     roleAnnotations: Map[String, String],
     roleSecretNamesToMountPaths: Map[String, String],
@@ -137,6 +138,7 @@ private[spark] object KubernetesConf {
       KubernetesDriverSpecificConf(mainAppResource, mainClass, appName, appArgs),
       appResourceNamePrefix,
       appId,
+      Some(s"$appResourceNamePrefix-submitted-local-files"),
       driverLabels,
       driverAnnotations,
       driverSecretNamesToMountPaths,
@@ -176,6 +178,7 @@ private[spark] object KubernetesConf {
       KubernetesExecutorSpecificConf(executorId, driverPod),
       sparkConf.get(KUBERNETES_EXECUTOR_POD_NAME_PREFIX),
       appId,
+      sparkConf.get(EXECUTOR_SUBMITTED_SMALL_FILES_SECRET),
       executorLabels,
       executorAnnotations,
       executorSecrets,

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

@@ -16,6 +16,8 @@
  */
 package org.apache.spark.deploy.k8s
 
+import java.io.File
+
 import io.fabric8.kubernetes.api.model.LocalObjectReference
 
 import org.apache.spark.SparkConf
@@ -52,6 +54,17 @@ private[spark] object KubernetesUtils {
     }
   }
 
+  def submitterLocalFiles(fileUris: Iterable[String]): Iterable[String] = {
+    fileUris
+      .map(Utils.resolveURI)
+      .filter { file =>
+        Option(file.getScheme).getOrElse("file") == "file"
+      }
+      .map(_.getPath)
+      .map(new File(_))
+      .map(_.getAbsolutePath)
+  }
+
   private def resolveFileUri(uri: String): String = {
     val fileUri = Utils.resolveURI(uri)
     val fileScheme = Option(fileUri.getScheme).getOrElse("file")

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

@@ -0,0 +1,100 @@
+/*
+ * 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.k8s.features
+
+import java.io.File
+import java.nio.file.Paths
+
+import com.google.common.io.{BaseEncoding, Files}
+import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, PodBuilder, SecretBuilder}
+import scala.collection.JavaConverters._
+
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesRoleSpecificConf, SparkPod}
+import org.apache.spark.deploy.k8s.Config._
+import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.util.Utils
+
+private[spark] class MountLocalFilesFeatureStep(
+    kubernetesConf: KubernetesConf[_ <: KubernetesRoleSpecificConf])
+  extends KubernetesFeatureConfigStep {
+  require(kubernetesConf.mountLocalFilesSecretName.isDefined,
+    "Shouldn't be using this feature without a secret name.")
+  private val secretName = kubernetesConf.mountLocalFilesSecretName.get
+
+  override def configurePod(pod: SparkPod): SparkPod = {
+    val resolvedPod = new PodBuilder(pod.pod)
+      .editOrNewSpec()
+        .addNewVolume()
+          .withName("submitted-files")
+            .withNewSecret()
+            .withSecretName(secretName)
+            .endSecret()
+          .endVolume()
+      .endSpec()
+      .build()
+    val resolvedContainer = new ContainerBuilder(pod.container)
+      .addNewEnv()
+        .withName(ENV_MOUNTED_FILES_FROM_SECRET_DIR)
+        .withValue(MOUNTED_FILES_SECRET_DIR)
+        .endEnv()
+      .addNewVolumeMount()
+        .withName("submitted-files")
+        .withMountPath(MOUNTED_FILES_SECRET_DIR)
+        .endVolumeMount()
+      .build()
+    SparkPod(resolvedPod, resolvedContainer)
+  }
+
+  override def getAdditionalPodSystemProperties(): Map[String, String] = {
+    val allFiles = kubernetesConf.sparkFiles()
+    val resolvedFiles = allFiles
+      .map(file => {
+        val uri = Utils.resolveURI(file)
+        val scheme = Option(uri.getScheme).getOrElse("file")
+        if (scheme != "file") {
+          file
+        } else {
+          val fileName = Paths.get(uri.getPath).getFileName.toString
+          s"$MOUNTED_FILES_SECRET_DIR/$fileName"
+        }
+      })
+    Map(
+      EXECUTOR_SUBMITTED_SMALL_FILES_SECRET.key -> secretName,
+      "spark.files" -> resolvedFiles.mkString(","))
+  }
+
+  override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+    val localFiles = kubernetesConf.sparkFiles()
+      .map(Utils.resolveURI)
+      .filter { file =>
+        Option(file.getScheme).getOrElse("file") == "file"
+      }
+      .map(_.getPath)
+      .map(new File(_))
+    val localFileBase64Contents = localFiles.map { file =>
+      val fileBase64 = BaseEncoding.base64().encode(Files.toByteArray(file))
+      (file.getName, fileBase64)
+    }.toMap
+    val localFilesSecret = new SecretBuilder()
+      .withNewMetadata()
+        .withName(secretName)
+        .endMetadata()
+      .withData(localFileBase64Contents.asJava)
+      .build()
+    Seq(localFilesSecret)
+  }
+}

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

@@ -16,8 +16,11 @@
  */
 package org.apache.spark.deploy.k8s.submit
 
-import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpec, KubernetesDriverSpecificConf, KubernetesRoleSpecificConf}
-import org.apache.spark.deploy.k8s.features.{BasicDriverFeatureStep, DriverKubernetesCredentialsFeatureStep, DriverServiceFeatureStep, MountSecretsFeatureStep}
+import java.io.File
+
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverSpec, KubernetesDriverSpecificConf, KubernetesRoleSpecificConf, KubernetesUtils}
+import org.apache.spark.deploy.k8s.features.{BasicDriverFeatureStep, DriverKubernetesCredentialsFeatureStep, DriverServiceFeatureStep, MountLocalFilesFeatureStep, MountSecretsFeatureStep}
+import org.apache.spark.util.Utils
 
 private[spark] class KubernetesDriverBuilder(
     provideBasicStep: (KubernetesConf[KubernetesDriverSpecificConf]) => BasicDriverFeatureStep =
@@ -29,18 +32,36 @@ private[spark] class KubernetesDriverBuilder(
       new DriverServiceFeatureStep(_),
     provideSecretsStep: (KubernetesConf[_ <: KubernetesRoleSpecificConf]
       => MountSecretsFeatureStep) =
-      new MountSecretsFeatureStep(_)) {
+      new MountSecretsFeatureStep(_),
+    provideMountLocalFilesStep: (KubernetesConf[_ <: KubernetesRoleSpecificConf]
+      => MountLocalFilesFeatureStep) =
+      new MountLocalFilesFeatureStep(_)) {
+  import KubernetesDriverBuilder._
 
   def buildFromFeatures(
     kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf]): KubernetesDriverSpec = {
     val baseFeatures = Seq(
       provideBasicStep(kubernetesConf),
       provideCredentialsStep(kubernetesConf),
       provideServiceStep(kubernetesConf))
-    val allFeatures = if (kubernetesConf.roleSecretNamesToMountPaths.nonEmpty) {
+    val withProvideSecretsStep = if (kubernetesConf.roleSecretNamesToMountPaths.nonEmpty) {
       baseFeatures ++ Seq(provideSecretsStep(kubernetesConf))
     } else baseFeatures
 
+    val sparkFiles = kubernetesConf.sparkFiles()
+    val localFiles = KubernetesUtils.submitterLocalFiles(sparkFiles).map(new File(_))
+    require(localFiles.forall(_.isFile), s"All submitted local files must be present and not" +
+      s" directories, Got got: ${localFiles.map(_.getAbsolutePath).mkString(",")}")
+
+    val totalFileSize = localFiles.map(_.length()).sum
+    val totalSizeBytesString = Utils.bytesToString(totalFileSize)
+    require(totalFileSize < MAX_SECRET_BUNDLE_SIZE_BYTES,
+      s"Total size of all files submitted must be less than $MAX_SECRET_BUNDLE_SIZE_BYTES_STRING." +
+        s" Total size for files ended up being $totalSizeBytesString")
+    val allFeatures = if (localFiles.nonEmpty) {
+      withProvideSecretsStep ++ Seq(provideMountLocalFilesStep(kubernetesConf))
+    } else withProvideSecretsStep
+
     var spec = KubernetesDriverSpec.initialSpec(kubernetesConf.sparkConf.getAll.toMap)
     for (feature <- allFeatures) {
       val configuredPod = feature.configurePod(spec.pod)
@@ -54,3 +75,9 @@ private[spark] class KubernetesDriverBuilder(
     spec
   }
 }
+
+private object KubernetesDriverBuilder {
+  val MAX_SECRET_BUNDLE_SIZE_BYTES = 10240
+  val MAX_SECRET_BUNDLE_SIZE_BYTES_STRING =
+    Utils.bytesToString(MAX_SECRET_BUNDLE_SIZE_BYTES)
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/steps/DriverMountLocalFilesStep.scala

@@ -17,21 +17,27 @@
 package org.apache.spark.scheduler.cluster.k8s
 
 import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesExecutorSpecificConf, KubernetesRoleSpecificConf, SparkPod}
-import org.apache.spark.deploy.k8s.features.{BasicExecutorFeatureStep, MountSecretsFeatureStep}
+import org.apache.spark.deploy.k8s.features.{BasicExecutorFeatureStep, MountLocalFilesFeatureStep, MountSecretsFeatureStep}
 
 private[spark] class KubernetesExecutorBuilder(
     provideBasicStep: (KubernetesConf[KubernetesExecutorSpecificConf]) => BasicExecutorFeatureStep =
       new BasicExecutorFeatureStep(_),
     provideSecretsStep:
       (KubernetesConf[_ <: KubernetesRoleSpecificConf]) => MountSecretsFeatureStep =
-      new MountSecretsFeatureStep(_)) {
+      new MountSecretsFeatureStep(_),
+    provideMountLocalFilesStep:
+      (KubernetesConf[_ <: KubernetesRoleSpecificConf]) => MountLocalFilesFeatureStep =
+      new MountLocalFilesFeatureStep(_)) {
 
   def buildFromFeatures(
     kubernetesConf: KubernetesConf[KubernetesExecutorSpecificConf]): SparkPod = {
     val baseFeatures = Seq(provideBasicStep(kubernetesConf))
-    val allFeatures = if (kubernetesConf.roleSecretNamesToMountPaths.nonEmpty) {
+    val withProvideSecretsStep = if (kubernetesConf.roleSecretNamesToMountPaths.nonEmpty) {
       baseFeatures ++ Seq(provideSecretsStep(kubernetesConf))
     } else baseFeatures
+    val allFeatures = if (kubernetesConf.mountLocalFilesSecretName.isDefined) {
+      withProvideSecretsStep ++ Seq(provideMountLocalFilesStep(kubernetesConf))
+    } else withProvideSecretsStep
     var executorPod = SparkPod.initialPod()
     for (feature <- allFeatures) {
       executorPod = feature.configurePod(executorPod)

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

@@ -66,6 +66,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
         APP_ARGS),
       RESOURCE_NAME_PREFIX,
       APP_ID,
+      None,
       DRIVER_LABELS,
       DRIVER_ANNOTATIONS,
       Map.empty,
@@ -135,6 +136,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
         APP_ARGS),
       RESOURCE_NAME_PREFIX,
       APP_ID,
+      None,
       DRIVER_LABELS,
       DRIVER_ANNOTATIONS,
       Map.empty,