Add initial integration test code

integration-test/pom.xml

@@ -31,6 +31,7 @@
     <sbt.project.name>kubernetes-integration-tests</sbt.project.name>
     <spark-distro-tgz>YOUR-SPARK-DISTRO-TARBALL-HERE</spark-distro-tgz>
     <spark-dockerfiles-dir>YOUR-DOCKERFILES-DIR-HERE</spark-dockerfiles-dir>
+    <test.exclude.tags></test.exclude.tags>
   </properties>
   <packaging>jar</packaging>
   <name>Spark Project Kubernetes Integration Tests</name>
@@ -85,6 +86,12 @@
       <version>2.2.6</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+      <version>1.7.24</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -118,7 +125,7 @@
               <executable>/bin/sh</executable>
               <arguments>
                 <argument>-c</argument>
-                <argument>mkdir spark-distro-tmp; cd spark-distro-tmp; tar xfz ${spark-distro-tgz}; mv * ../spark-distro; cd ..; rm -rf spark-distro-tmp</argument>
+                <argument>rm -rf spark-distro; mkdir spark-distro-tmp; cd spark-distro-tmp; tar xfz ${spark-distro-tgz}; mv * ../spark-distro; cd ..; rm -rf spark-distro-tmp</argument>
               </arguments>
             </configuration>
           </execution>
@@ -139,7 +146,7 @@
             </configuration>
           </execution>
           <execution>
-            <!-- TODO: Remove this hack once the upstream is fixed -->
+            <!-- TODO: Remove this hack once upstream is fixed by SPARK-22777 -->
             <id>set-exec-bit-on-docker-entrypoint-sh</id>
             <phase>pre-integration-test</phase>
             <goals>
@@ -192,6 +199,19 @@
              the test phase. -->
         <groupId>org.scalatest</groupId>
         <artifactId>scalatest-maven-plugin</artifactId>
+        <version>1.0</version>
+        <configuration>
+          <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+          <junitxml>.</junitxml>
+          <filereports>SparkTestSuite.txt</filereports>
+          <argLine>-ea -Xmx3g -XX:ReservedCodeCacheSize=512m ${extraScalaTestArgs}</argLine>
+          <stderr/>
+          <systemProperties>
+            <log4j.configuration>file:src/test/resources/log4j.properties</log4j.configuration>
+            <java.awt.headless>true</java.awt.headless>
+          </systemProperties>
+          <tagsToExclude>${test.exclude.tags}</tagsToExclude>
+        </configuration>
         <executions>
           <execution>
             <id>test</id>

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala

@@ -21,7 +21,7 @@ import java.nio.file.Paths
 import java.util.UUID
 import java.util.regex.Pattern
 
