initial logic needed for porting PR 521

Author: ifilonenko

README.md

@@ -10,6 +10,10 @@ is subject to change.
 
 Note that currently the integration tests only run with Java 8.
 
+Integration tests firstly require installing [Minikube](https://kubernetes.io/docs/getting-started-guides/minikube/) on
+your machine, and for the `Minikube` binary to be on your `PATH`.. Refer to the Minikube documentation for instructions
+on how to install it. It is recommended to allocate at least 8 CPUs and 8GB of memory to the Minikube cluster.
+
 Running the integration tests requires a Spark distribution package tarball that
 contains Spark jars, submission clients, etc. You can download a tarball from
 http://spark.apache.org/downloads.html. Or, you can create a distribution from

integration-test/pom.xml

@@ -39,6 +39,7 @@
     <scalatest-maven-plugin.version>1.0</scalatest-maven-plugin.version>
     <slf4j-log4j12.version>1.7.24</slf4j-log4j12.version>
     <sbt.project.name>kubernetes-integration-tests</sbt.project.name>
+    <spark.version></spark.version>
     <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>
@@ -141,37 +142,6 @@
           </execution>
         </executions>
       </plugin>
-      <plugin>
-        <groupId>com.googlecode.maven-download-plugin</groupId>
-        <artifactId>download-maven-plugin</artifactId>
-        <version>${download-maven-plugin.version}</version>
-        <executions>
-          <execution>
-            <id>download-minikube-linux</id>
-            <phase>pre-integration-test</phase>
-            <goals>
-              <goal>wget</goal>
-            </goals>
-            <configuration>
-              <url>https://storage.googleapis.com/minikube/releases/v0.22.0/minikube-linux-amd64</url>
-              <outputDirectory>${project.build.directory}/minikube-bin/linux-amd64</outputDirectory>
-              <outputFileName>minikube</outputFileName>
-            </configuration>
-          </execution>
-          <execution>
-            <id>download-minikube-darwin</id>
-            <phase>pre-integration-test</phase>
-            <goals>
-              <goal>wget</goal>
-            </goals>
-            <configuration>
-              <url>https://storage.googleapis.com/minikube/releases/v0.22.0/minikube-darwin-amd64</url>
-              <outputDirectory>${project.build.directory}/minikube-bin/darwin-amd64</outputDirectory>
-              <outputFileName>minikube</outputFileName>
-            </configuration>
-          </execution>
-        </executions>
-      </plugin>
       <plugin>
         <!-- Triggers scalatest plugin in the integration-test phase instead of
              the test phase. -->

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

@@ -27,8 +27,10 @@ import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Seconds, Span}
 
 import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackendFactory
-import org.apache.spark.deploy.k8s.integrationtest.constants.MINIKUBE_TEST_BACKEND
-import org.apache.spark.deploy.k8s.integrationtest.constants.SPARK_DISTRO_PATH
+import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.MinikubeTestBackend
+import org.apache.spark.deploy.k8s.integrationtest.constants._
+import org.apache.spark.deploy.k8s.integrationtest.config._
+
 
 private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll with BeforeAndAfter {
 
@@ -50,6 +52,9 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
   before {
     sparkAppConf = kubernetesTestComponents.newSparkAppConf()
       .set("spark.kubernetes.driver.label.spark-app-locator", APP_LOCATOR_LABEL)
+      .set(INIT_CONTAINER_DOCKER_IMAGE, tagImage("spark-init"))
+      .set(DRIVER_DOCKER_IMAGE, tagImage("spark-driver"))
+      .set(EXECUTOR_DOCKER_IMAGE, tagImage("spark-executor"))
     kubernetesTestComponents.createNamespace()
   }
 
@@ -58,10 +63,12 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
   }
 
   test("Run SparkPi with no resources") {
+    doMinikubeCheck
     runSparkPiAndVerifyCompletion()
   }
 
   test("Run SparkPi with a very long application name.") {
+    doMinikubeCheck
     sparkAppConf.set("spark.app.name", "long" * 40)
     runSparkPiAndVerifyCompletion()
   }
@@ -100,6 +107,10 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
       }
     }
   }
