Included integration tests for Stage 1

Author: ifilonenko

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

@@ -86,15 +86,17 @@ package object constants {
   private[spark] val INIT_CONTAINER_PROPERTIES_FILE_VOLUME = "spark-init-properties"
   private[spark] val INIT_CONTAINER_PROPERTIES_FILE_DIR = "/etc/spark-init"
   private[spark] val INIT_CONTAINER_PROPERTIES_FILE_NAME = "spark-init.properties"
+  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"
+  private[spark] val INIT_CONTAINER_SECRET_VOLUME_NAME = "spark-init-secret"
+
   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_CONF_DIR_LOC = "spark.kubernetes.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"
-  private[spark] val INIT_CONTAINER_SECRET_VOLUME_NAME = "spark-init-secret"
 
   // Miscellaneous
   private[spark] val ANNOTATION_EXECUTOR_NODE_AFFINITY = "scheduler.alpha.kubernetes.io/affinity"

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

@@ -149,7 +149,9 @@ private[spark] class Client(
 }
 
 private[spark] object Client {
-  def run(sparkConf: SparkConf, clientArguments: ClientArguments): Unit = {
+  def run(sparkConf: SparkConf,
+          clientArguments: ClientArguments,
+          hadoopConfDir: Option[String]): Unit = {
     val namespace = sparkConf.get(KUBERNETES_NAMESPACE)
     val kubernetesAppId = s"spark-${UUID.randomUUID().toString.replaceAll("-", "")}"
     val launchTime = System.currentTimeMillis()
@@ -168,6 +170,7 @@ private[spark] object Client {
         clientArguments.mainClass,
         clientArguments.driverArgs,
         clientArguments.otherPyFiles,
+        hadoopConfDir,
         sparkConf)
     Utils.tryWithResource(SparkKubernetesClientFactory.createKubernetesClient(
         master,
@@ -195,6 +198,7 @@ private[spark] object Client {
   def main(args: Array[String]): Unit = {
     val parsedArguments = ClientArguments.fromCommandLineArgs(args)
     val sparkConf = new SparkConf()
-    run(sparkConf, parsedArguments)
+    val hadoopConfDir = sys.env.get("HADOOP_CONF_DIR")
+    run(sparkConf, parsedArguments, hadoopConfDir)
   }
 }

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

@@ -40,6 +40,7 @@ private[spark] class DriverConfigurationStepsOrchestrator(
     mainClass: String,
     appArgs: Array[String],
     additionalPythonFiles: Seq[String],
+    hadoopConfDir: Option[String],
     submissionSparkConf: SparkConf) {
 
   // The resource name prefix is derived from the application name, making it easy to connect the
@@ -98,9 +99,8 @@ private[spark] class DriverConfigurationStepsOrchestrator(
         submissionSparkConf)
     val kubernetesCredentialsStep = new DriverKubernetesCredentialsStep(
         submissionSparkConf, kubernetesResourceNamePrefix)
-    val hadoopConfigurations =
-      sys.env.get("HADOOP_CONF_DIR").map{ conf => getHadoopConfFiles(conf)}
-          .getOrElse(Array.empty[File])
+    val hadoopConfigurations = hadoopConfDir.map(conf => getHadoopConfFiles(conf))
+      .getOrElse(Array.empty[File])
     val hadoopConfigSteps =
       if (hadoopConfigurations.isEmpty) {
         Option.empty[DriverConfigurationStep]
@@ -109,7 +109,8 @@ private[spark] class DriverConfigurationStepsOrchestrator(
           namespace,
           hadoopConfigMapName,
           submissionSparkConf,
-          hadoopConfigurations)
+          hadoopConfigurations,
+          hadoopConfDir)
         val hadoopConfSteps =
           hadoopStepsOrchestrator.getHadoopSteps()
         Some(new HadoopConfigBootstrapStep(hadoopConfSteps, hadoopConfigMapName))

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

@@ -37,7 +37,8 @@ private[spark] class HadoopConfigBootstrapStep(
     var currentHadoopSpec = HadoopConfigSpec(
       driverPod = driverSpec.driverPod,
       driverContainer = driverSpec.driverContainer,
-      configMapProperties = Map.empty[String, String])
+      configMapProperties = Map.empty[String, String],
+      additionalDriverSparkConf = Map.empty[String, String])
     for (nextStep <- hadoopConfigurationSteps) {
       currentHadoopSpec = nextStep.configureContainers(currentHadoopSpec)
     }
@@ -50,6 +51,7 @@ private[spark] class HadoopConfigBootstrapStep(
       .build()
     val executorSparkConf = driverSpec.driverSparkConf.clone()
       .set(HADOOP_CONFIG_MAP_SPARK_CONF_NAME, hadoopConfigMapName)
+      .setAll(currentHadoopSpec.additionalDriverSparkConf)
     driverSpec.copy(
       driverPod = currentHadoopSpec.driverPod,
       driverContainer = currentHadoopSpec.driverContainer,

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

@@ -17,17 +17,20 @@
 package org.apache.spark.deploy.kubernetes.submit.submitsteps.hadoopsteps
 
 import java.io.File
+
 import org.apache.commons.io.FileUtils.readFileToString
 
 import org.apache.spark.deploy.kubernetes.{HadoopConfBootstrap, PodWithMainContainer}
+import org.apache.spark.deploy.kubernetes.constants._
 
  /**
   * Step that configures the ConfigMap + Volumes for the driver
   */
 private[spark] class HadoopConfMounterStep(
     hadoopConfigMapName: String,
     hadoopConfigurationFiles: Array[File],
-    hadoopConfBootstrapConf: HadoopConfBootstrap)
+    hadoopConfBootstrapConf: HadoopConfBootstrap,
+    hadoopConfDir: Option[String])
   extends HadoopConfigurationStep {
 
    override def configureContainers(hadoopConfigSpec: HadoopConfigSpec): HadoopConfigSpec = {
@@ -42,7 +45,10 @@ private[spark] class HadoopConfMounterStep(
        driverContainer = bootstrappedPodAndMainContainer.mainContainer,
        configMapProperties =
          hadoopConfigurationFiles.map(file =>
-           (file.toPath.getFileName.toString, readFileToString(file))).toMap
+           (file.toPath.getFileName.toString, readFileToString(file))).toMap,
+       additionalDriverSparkConf = hadoopConfigSpec.additionalDriverSparkConf ++
+        hadoopConfDir.map(conf_dir => Map(HADOOP_CONF_DIR_LOC -> conf_dir)).getOrElse(
+          Map.empty[String, String])
      )
   }
 }

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

@@ -30,7 +30,7 @@ import io.fabric8.kubernetes.api.model.{Container, HasMetadata, Pod}
   *   pairs of (path, data)
   */
 private[spark] case class HadoopConfigSpec(
-//  additionalDriverSparkConf: Map[String, String],
+  additionalDriverSparkConf: Map[String, String],
   driverPod: Pod,
   driverContainer: Container,
   configMapProperties: Map[String, String])

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

@@ -30,7 +30,8 @@ private[spark] class HadoopStepsOrchestrator(
   namespace: String,
   hadoopConfigMapName: String,
   submissionSparkConf: SparkConf,
-  hadoopConfigurationFiles: Array[File]) {
+  hadoopConfigurationFiles: Array[File],
+  hadoopConfDir: Option[String]) {
   private val maybeKerberosSupport = submissionSparkConf.get(KUBERNETES_KERBEROS_SUPPORT)
 
   def getHadoopSteps(): Seq[HadoopConfigurationStep] = {
@@ -40,7 +41,8 @@ private[spark] class HadoopStepsOrchestrator(
     val hadoopConfMounterStep = new HadoopConfMounterStep(
       hadoopConfigMapName,
       hadoopConfigurationFiles,
-      hadoopConfBootstrapImpl)
+      hadoopConfBootstrapImpl,
+      hadoopConfDir)
     val maybeHadoopKerberosMountingStep =
       if (maybeKerberosSupport) {
         // TODO: Implement mounting secrets

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

@@ -42,6 +42,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     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 maybeHadoopConfDir = sparkConf.getOption(HADOOP_CONF_DIR_LOC)
 
     val maybeExecutorInitContainerSecretName =
       sparkConf.get(EXECUTOR_INIT_CONTAINER_SECRET)
@@ -75,9 +76,8 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val hadoopBootStrap = for {
       hadoopConfigMap <- maybeHadoopConfigMap
     } yield {
-      val hadoopConfigurations =
-        sys.env.get("HADOOP_CONF_DIR").map{ conf => getHadoopConfFiles(conf)}
-          .getOrElse(Array.empty[File])
+      val hadoopConfigurations = maybeHadoopConfDir.map(
+          conf_dir => getHadoopConfFiles(conf_dir)).getOrElse(Array.empty[File])
       new HadoopConfBootstrapImpl(
         hadoopConfigMap,
         hadoopConfigurations

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/DriverConfigurationStepsOrchestratorSuite.scala

@@ -42,6 +42,7 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         MAIN_CLASS,
         APP_ARGS,
         ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     val steps = orchestrator.getAllConfigurationSteps()
     assert(steps.size === 3)
@@ -63,6 +64,7 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         MAIN_CLASS,
         APP_ARGS,
         ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     val steps = orchestrator.getAllConfigurationSteps()
     assert(steps.size === 4)
@@ -84,6 +86,7 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         MAIN_CLASS,
         APP_ARGS,
         ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     val steps = orchestrator.getAllConfigurationSteps()
     assert(steps.size === 4)

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

@@ -74,31 +74,19 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
 
   test("Include HADOOP_CONF for HDFS based jobs ") {
     assume(testBackend.name == MINIKUBE_TEST_BACKEND)
-    // Ensuring that HADOOP_CONF_DIR env variable is set
-    val builder = new ProcessBuilder(
-      Seq("/bin/bash", "-c", "export HADOOP_CONF_DIR=" +
-        "test-data/hadoop-conf-files && exec").asJava)
-    builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
-    val process = builder.start()
-    new RedirectThread(process.getInputStream, System.out, "redirect output").start()
-    val exitCode = process.waitFor()
-    if (exitCode != 0) {
-      logInfo(s"exitCode: $exitCode")
-    }
+    // Ensuring that HADOOP_CONF_DIR variable is set, could also be one via env HADOOP_CONF_DIR
     sparkConf.setJars(Seq(CONTAINER_LOCAL_HELPER_JAR_PATH))
-    runSparkPiAndVerifyCompletion(CONTAINER_LOCAL_MAIN_APP_RESOURCE)
+    runSparkApplicationAndVerifyCompletion(
+      JavaMainAppResource(CONTAINER_LOCAL_MAIN_APP_RESOURCE),
+      SPARK_PI_MAIN_CLASS,
+      Seq("HADOOP_CONF_DIR defined. Mounting HDFS specific .xml files", "Pi is roughly 3"),
+      Array("5"),
+      Seq.empty[String],
+      Some("test-data/hadoop-conf-files"))
   }
 
   test("Run PySpark Job on file from SUBMITTER with --py-files") {
     assume(testBackend.name == MINIKUBE_TEST_BACKEND)
-    // Ensuring that HADOOP_CONF_DIR env variable is unset
-    val builder = new ProcessBuilder(
-      Seq("/bin/bash", "-c", "export HADOOP_CONF_DIR=" +
-        " && exec").asJava)
-    builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
-    val process = builder.start()
-    new RedirectThread(process.getInputStream, System.out, "redirect output").start()
-    val exitCode = process.waitFor()
     launchStagingServer(SSLOptions(), None)
     sparkConf
       .set(DRIVER_DOCKER_IMAGE,
@@ -183,7 +171,8 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
         GROUP_BY_MAIN_CLASS,
         Seq("The Result is"),
         Array.empty[String],
-        Seq.empty[String])
+        Seq.empty[String],
+        None)
   }
 
   test("Use remote resources without the resource staging server.") {
@@ -247,7 +236,8 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
         FILE_EXISTENCE_MAIN_CLASS,
         Seq(s"File found at /opt/spark/${testExistenceFile.getName} with correct contents."),
         Array(testExistenceFile.getName, TEST_EXISTENCE_FILE_CONTENTS),
-        Seq.empty[String])
+        Seq.empty[String],
+        None)
   }
 
   test("Use a very long application name.") {
@@ -277,9 +267,12 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
     runSparkApplicationAndVerifyCompletion(
         JavaMainAppResource(appResource),
         SPARK_PI_MAIN_CLASS,
-        Seq("Pi is roughly 3"),
+        Seq(
+          "hadoop config map key was not specified",
+          "Pi is roughly 3"),
         Array.empty[String],
-        Seq.empty[String])
+        Seq.empty[String],
+        None)
   }
 
   private def runPySparkPiAndVerifyCompletion(
@@ -289,21 +282,23 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
       PYSPARK_PI_MAIN_CLASS,
       Seq("Submitting 5 missing tasks from ResultStage", "Pi is roughly 3"),
       Array("5"),
-      otherPyFiles)
+      otherPyFiles,
+      None)
   }
 
   private def runSparkApplicationAndVerifyCompletion(
       appResource: MainAppResource,
       mainClass: String,
       expectedLogOnCompletion: Seq[String],
       appArgs: Array[String],
-      otherPyFiles: Seq[String]): Unit = {
+      otherPyFiles: Seq[String],
+      hadoopConfDir: Option[String]): Unit = {
     val clientArguments = ClientArguments(
       mainAppResource = appResource,
       mainClass = mainClass,
       driverArgs = appArgs,
       otherPyFiles = otherPyFiles)
-    Client.run(sparkConf, clientArguments)
+    Client.run(sparkConf, clientArguments, hadoopConfDir)
     val driverPod = kubernetesTestComponents.kubernetesClient
       .pods()
       .withLabel("spark-app-locator", APP_LOCATOR_LABEL)