Added proper logic for mounting ConfigMaps

Author: ifilonenko

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

@@ -43,17 +43,17 @@ private[spark] class HadoopConfBootstrapImpl(
   override def bootstrapMainContainerAndVolumes(
     originalPodWithMainContainer: PodWithMainContainer)
     : PodWithMainContainer = {
-    import collection.JavaConverters._
-    val fileContents = hadoopConfigFiles.map(file => (file.getPath, file.toString)).toMap
+    import scala.collection.JavaConverters._
     val keyPaths = hadoopConfigFiles.map(file =>
-      new KeyToPathBuilder().withKey(file.getPath).withPath(file.getAbsolutePath).build())
+      new KeyToPathBuilder().withKey(file.toPath.getFileName.toString)
+        .withPath(file.toPath.getFileName.toString).build()).toList
     val hadoopSupportedPod = new PodBuilder(originalPodWithMainContainer.pod)
       .editSpec()
         .addNewVolume()
           .withName(HADOOP_FILE_VOLUME)
             .withNewConfigMap()
               .withName(hadoopConfConfigMapName)
-              .addAllToItems(keyPaths.toList.asJavaCollection)
+              .withItems(keyPaths.asJava)
             .endConfigMap()
           .endVolume()
         .endSpec()

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

@@ -89,6 +89,8 @@ package object constants {
   private[spark] val HADOOP_FILE_VOLUME = "hadoop-properties"
   private[spark] val HADOOP_FILE_DIR = "/etc/hadoop"
   private[spark] val HADOOP_CONF_DIR = "HADOOP_CONF_DIR"
+  private[spark] val HADOOP_CONFIG_MAP_SPARK_CONF_NAME =
+    "spark.kubernetes.hadoop.executor.hadoopconfigmapname"
   private[spark] val INIT_CONTAINER_PROPERTIES_FILE_PATH =
     s"$INIT_CONTAINER_PROPERTIES_FILE_DIR/$INIT_CONTAINER_PROPERTIES_FILE_NAME"
   private[spark] val DEFAULT_SHUFFLE_MOUNT_NAME = "shuffle"

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

@@ -112,7 +112,7 @@ private[spark] class DriverConfigurationStepsOrchestrator(
           hadoopConfigurations)
         val hadoopConfSteps =
           hadoopStepsOrchestrator.getHadoopSteps()
-        Some(new HadoopConfigBootstrapStep(hadoopConfSteps))
+        Some(new HadoopConfigBootstrapStep(hadoopConfSteps, kubernetesResourceNamePrefix))
       }
     val pythonStep = mainAppResource match {
       case PythonMainAppResource(mainPyResource) =>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/HadoopConfigBootstrapStep.scala

@@ -16,24 +16,44 @@
  */
 package org.apache.spark.deploy.kubernetes.submit.submitsteps
 
+import java.io.StringWriter
+import java.util.Properties
+
+import io.fabric8.kubernetes.api.model.{ConfigMap, ConfigMapBuilder, HasMetadata}
+import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.deploy.kubernetes.submit.submitsteps.hadoopsteps.{HadoopConfigSpec, HadoopConfigurationStep}
 
  /**
   * Configures the driverSpec that bootstraps dependencies into the driver pod.
   */
 private[spark] class HadoopConfigBootstrapStep(
-  hadoopConfigurationSteps: Seq[HadoopConfigurationStep])
+  hadoopConfigurationSteps: Seq[HadoopConfigurationStep], kubernetesResourceNamePrefix: String)
   extends DriverConfigurationStep {
+  private val hadoopConfigMapName = s"$kubernetesResourceNamePrefix-hadoop-config"
 
   override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
+    import scala.collection.JavaConverters._
     var currentHadoopSpec = HadoopConfigSpec(
       driverPod = driverSpec.driverPod,
-      driverContainer = driverSpec.driverContainer)
+      driverContainer = driverSpec.driverContainer,
+      configMapProperties = Map.empty[String, String])
     for (nextStep <- hadoopConfigurationSteps) {
       currentHadoopSpec = nextStep.configureContainers(currentHadoopSpec)
     }
+    val configMap =
+      new ConfigMapBuilder()
+        .withNewMetadata()
+        .withName(hadoopConfigMapName)
+        .endMetadata()
+          .addToData(currentHadoopSpec.configMapProperties.asJava)
+      .build()
+    val executorSparkConf = driverSpec.driverSparkConf.clone()
+      .set(HADOOP_CONFIG_MAP_SPARK_CONF_NAME, hadoopConfigMapName)
     driverSpec.copy(
       driverPod = currentHadoopSpec.driverPod,
-      driverContainer = currentHadoopSpec.driverContainer)
+      driverContainer = currentHadoopSpec.driverContainer,
+      driverSparkConf = executorSparkConf,
+      otherKubernetesResources = Seq(configMap)
+      )
   }
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopConfMounterStep.scala