+  private def doMinikubeCheck(): Unit = {
+    assume(testBackend == MinikubeTestBackend)
+  }
+  private def tagImage(image: String): String = s"$image:${testBackend.dockerImageTag()}"
 }
 
 private[spark] object KubernetesSuite {

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

@@ -19,13 +19,41 @@ package org.apache.spark.deploy.k8s.integrationtest
 import java.io.Closeable
 import java.net.URI
 
+import java.io.{IOException,InputStream,OutputStream}
+
 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 tryWithSafeFinally[T](block: => T)(finallyBlock: => Unit): T = {
+    var originalThrowable: Throwable = null
+    try {
+      block
+    } catch {
+      case t: Throwable =>
+        // Purposefully not using NonFatal, because even fatal exceptions
+        // we don't want to have our finallyBlock suppress
+        originalThrowable = t
+        throw originalThrowable
+    } finally {
+      try {
+        finallyBlock
+      } catch {
+        case t: Throwable =>
+          if (originalThrowable != null) {
+            originalThrowable.addSuppressed(t)
+            logWarning(s"Suppressing exception in finally: " + t.getMessage, t)
+            throw originalThrowable
+          } else {
+            throw t
+          }
+      }
+    }
+  }
+
   def checkAndGetK8sMasterUrl(rawMasterURL: String): String = {
     require(rawMasterURL.startsWith("k8s://"),
       "Kubernetes master URL must start with k8s://.")
@@ -57,4 +85,30 @@ object Utils extends Logging {
 
     s"k8s://$resolvedURL"
   }
+
+  class RedirectThread(
+     in: InputStream,
+     out: OutputStream,
+     name: String,
+     propagateEof: Boolean = false) extends Thread(name) {
+      setDaemon(true)
+      override def run() {
+        scala.util.control.Exception.ignoring(classOf[IOException]) {
+          // FIXME: We copy the stream on the level of bytes to avoid encoding problems.
+          Utils.tryWithSafeFinally {
+            val buf = new Array[Byte](1024)
+            var len = in.read(buf)
+            while (len != -1) {
+              out.write(buf, 0, len)
+              out.flush()
+              len = in.read(buf)
+            }
+          } {
+            if (propagateEof) {
+              out.close()
+            }
+          }
+        }
+      }
+  }
 }

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

@@ -20,7 +20,7 @@ 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
+import org.apache.spark.deploy.k8s.integrationtest.config._
 
 private[spark] class GCETestBackend(val master: String) extends IntegrationTestBackend {
   private var defaultClient: DefaultKubernetesClient = _
@@ -37,5 +37,7 @@ private[spark] class GCETestBackend(val master: String) extends IntegrationTestB
     defaultClient
   }
 
-  override def name(): String = GCE_TEST_BACKEND
+  override def dockerImageTag(): String = {
+    return System.getProperty(KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY, "latest")
+  }
 }

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

@@ -21,18 +21,19 @@ import io.fabric8.kubernetes.client.DefaultKubernetesClient
 
 import org.apache.spark.deploy.k8s.integrationtest.backend.GCE.GCETestBackend
 import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.MinikubeTestBackend
+import org.apache.spark.deploy.k8s.integrationtest.docker.KubernetesSuiteDockerManager
 
 private[spark] trait IntegrationTestBackend {
-  def name(): String
   def initialize(): Unit
   def getKubernetesClient(): DefaultKubernetesClient
+  def dockerImageTag(): String
   def cleanUp(): Unit = {}
 }
 
 private[spark] object IntegrationTestBackendFactory {
   def getTestBackend(): IntegrationTestBackend = {
     Option(System.getProperty("spark.kubernetes.test.master"))
       .map(new GCETestBackend(_))
-      .getOrElse(new MinikubeTestBackend())
+      .getOrElse(MinikubeTestBackend)
   }
 }

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

@@ -20,73 +20,37 @@ import java.nio.file.Paths
 
 import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
 
-import org.apache.commons.lang3.SystemUtils
 import org.apache.spark.deploy.k8s.integrationtest.{Logging, ProcessUtils}
 
 // TODO support windows
 private[spark] object Minikube extends Logging {
-  private val MINIKUBE_EXECUTABLE_DEST = if (SystemUtils.IS_OS_MAC_OSX) {
-    Paths.get("target", "minikube-bin", "darwin-amd64", "minikube").toFile
-  } else if (SystemUtils.IS_OS_WINDOWS) {
-    throw new IllegalStateException("Executing Minikube based integration tests not yet " +
-      " available on Windows.")
-  } else {
-    Paths.get("target", "minikube-bin", "linux-amd64", "minikube").toFile
-  }
-
-  private val EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE = "Minikube is not downloaded, expected at " +
-    s"${MINIKUBE_EXECUTABLE_DEST.getAbsolutePath}"
-
   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) {
-      executeMinikube("start", "--memory", "6000", "--cpus", "8")
-    } else {
-      logInfo("Minikube is already started.")
-    }
-  }
-
   def getMinikubeIp: String = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists(), EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
     val outputs = executeMinikube("ip")
       .filter(_.matches("^\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}$"))
     assert(outputs.size == 1, "Unexpected amount of output from minikube ip")
     outputs.head
   }
 
   def getMinikubeStatus: MinikubeStatus.Value = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists(), EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
     val statusString = executeMinikube("status")
-      .filter(_.contains("minikube: "))
+      .filter(line => line.contains("minikubeVM: ") || line.contains("minikube:"))
       .head