-import com.google.common.io.{Files, PatternFilenameFilter}
+import com.google.common.io.PatternFilenameFilter
 import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSuite}
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Seconds, Span}
@@ -31,6 +31,7 @@ import org.apache.spark.deploy.k8s.integrationtest.constants.MINIKUBE_TEST_BACKE
 import org.apache.spark.deploy.k8s.integrationtest.constants.SPARK_DISTRO_PATH
 
 private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll with BeforeAndAfter {
+
   import KubernetesSuite._
   private val testBackend = IntegrationTestBackendFactory.getTestBackend()
   private val APP_LOCATOR_LABEL = UUID.randomUUID().toString.replaceAll("-", "")
@@ -48,8 +49,6 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
 
   before {
     sparkAppConf = kubernetesTestComponents.newSparkAppConf()
-      .set("spark.kubernetes.initcontainer.docker.image", "spark-init:latest")
-      .set("spark.kubernetes.driver.docker.image", "spark-driver:latest")
       .set("spark.kubernetes.driver.label.spark-app-locator", APP_LOCATOR_LABEL)
     kubernetesTestComponents.createNamespace()
   }

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesTestComponents.scala

@@ -19,11 +19,12 @@ package org.apache.spark.deploy.k8s.integrationtest
 import java.nio.file.Paths
 import java.util.UUID
 
-import io.fabric8.kubernetes.client.DefaultKubernetesClient
-import org.scalatest.concurrent.Eventually
 import scala.collection.mutable
 import scala.collection.JavaConverters._
 
+import io.fabric8.kubernetes.client.DefaultKubernetesClient
+import org.scalatest.concurrent.Eventually
+
 import org.apache.spark.deploy.k8s.integrationtest.constants.SPARK_DISTRO_PATH
 
 private[spark] class KubernetesTestComponents(defaultClient: DefaultKubernetesClient) {
@@ -56,6 +57,8 @@ private[spark] class KubernetesTestComponents(defaultClient: DefaultKubernetesCl
     new SparkAppConf()
       .set("spark.master", s"k8s://${kubernetesClient.getMasterUrl}")
       .set("spark.kubernetes.namespace", namespace)
+      // TODO: apache/spark#19995 is changing docker.image to container.image in these properties.
+      // Update them once the PR is merged.
       .set("spark.kubernetes.driver.docker.image",
         System.getProperty("spark.docker.test.driverImage", "spark-driver:latest"))
       .set("spark.kubernetes.executor.docker.image",

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ProcessUtils.scala

@@ -16,10 +16,10 @@
  */
 package org.apache.spark.deploy.k8s.integrationtest
 
-import java.io.{BufferedReader, InputStreamReader}
 import java.util.concurrent.TimeUnit
 
 import scala.collection.mutable.ArrayBuffer
+import scala.io.Source
 
 object ProcessUtils extends Logging {
   /**
@@ -32,17 +32,13 @@ object ProcessUtils extends Logging {
     val proc = pb.start()
     val outputLines = new ArrayBuffer[String]
 
-    Utils.tryWithResource(new InputStreamReader(proc.getInputStream)) { procOutput =>
-      Utils.tryWithResource(new BufferedReader(procOutput)) { (bufferedOutput: BufferedReader) =>
-        var line: String = null
-        do {
-          line = bufferedOutput.readLine()
-          if (line != null) {
-            logInfo(line)
-            outputLines += line
-          }
-        } while (line != null)
+    Utils.tryWithResource(Source.fromInputStream(proc.getInputStream, "UTF-8")) { output =>
+      for (line <- output.getLines) {
+        logInfo(line)
+        outputLines += line
       }
+    }{
+      output => output.close()
     }
     assert(proc.waitFor(timeout, TimeUnit.SECONDS),
       s"Timed out while executing ${fullCommand.mkString(" ")}")

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/Utils.scala

@@ -17,11 +17,49 @@
 package org.apache.spark.deploy.k8s.integrationtest
 
 import java.io.Closeable
+import java.net.URI
 
-object Utils {
+object Utils extends Logging {
 
   def tryWithResource[R <: Closeable, T](createResource: => R)(f: R => T): T = {
     val resource = createResource
     try f.apply(resource) finally resource.close()
   }
+
+  def tryWithResource[R, T](createResource: => R)(f: R => T)(closeResource: R => T): T = {
+    val resource = createResource
+    try f.apply(resource) finally closeResource(resource)
+  }
+
+  def checkAndGetK8sMasterUrl(rawMasterURL: String): String = {
+    require(rawMasterURL.startsWith("k8s://"),
+      "Kubernetes master URL must start with k8s://.")
+    val masterWithoutK8sPrefix = rawMasterURL.substring("k8s://".length)
+
+    // To handle master URLs, e.g., k8s://host:port.
+    if (!masterWithoutK8sPrefix.contains("://")) {
+      val resolvedURL = s"https://$masterWithoutK8sPrefix"
+      logInfo("No scheme specified for kubernetes master URL, so defaulting to https. Resolved " +
+        s"URL is $resolvedURL.")
+      return s"k8s://$resolvedURL"
+    }
+
+    val masterScheme = new URI(masterWithoutK8sPrefix).getScheme
+    val resolvedURL = masterScheme.toLowerCase match {
+      case "https" =>
+        masterWithoutK8sPrefix
+      case "http" =>
+        logWarning("Kubernetes master URL uses HTTP instead of HTTPS.")
+        masterWithoutK8sPrefix
+      case null =>
+        val resolvedURL = s"https://$masterWithoutK8sPrefix"
+        logInfo("No scheme specified for kubernetes master URL, so defaulting to https. Resolved " +
+          s"URL is $resolvedURL.")
+        resolvedURL
+      case _ =>
+        throw new IllegalArgumentException("Invalid Kubernetes master scheme: " + masterScheme)
+    }
+
+    s"k8s://$resolvedURL"
+  }
 }

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/GCE/GCETestBackend.scala

@@ -18,17 +18,19 @@ package org.apache.spark.deploy.k8s.integrationtest.backend.GCE
 
 import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
 
+import org.apache.spark.deploy.k8s.integrationtest.Utils
 import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackend
 import org.apache.spark.deploy.k8s.integrationtest.constants.GCE_TEST_BACKEND
 
 private[spark] class GCETestBackend(val master: String) extends IntegrationTestBackend {
   private var defaultClient: DefaultKubernetesClient = _
 
   override def initialize(): Unit = {
-    var k8ConfBuilder = new ConfigBuilder()
+    val k8sConf = new ConfigBuilder()
       .withApiVersion("v1")
-      .withMasterUrl(master.replaceFirst("k8s://", ""))
-    defaultClient = new DefaultKubernetesClient(k8ConfBuilder.build)
+      .withMasterUrl(Utils.checkAndGetK8sMasterUrl(master).replaceFirst("k8s://", ""))
+      .build()
+    defaultClient = new DefaultKubernetesClient(k8sConf)
   }
 
   override def getKubernetesClient(): DefaultKubernetesClient = {

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala

@@ -39,6 +39,9 @@ private[spark] object Minikube extends Logging {
 
   private val MINIKUBE_STARTUP_TIMEOUT_SECONDS = 60
 
+  // NOTE: This and the following methods are synchronized to prevent deleteMinikube from
+  // destroying the minikube VM while other methods try to use the VM.
+  // Such a race condition can corrupt the VM or some VM provisioning tools like VirtualBox.
   def startMinikube(): Unit = synchronized {
     assert(MINIKUBE_EXECUTABLE_DEST.exists(), EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
     if (getMinikubeStatus != MinikubeStatus.RUNNING) {

integration-test/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageBuilder.scala

@@ -42,20 +42,20 @@ private[spark] class SparkDockerImageBuilder
 
   private val originalDockerUri = URI.create(dockerHost)
   private val httpsDockerUri = new URIBuilder()
-      .setHost(originalDockerUri.getHost)
-      .setPort(originalDockerUri.getPort)
-      .setScheme("https")
-      .build()
+    .setHost(originalDockerUri.getHost)
+    .setPort(originalDockerUri.getPort)
+    .setScheme("https")
+    .build()
 
   private val dockerCerts = dockerEnv.getOrElse("DOCKER_CERT_PATH",
       throw new IllegalStateException("DOCKER_CERT_PATH env not found."))
 
   private val dockerClient = new DefaultDockerClient.Builder()
     .uri(httpsDockerUri)
-    .dockerCertificates(DockerCertificates
-        .builder()
-        .dockerCertPath(Paths.get(dockerCerts))
-        .build().get())
+    .dockerCertificates(DockerCertificates.builder()
+      .dockerCertPath(Paths.get(dockerCerts))
+      .build()
+      .get())
     .build()
 
   def buildSparkDockerImages(): Unit = {