@@ -16,13 +16,16 @@
  */
 package org.apache.spark.deploy.kubernetes.submit.submitsteps.hadoopsteps
 
+import java.io.File
+
 import org.apache.spark.deploy.kubernetes.{HadoopConfBootstrap, PodWithMainContainer}
 
  /**
   * Step that configures the ConfigMap + Volumes for the driver
   */
 private[spark] class HadoopConfMounterStep(
     hadoopConfigMapName: String,
+    hadoopConfigurationFiles: Array[File],
     hadoopConfBootstrapConf: HadoopConfBootstrap)
   extends HadoopConfigurationStep {
 
@@ -35,7 +38,10 @@ private[spark] class HadoopConfMounterStep(
           ))
      hadoopConfigSpec.copy(
        driverPod = bootstrappedPodAndMainContainer.pod,
-       driverContainer = bootstrappedPodAndMainContainer.mainContainer
+       driverContainer = bootstrappedPodAndMainContainer.mainContainer,
+       configMapProperties =
+         hadoopConfigurationFiles.map(file =>
+           (file.toPath.getFileName.toString, file.toString)).toMap
      )
   }
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopConfigSpec.scala

@@ -26,8 +26,11 @@ import io.fabric8.kubernetes.api.model.{Container, HasMetadata, Pod}
   * - The spec of the main container so that it can be modified to share volumes
   * - The spec of the driver pod EXCEPT for the addition of the given hadoop configs (e.g. volumes
   *   the hadoop logic needs)
+  * - The properties that will be stored into the config map which have (key, value)
+  *   pairs of (path, data)
   */
 private[spark] case class HadoopConfigSpec(
 //  additionalDriverSparkConf: Map[String, String],
   driverPod: Pod,
-  driverContainer: Container)
+  driverContainer: Container,
+  configMapProperties: Map[String, String])

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopStepsOrchestrator.scala

@@ -40,6 +40,7 @@ private[spark] class HadoopStepsOrchestrator(
       hadoopConfigurationFiles)
     val hadoopConfMounterStep = new HadoopConfMounterStep(
       hadoopConfigMapName,
+      hadoopConfigurationFiles,
       hadoopConfBootstrapImpl)
     val maybeHadoopKerberosMountingStep =
       if (maybeKerberosSupport) {

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

@@ -20,7 +20,7 @@ import java.io.File
 
 import io.fabric8.kubernetes.client.Config
 
-import org.apache.spark.deploy.kubernetes.{InitContainerResourceStagingServerSecretPluginImpl, SparkKubernetesClientFactory, SparkPodInitContainerBootstrapImpl}
+import org.apache.spark.deploy.kubernetes._
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.internal.Logging
@@ -41,6 +41,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val sparkConf = sc.getConf
     val maybeConfigMap = sparkConf.get(EXECUTOR_INIT_CONTAINER_CONFIG_MAP)
     val maybeConfigMapKey = sparkConf.get(EXECUTOR_INIT_CONTAINER_CONFIG_MAP_KEY)
+    val maybeHadoopConfigMap = sparkConf.getOption(HADOOP_CONFIG_MAP_SPARK_CONF_NAME)
 
     val maybeExecutorInitContainerSecretName =
       sparkConf.get(EXECUTOR_INIT_CONTAINER_SECRET)
@@ -71,6 +72,17 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
         configMap,
         configMapKey)
     }
+    val hadoopBootStrap = for {
+      hadoopConfigMap <- maybeHadoopConfigMap
+    } yield {
+      val hadoopConfigurations =
+        sys.env.get("HADOOP_CONF_DIR").map{ conf => getHadoopConfFiles(conf)}
+          .getOrElse(Array.empty[File])
+      new HadoopConfBootstrapImpl(
+        hadoopConfigMap,
+        hadoopConfigurations
+      )
+    }
     if (maybeConfigMap.isEmpty) {
       logWarning("The executor's init-container config map was not specified. Executors will" +
         " therefore not attempt to fetch remote or submitted dependencies.")
@@ -79,6 +91,10 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
       logWarning("The executor's init-container config map key was not specified. Executors will" +
         " therefore not attempt to fetch remote or submitted dependencies.")
     }