+      .replaceFirst("minikubeVM: ", "")
       .replaceFirst("minikube: ", "")
     MinikubeStatus.unapply(statusString)
         .getOrElse(throw new IllegalStateException(s"Unknown status $statusString"))
   }
 
   def getDockerEnv: Map[String, String] = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists(), EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
     executeMinikube("docker-env", "--shell", "bash")
         .filter(_.startsWith("export"))
         .map(_.replaceFirst("export ", "").split('='))
         .map(arr => (arr(0), arr(1).replaceAllLiterally("\"", "")))
         .toMap
   }
 
-  def deleteMinikube(): Unit = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists, EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
-    if (getMinikubeStatus != MinikubeStatus.NONE) {
-      executeMinikube("delete")
-    } else {
-      logInfo("Minikube was already not running.")
-    }
-  }
-
   def getKubernetesClient: DefaultKubernetesClient = synchronized {
     val kubernetesMaster = s"https://${getMinikubeIp}:8443"
     val userHome = System.getProperty("user.home")
@@ -105,13 +69,8 @@ private[spark] object Minikube extends Logging {
   }
 
   private def executeMinikube(action: String, args: String*): Seq[String] = {
-    if (!MINIKUBE_EXECUTABLE_DEST.canExecute) {
-      if (!MINIKUBE_EXECUTABLE_DEST.setExecutable(true)) {
-        throw new IllegalStateException("Failed to make the Minikube binary executable.")
-      }
-    }
-    ProcessUtils.executeProcess(Array(MINIKUBE_EXECUTABLE_DEST.getAbsolutePath, action) ++ args,
-      MINIKUBE_STARTUP_TIMEOUT_SECONDS)
+    ProcessUtils.executeProcess(
+      Array("minikube", action) ++ args, MINIKUBE_STARTUP_TIMEOUT_SECONDS)
   }
 }
 

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

@@ -16,32 +16,43 @@
  */
 package org.apache.spark.deploy.k8s.integrationtest.backend.minikube
 
+import java.util.UUID
+
 import io.fabric8.kubernetes.client.DefaultKubernetesClient
 
 import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackend
-import org.apache.spark.deploy.k8s.integrationtest.constants.MINIKUBE_TEST_BACKEND
-import org.apache.spark.deploy.k8s.integrationtest.docker.SparkDockerImageBuilder
+import org.apache.spark.deploy.k8s.integrationtest.config._
+import org.apache.spark.deploy.k8s.integrationtest.docker.KubernetesSuiteDockerManager
 
-private[spark] class MinikubeTestBackend extends IntegrationTestBackend {
+private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
   private var defaultClient: DefaultKubernetesClient = _
-
+  private val userProvidedDockerImageTag = Option(
+    System.getProperty(KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY))
+  private val resolvedDockerImageTag =
+    userProvidedDockerImageTag.getOrElse(UUID.randomUUID().toString.replaceAll("-", ""))
+  private val dockerManager = new KubernetesSuiteDockerManager(
+    Minikube.getDockerEnv, resolvedDockerImageTag)
   override def initialize(): Unit = {
-    Minikube.startMinikube()
-    if (!System.getProperty("spark.docker.test.skipBuildImages", "false").toBoolean) {
-      new SparkDockerImageBuilder(Minikube.getDockerEnv).buildSparkDockerImages()
+    val minikubeStatus = Minikube.getMinikubeStatus
+    require(minikubeStatus == MinikubeStatus.RUNNING,
+      s"Minikube must be running before integration tests can execute. Current status" +
+        s" is: $minikubeStatus")
+    if (userProvidedDockerImageTag.isEmpty) {
+      dockerManager.buildSparkDockerImages()
     }
     defaultClient = Minikube.getKubernetesClient
   }
 
-  override def getKubernetesClient(): DefaultKubernetesClient = {
-    defaultClient
-  }
 
   override def cleanUp(): Unit = {
-    if (!System.getProperty("spark.docker.test.persistMinikube", "false").toBoolean) {
-      Minikube.deleteMinikube()
+    super.cleanUp()
+    if (userProvidedDockerImageTag.isEmpty) {
+      dockerManager.deleteImages()
     }
   }
 
-  override def name(): String = MINIKUBE_TEST_BACKEND
+  override def getKubernetesClient(): DefaultKubernetesClient = {
+    defaultClient
+  }
+  override def dockerImageTag(): String = resolvedDockerImageTag
 }

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

@@ -0,0 +1,24 @@
+/*
+ * 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.integrationtest
+
+package object config {
+  val KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY = "spark.kubernetes.test.imageDockerTag"
+  val DRIVER_DOCKER_IMAGE = "spark.kubernetes.driver.docker.image"
+  val EXECUTOR_DOCKER_IMAGE = "spark.kubernetes.executor.docker.image"
+  val INIT_CONTAINER_DOCKER_IMAGE = "spark.kubernetes.initcontainer.docker.image"
+}