Use a pre-installed Minikube instance for integration tests.

This changes the integration test workflow to assume that Minikube is
already installed on the testing machine. Previously, the integration
tests downloaded Minikube in the build cycle, and started/stopped the
Minikube VM on every test execution. However, this made it such that
multiple integration tests cannot run concurrently.

This commit allows multiple tests to share a single Minikube instance,
and also requires users that run integration tests to have Minikube
pre-installed. If the minikube instance has enough resources, multiple
tests can run against it at the same time. Each test needs to use its
own set of Docker images, so the docker image builder now tags images
uniquely on every test execution.

Author: mccheah

resource-managers/kubernetes/README.md

@@ -42,10 +42,12 @@ Below is a list of the submodules for this cluster manager and what they do.
 
 # Running the Kubernetes Integration Tests
 
-Note that the integration test framework is currently being heavily revised and 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. Refer to the Minikube documentation for instructions on  how to install it. It is recommended to allocate at
+least 88 CPUs and 8GB of memory to the Minikube cluster.
+
 Running any of the integration tests requires including `kubernetes-integration-tests` profile in the build command. In
 order to prepare the environment for running the integration tests, the `pre-integration-test` step must be run in Maven
 on the `resource-managers/kubernetes/integration-tests` module:

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

@@ -502,6 +502,9 @@ package object config extends Logging {
 
   private[spark] val KUBERNETES_NODE_SELECTOR_PREFIX = "spark.kubernetes.node.selector."
 
+  private[spark] val KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY =
+      "spark.kubernetes.test.imageDockerTag"
+
   private[spark] def resolveK8sMaster(rawMasterString: String): String = {
     if (!rawMasterString.startsWith("k8s://")) {
       throw new IllegalArgumentException("Master URL should start with k8s:// in Kubernetes mode.")

resource-managers/kubernetes/integration-tests/pom.xml

@@ -324,37 +324,6 @@
           </execution>
         </executions>
       </plugin>
-      <plugin>
-        <groupId>com.googlecode.maven-download-plugin</groupId>
-        <artifactId>download-maven-plugin</artifactId>
-        <version>1.3.0</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.16.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.16.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, so that test jobs are copied over beforehand.

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

@@ -32,7 +32,7 @@ import org.apache.spark.{SparkConf, SparkFunSuite, SSLOptions}
 import org.apache.spark.deploy.k8s.SSLUtils
 import org.apache.spark.deploy.k8s.config._
 import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackendFactory
-import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.Minikube
+import org.apache.spark.deploy.k8s.integrationtest.backend.minikube.{Minikube, MinikubeTestBackend}
 import org.apache.spark.deploy.k8s.integrationtest.constants.MINIKUBE_TEST_BACKEND
 import org.apache.spark.deploy.k8s.submit.{Client, ClientArguments, JavaMainAppResource, KeyAndCertPem, MainAppResource, PythonMainAppResource}
 import org.apache.spark.launcher.SparkLauncher
@@ -51,9 +51,13 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
     testBackend.initialize()
     kubernetesTestComponents = new KubernetesTestComponents(testBackend.getKubernetesClient)
     resourceStagingServerLauncher = new ResourceStagingServerLauncher(
-      kubernetesTestComponents.kubernetesClient.inNamespace(kubernetesTestComponents.namespace))
+        kubernetesTestComponents
+            .kubernetesClient
+            .inNamespace(kubernetesTestComponents.namespace), testBackend.dockerImageTag())
     staticAssetServerLauncher = new StaticAssetServerLauncher(
-      kubernetesTestComponents.kubernetesClient.inNamespace(kubernetesTestComponents.namespace))
+        kubernetesTestComponents
+            .kubernetesClient
+            .inNamespace(kubernetesTestComponents.namespace), testBackend.dockerImageTag())
   }
 
   override def afterAll(): Unit = {
@@ -62,8 +66,9 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
 
   before {
     sparkConf = kubernetesTestComponents.newSparkConf()
-      .set(INIT_CONTAINER_DOCKER_IMAGE, s"spark-init:latest")
-      .set(DRIVER_DOCKER_IMAGE, s"spark-driver:latest")
+      .set(INIT_CONTAINER_DOCKER_IMAGE, tagImage("spark-init"))
+      .set(DRIVER_DOCKER_IMAGE, tagImage("spark-driver"))
+      .set(EXECUTOR_DOCKER_IMAGE, tagImage("spark-executor"))
       .set(s"${KUBERNETES_DRIVER_LABEL_PREFIX}spark-app-locator", APP_LOCATOR_LABEL)
     kubernetesTestComponents.createNamespace()
   }
@@ -73,14 +78,13 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Run PySpark Job on file from SUBMITTER with --py-files") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     launchStagingServer(SSLOptions(), None)
     sparkConf
-      .set(DRIVER_DOCKER_IMAGE,
-        System.getProperty("spark.docker.test.driverImage", "spark-driver-py:latest"))
-      .set(EXECUTOR_DOCKER_IMAGE,
-        System.getProperty("spark.docker.test.executorImage", "spark-executor-py:latest"))
+      .set(DRIVER_DOCKER_IMAGE, tagImage("spark-driver-py"))
+      .set(EXECUTOR_DOCKER_IMAGE, tagImage("spark-executor-py"))
+
 
     runPySparkPiAndVerifyCompletion(
       PYSPARK_PI_SUBMITTER_LOCAL_FILE_LOCATION,
@@ -89,27 +93,25 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Run PySpark Job on file from CONTAINER with spark.jar defined") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     sparkConf.setJars(Seq(CONTAINER_LOCAL_HELPER_JAR_PATH))
     sparkConf
-      .set(DRIVER_DOCKER_IMAGE,
-      System.getProperty("spark.docker.test.driverImage", "spark-driver-py:latest"))
-      .set(EXECUTOR_DOCKER_IMAGE,
-      System.getProperty("spark.docker.test.executorImage", "spark-executor-py:latest"))
+      .set(DRIVER_DOCKER_IMAGE, tagImage("spark-driver-py"))
+      .set(EXECUTOR_DOCKER_IMAGE, tagImage("spark-executor-py"))
 
     runPySparkPiAndVerifyCompletion(PYSPARK_PI_CONTAINER_LOCAL_FILE_LOCATION, Seq.empty[String])
   }
 
   test("Simple submission test with the resource staging server.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     launchStagingServer(SSLOptions(), None)
     runSparkPiAndVerifyCompletion(SUBMITTER_LOCAL_MAIN_APP_RESOURCE)
   }
 
   test("Enable SSL on the resource staging server") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     val keyStoreAndTrustStore = SSLUtils.generateKeyStoreTrustStorePair(
       ipAddress = Minikube.getMinikubeIp,
@@ -136,14 +138,14 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Use container-local resources without the resource staging server") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     sparkConf.setJars(Seq(CONTAINER_LOCAL_HELPER_JAR_PATH))
     runSparkPiAndVerifyCompletion(CONTAINER_LOCAL_MAIN_APP_RESOURCE)
   }
 
   test("Dynamic executor scaling basic test") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     launchStagingServer(SSLOptions(), None)
     createShuffleServiceDaemonSet()
@@ -163,7 +165,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Use remote resources without the resource staging server.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     val assetServerUri = staticAssetServerLauncher.launchStaticAssetServer()
     sparkConf.setJars(Seq(
       s"$assetServerUri/${EXAMPLES_JAR_FILE.getName}",
@@ -173,7 +175,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Mix remote resources with submitted ones.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     launchStagingServer(SSLOptions(), None)
     val assetServerUri = staticAssetServerLauncher.launchStaticAssetServer()
     sparkConf.setJars(Seq(
@@ -183,7 +185,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Use key and certificate PEM files for TLS.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     val keyAndCertificate = SSLUtils.generateKeyCertPemPair(Minikube.getMinikubeIp)
     launchStagingServer(
         SSLOptions(enabled = true),
@@ -195,7 +197,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Use client key and client cert file when requesting executors") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     sparkConf.setJars(Seq(
         CONTAINER_LOCAL_MAIN_APP_RESOURCE,
         CONTAINER_LOCAL_HELPER_JAR_PATH))
@@ -212,7 +214,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Added files should be placed in the driver's working directory.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     launchStagingServer(SSLOptions(), None)
     val testExistenceFileTempDir = Utils.createTempDir(namePrefix = "test-existence-file-temp-dir")
     val testExistenceFile = new File(testExistenceFileTempDir, "input.txt")
@@ -230,7 +232,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Setting JVM options on the driver and executors with spaces.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     launchStagingServer(SSLOptions(), None)
     val driverJvmOptionsFile = storeJvmOptionsInTempFile(
         Map("simpleDriverConf" -> "simpleDriverConfValue",
@@ -260,7 +262,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Submit small local files without the resource staging server.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
     sparkConf.setJars(Seq(CONTAINER_LOCAL_HELPER_JAR_PATH))
     val testExistenceFileTempDir = Utils.createTempDir(namePrefix = "test-existence-file-temp-dir")
     val testExistenceFile = new File(testExistenceFileTempDir, "input.txt")
@@ -278,15 +280,15 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Use a very long application name.") {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     sparkConf.setJars(Seq(CONTAINER_LOCAL_HELPER_JAR_PATH)).setAppName("long" * 40)
     runSparkPiAndVerifyCompletion(CONTAINER_LOCAL_MAIN_APP_RESOURCE)
   }
 
   private def launchStagingServer(
       resourceStagingServerSslOptions: SSLOptions, keyAndCertPem: Option[KeyAndCertPem]): Unit = {
-    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    assume(testBackend == MinikubeTestBackend)
 
     val resourceStagingServerPort = resourceStagingServerLauncher.launchStagingServer(
       resourceStagingServerSslOptions, keyAndCertPem)
@@ -368,7 +370,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
             .endVolume()
             .addNewContainer()
               .withName("shuffle")
-              .withImage("spark-shuffle:latest")
+              .withImage(s"spark-shuffle:${testBackend.dockerImageTag()}")
               .withImagePullPolicy("IfNotPresent")
               .addNewVolumeMount()
                 .withName("shuffle-dir")
@@ -404,6 +406,8 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
     }
     propertiesFile
   }
+
+  private def tagImage(image: String): String = s"$image:${testBackend.dockerImageTag()}"
 }
 
 private[spark] object KubernetesSuite {

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

@@ -32,7 +32,8 @@ import org.apache.spark.util.Utils
 /**
  * Launches a pod that runs the resource staging server, exposing it over a NodePort.
  */
-private[spark] class ResourceStagingServerLauncher(kubernetesClient: KubernetesClient) {
+private[spark] class ResourceStagingServerLauncher(
+    kubernetesClient: KubernetesClient, dockerImageTag: String) {
 
   private val SECRETS_ROOT_DIR = "/mnt/secrets/spark-staging"
   private val KEYSTORE_SECRET_KEY = "keyStore"
@@ -123,7 +124,7 @@ private[spark] class ResourceStagingServerLauncher(kubernetesClient: KubernetesC
           .endVolume()
         .addNewContainer()
           .withName("staging-server-container")
-          .withImage("spark-resource-staging-server:latest")
+          .withImage(s"spark-resource-staging-server:$dockerImageTag")
           .withImagePullPolicy("IfNotPresent")
           .withNewReadinessProbe()
             .withHttpGet(probePingHttpGet)

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

@@ -25,7 +25,8 @@ import org.apache.spark.util.Utils
  * Launches a simple HTTP server which provides jars that can be downloaded by Spark applications
  * in integration tests.
  */
-private[spark] class StaticAssetServerLauncher(kubernetesClient: KubernetesClient) {
+private[spark] class StaticAssetServerLauncher(
+    kubernetesClient: KubernetesClient, dockerImageTag: String) {
 
   // Returns the HTTP Base URI of the server.
   def launchStaticAssetServer(): String = {
@@ -46,7 +47,7 @@ private[spark] class StaticAssetServerLauncher(kubernetesClient: KubernetesClien
         .withNewSpec()
           .addNewContainer()
             .withName("static-asset-server-container")
-            .withImage("spark-integration-test-asset-server:latest")
+            .withImage(s"spark-integration-test-asset-server:$dockerImageTag")
             .withImagePullPolicy("IfNotPresent")
             .withNewReadinessProbe()
               .withHttpGet(probePingHttpGet)

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

@@ -18,9 +18,8 @@ package org.apache.spark.deploy.k8s.integrationtest.backend.GCE
 
 import io.fabric8.kubernetes.client.{ConfigBuilder, DefaultKubernetesClient}
 
-import org.apache.spark.deploy.k8s.config.resolveK8sMaster
+import org.apache.spark.deploy.k8s.config._
 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 = _
@@ -36,5 +35,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")
+  }
 }

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

@@ -21,19 +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.{Minikube, MinikubeTestBackend}
-import org.apache.spark.deploy.k8s.integrationtest.docker.SparkDockerImageBuilder
+import org.apache.spark.deploy.k8s.integrationtest.docker.SparkDockerImageManager
 
 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)
   }
 }