Fix local files k8s tests

Author: Robert Kruszewski

pom.xml

@@ -107,8 +107,6 @@
     <module>external/avro</module>
     <module>dists/hadoop-palantir-bom</module>
     <module>dists/hadoop-palantir</module>
-    <module>resource-managers/kubernetes/core</module>
-    <module>resource-managers/kubernetes/integration-tests</module>
     <!-- See additional modules enabled by profiles below -->
   </modules>
 

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

@@ -34,7 +34,7 @@ private[spark] class MountLocalDriverFilesFeatureStep(
     kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf])
   extends MountLocalFilesFeatureStep(kubernetesConf) {
 
-  lazy val allFiles: Seq[String] = {
+  val allFiles: Seq[String] = {
     Utils.stringToSeq(kubernetesConf.sparkConf.get("spark.files", "")) ++
       kubernetesConf.roleSpecificConf.pyFiles ++
       (kubernetesConf.roleSpecificConf.mainAppResource match {
@@ -49,15 +49,17 @@ private[spark] class MountLocalExecutorFilesFeatureStep(
     kubernetesConf: KubernetesConf[KubernetesExecutorSpecificConf])
   extends MountLocalFilesFeatureStep(kubernetesConf) {
 
-  lazy val allFiles: Seq[String] = Nil
+  val allFiles: Seq[String] = Nil
 }
 
 private[spark] abstract 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
+  private lazy val secretName = {
+    require(kubernetesConf.mountLocalFilesSecretName.isDefined,
+      "Shouldn't be using this feature without a secret name.")
+    kubernetesConf.mountLocalFilesSecretName.get
+  }
 
   override def configurePod(pod: SparkPod): SparkPod = {
     val resolvedPod = new PodBuilder(pod.pod)

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

@@ -95,7 +95,7 @@ private[spark] class KubernetesDriverBuilder(
     val localFiles = KubernetesUtils.submitterLocalFiles(localFilesFeature.allFiles)
         .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(",")}")
+      s" directories, Got: ${localFiles.map(_.getAbsolutePath).mkString(",")}")
 
     val totalFileSize = localFiles.map(_.length()).sum
     val totalSizeBytesString = Utils.bytesToString(totalFileSize)

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

@@ -16,15 +16,19 @@
  */
 package org.apache.spark.deploy.k8s.features
 
-import scala.collection.JavaConverters._
+import java.io.File
 
+import com.google.common.base.Charsets
+import com.google.common.io.Files
 import io.fabric8.kubernetes.api.model.{Container, HasMetadata, PodBuilder, SecretBuilder}
 import org.mockito.Matchers
 import org.mockito.Mockito._
 import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
+import scala.collection.JavaConverters._
 
 import org.apache.spark.deploy.k8s.SparkPod
+import org.apache.spark.util.Utils
 
 object KubernetesFeaturesTestUtils {
 
@@ -51,6 +55,16 @@ object KubernetesFeaturesTestUtils {
     mockStep
   }
 
+  def getMockConfigStepForLocalFiles[T <: MountLocalFilesFeatureStep](
+       stepType: String, stepClass: Class[T]): T = {
+    val mockStep = getMockConfigStepForStepType(stepType, stepClass)
+    val tempDir = Utils.createTempDir()
+    val tempFile1 = new File(tempDir, "file1.txt")
+    Files.write("a", tempFile1, Charsets.UTF_8)
+    when(mockStep.allFiles()).thenReturn(Seq(tempFile1.getAbsolutePath))
+    mockStep
+  }
+
   def getSecretsForStepType[T <: KubernetesFeatureConfigStep](stepType: String)
     : Seq[HasMetadata] = {
     Seq(new SecretBuilder()

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

@@ -51,6 +51,7 @@ class MountLocalFilesFeatureStepSuite extends SparkFunSuite with BeforeAndAfter
       "https://localhost:9000/file3.txt")
     localFiles = Seq(firstLocalFile, secondLocalFile)
     val sparkConf = new SparkConf(false)
+        .set("spark.files", sparkFiles.mkString(","))
     kubernetesConf = KubernetesConf(
       sparkConf,
       KubernetesDriverSpecificConf(

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

@@ -16,10 +16,6 @@
  */
 package org.apache.spark.deploy.k8s.submit
 
-import java.io.File
-
-import com.google.common.base.Charsets
-import com.google.common.io.Files
 import io.fabric8.kubernetes.api.model.PodBuilder
 import io.fabric8.kubernetes.client.KubernetesClient
 import org.mockito.Mockito._
@@ -28,7 +24,6 @@ import org.apache.spark.{SparkConf, SparkException, SparkFunSuite}
 import org.apache.spark.deploy.k8s._
 import org.apache.spark.deploy.k8s.Config.{CONTAINER_IMAGE, KUBERNETES_DRIVER_PODTEMPLATE_FILE}
 import org.apache.spark.deploy.k8s.features._
-import org.apache.spark.util.Utils
 
 class KubernetesDriverBuilderSuite extends SparkFunSuite {
 
@@ -62,7 +57,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
   private val driverCommandStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType(
     DRIVER_CMD_STEP_TYPE, classOf[DriverCommandFeatureStep])
 
-  private val localFilesStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType(
+  private val localFilesStep = KubernetesFeaturesTestUtils.getMockConfigStepForLocalFiles(
     LOCAL_FILES_STEP_TYPE, classOf[MountLocalDriverFilesFeatureStep])
 
   private val envSecretsStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType(
@@ -116,6 +111,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
+      LOCAL_FILES_STEP_TYPE,
       DRIVER_CMD_STEP_TYPE)
   }
 
@@ -143,25 +139,15 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
+      LOCAL_FILES_STEP_TYPE,
       SECRETS_STEP_TYPE,
       ENV_SECRETS_STEP_TYPE,
       DRIVER_CMD_STEP_TYPE)
   }
 
-  test("Apply mounting small local files when present..") {
-    val tempDir = Utils.createTempDir()
-    val tempFile1 = new File(tempDir, "file1.txt")
-    Files.write("a", tempFile1, Charsets.UTF_8)
-    val tempFile2 = new File(tempDir, "file2.txt")
-    Files.write("b", tempFile2, Charsets.UTF_8)
-    val allFiles = Seq(
-      tempFile1.getAbsolutePath,
-      s"file://${tempFile2.getAbsolutePath}",
-      "https://localhost:9000/file3.txt")
-    val sparkConf = new SparkConf(false)
-      .set("spark.files", allFiles.mkString(","))
+  test("Apply mounting small local files when present.") {
     val conf = KubernetesConf(
-      sparkConf,
+      new SparkConf(false),
       KubernetesDriverSpecificConf(
         JavaMainAppResource(None),
         "test-app",
@@ -216,6 +202,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
+      LOCAL_FILES_STEP_TYPE,
       MOUNT_VOLUMES_STEP_TYPE,
       DRIVER_CMD_STEP_TYPE)
   }
@@ -244,8 +231,8 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
-      DRIVER_CMD_STEP_TYPE,
-      TEMPLATE_VOLUME_STEP_TYPE)
+      LOCAL_FILES_STEP_TYPE,
+      DRIVER_CMD_STEP_TYPE)
   }
 
   test("Apply HadoopSteps if HADOOP_CONF_DIR is defined.") {
@@ -275,6 +262,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
+      LOCAL_FILES_STEP_TYPE,
       DRIVER_CMD_STEP_TYPE,
       HADOOP_GLOBAL_STEP_TYPE)
   }
@@ -306,6 +294,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       CREDENTIALS_STEP_TYPE,
       SERVICE_STEP_TYPE,
       LOCAL_DIRS_STEP_TYPE,
+      LOCAL_FILES_STEP_TYPE,
       DRIVER_CMD_STEP_TYPE,
       HADOOP_GLOBAL_STEP_TYPE)
   }