+    if (maybeHadoopConfigMap.isEmpty) {
+      logWarning("The executor's hadoop config map key was not specified. Executors will" +
+        " therefore not attempt to fetch hadoop configuration files.")
+    }
     val kubernetesClient = SparkKubernetesClientFactory.createKubernetesClient(
         KUBERNETES_MASTER_INTERNAL_URL,
         Some(sparkConf.get(KUBERNETES_NAMESPACE)),
@@ -90,11 +106,21 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
         sc.taskScheduler.asInstanceOf[TaskSchedulerImpl],
         sc,
         initBootStrap,
+        hadoopBootStrap,
         executorInitContainerSecretVolumePlugin,
         kubernetesClient)
   }
 
   override def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit = {
     scheduler.asInstanceOf[TaskSchedulerImpl].initialize(backend)
   }
+  private def getHadoopConfFiles(path: String) : Array[File] = {
+    def isFile(file: File) = if (file.isFile) Some(file) else None
+    val dir = new File(path)
+    if (dir.isDirectory) {
+      dir.listFiles.flatMap { file => isFile(file) }
+    } else {
+      Array.empty[File]
+    }
+  }
 }

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

@@ -21,18 +21,18 @@ import java.net.InetAddress
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.{AtomicInteger, AtomicLong, AtomicReference}
 
+import scala.collection.mutable
+import scala.concurrent.{ExecutionContext, Future}
+
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, ContainerPortBuilder, EnvVarBuilder, EnvVarSourceBuilder, Pod, PodBuilder, QuantityBuilder}
+import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientException, Watcher}
 import io.fabric8.kubernetes.client.Watcher.Action
 import org.apache.commons.io.FilenameUtils
-import scala.collection.JavaConverters._
-import scala.collection.mutable
-import scala.concurrent.{ExecutionContext, Future}
 
 import org.apache.spark.{SparkContext, SparkEnv, SparkException}
-import org.apache.spark.deploy.kubernetes.{ConfigurationUtils, InitContainerResourceStagingServerSecretPlugin, PodWithDetachedInitContainer, SparkPodInitContainerBootstrap}
+import org.apache.spark.deploy.kubernetes._
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.deploy.kubernetes.submit.InitContainerUtil
@@ -48,6 +48,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
     scheduler: TaskSchedulerImpl,
     val sc: SparkContext,
     executorInitContainerBootstrap: Option[SparkPodInitContainerBootstrap],
+    executorHadoopBootStrap: Option[HadoopConfBootstrap],
     executorMountInitContainerSecretPlugin: Option[InitContainerResourceStagingServerSecretPlugin],
     kubernetesClient: KubernetesClient)
   extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
@@ -366,6 +367,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
    * @return A tuple of the new executor name and the Pod data structure.
    */
   private def allocateNewExecutorPod(nodeToLocalTaskCount: Map[String, Int]): (String, Pod) = {
+    import scala.collection.JavaConverters._
     val executorId = EXECUTOR_ID_COUNTER.incrementAndGet().toString
     val name = s"$executorPodNamePrefix-exec-$executorId"
 
@@ -520,9 +522,16 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
     val executorPodWithNodeAffinity = addNodeAffinityAnnotationIfUseful(
         executorPodWithInitContainer, nodeToLocalTaskCount)
-    val resolvedExecutorPod = new PodBuilder(executorPodWithNodeAffinity)
+    val (executorHadoopConfPod, executorHadoopConfContainer) =
+      executorHadoopBootStrap.map { bootstrap =>
+        val podWithMainContainer = bootstrap.bootstrapMainContainerAndVolumes(
+          PodWithMainContainer(executorPodWithNodeAffinity, initBootstrappedExecutorContainer)
+        )
+        (podWithMainContainer.pod, podWithMainContainer.mainContainer)
+      }.getOrElse(executorPodWithNodeAffinity, initBootstrappedExecutorContainer)
+    val resolvedExecutorPod = new PodBuilder(executorHadoopConfPod)
       .editSpec()
-        .addToContainers(initBootstrappedExecutorContainer)
+        .addToContainers(executorHadoopConfContainer)
         .endSpec()
       .build()
     try {

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

@@ -72,6 +72,12 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
     kubernetesTestComponents.deleteNamespace()
   }
 
+  test("Include HADOOP_CONF for HDFS based jobs ") {
+    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+
+    runSparkPiAndVerifyCompletion(CONTAINER_LOCAL_MAIN_APP_RESOURCE)
+  }
+
   test("Run PySpark Job on file from SUBMITTER with --py-files") {
     assume(testBackend.name == MINIKUBE_TEST_BACKEND)