From a244db39cc1a99cf7031598024da023d05af574d Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Thu, 5 Oct 2017 17:30:33 -0700
Subject: [PATCH 01/10] 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.
---
 resource-managers/kubernetes/README.md        |  6 +-
 .../org/apache/spark/deploy/k8s/config.scala  |  3 +
 .../kubernetes/integration-tests/pom.xml      | 31 --------
 .../k8s/integrationtest/KubernetesSuite.scala | 62 ++++++++--------
 .../ResourceStagingServerLauncher.scala       |  5 +-
 .../StaticAssetServerLauncher.scala           |  5 +-
 .../backend/GCE/GCETestBackend.scala          |  7 +-
 .../backend/IntegrationTestBackend.scala      |  6 +-
 .../backend/minikube/Minikube.scala           | 45 ++----------
 .../minikube/MinikubeTestBackend.scala        | 39 +++++++----
 .../k8s/integrationtest/constants.scala       | 22 ------
 ...er.scala => SparkDockerImageManager.scala} | 70 +++++++++++++++----
 12 files changed, 140 insertions(+), 161 deletions(-)
 delete mode 100644 resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/constants.scala
 rename resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/{SparkDockerImageBuilder.scala => SparkDockerImageManager.scala} (66%)

diff --git a/resource-managers/kubernetes/README.md b/resource-managers/kubernetes/README.md
index 31b721d193362..b726b2509433d 100644
--- a/resource-managers/kubernetes/README.md
+++ b/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:
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/config.scala
index fc08c0ad42f82..cbae0703be3d2 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/config.scala
+++ b/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.")
diff --git a/resource-managers/kubernetes/integration-tests/pom.xml b/resource-managers/kubernetes/integration-tests/pom.xml
index d95f2f68db58f..3d50bdaad5696 100644
--- a/resource-managers/kubernetes/integration-tests/pom.xml
+++ b/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.
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
index d6e12df9d1ba2..85cea10a70d76 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
+++ b/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,7 +280,7 @@ 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)
@@ -286,7 +288,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
 
   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 {
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ResourceStagingServerLauncher.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ResourceStagingServerLauncher.scala
index 1aa90f4ce691d..f24f1cd9b85ff 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ResourceStagingServerLauncher.scala
+++ b/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)
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/StaticAssetServerLauncher.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/StaticAssetServerLauncher.scala
index 69a9df63bc4af..425f7d1a8e069 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/StaticAssetServerLauncher.scala
+++ b/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)
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/GCE/GCETestBackend.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/GCE/GCETestBackend.scala
index 09a3118a202eb..d2f51d449f707 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/GCE/GCETestBackend.scala
+++ b/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")
+  }
 }
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
index b61daf65a4093..155be7528815b 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
@@ -21,12 +21,12 @@ 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 = {}
 }
 
@@ -34,6 +34,6 @@ private[spark] object IntegrationTestBackendFactory {
   def getTestBackend(): IntegrationTestBackend = {
     Option(System.getProperty("spark.kubernetes.test.master"))
       .map(new GCETestBackend(_))
-      .getOrElse(new MinikubeTestBackend())
+      .getOrElse(MinikubeTestBackend)
   }
 }
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
index bd31bde6dabf4..2f003e09cee23 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
@@ -26,31 +26,9 @@ import org.apache.spark.util.Utils
 
 // TODO support windows
 private[spark] object Minikube extends Logging {
-  private val MINIKUBE_EXECUTABLE_DEST = if (Utils.isMac) {
-    Paths.get("target", "minikube-bin", "darwin-amd64", "minikube").toFile
-  } else if (Utils.isWindows) {
-    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
 
-  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")
@@ -58,17 +36,16 @@ private[spark] object Minikube extends Logging {
   }
 
   def getMinikubeStatus: MinikubeStatus.Value = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists(), EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
     val statusString = executeMinikube("status")
-      .filter(_.contains("minikubeVM: "))
+      .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('='))
@@ -76,15 +53,6 @@ private[spark] object Minikube extends Logging {
         .toMap
   }
 
-  def deleteMinikube(): Unit = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists, EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
-    if (getMinikubeStatus != MinikubeStatus.DOES_NOT_EXIST) {
-      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")
@@ -103,13 +71,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)
   }
 }
 
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
index 8e94f13360536..6c02eb8c1ce41 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
@@ -16,30 +16,45 @@
  */
 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.config.KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY
 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.docker.SparkDockerImageManager
+
+private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
 
-private[spark] class 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 dockerImageManager = new SparkDockerImageManager(
+      Minikube.getDockerEnv, resolvedDockerImageTag)
 
   override def initialize(): Unit = {
-    Minikube.startMinikube()
-    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) {
+       dockerImageManager.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) {
+      dockerImageManager.deleteImages()
     }
   }
 
-  override def name(): String = MINIKUBE_TEST_BACKEND
+  override def getKubernetesClient(): DefaultKubernetesClient = {
+    defaultClient
+  }
+
+  override def dockerImageTag(): String = resolvedDockerImageTag
 }
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/constants.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/constants.scala
deleted file mode 100644
index 0807a68cd823c..0000000000000
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/constants.scala
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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 constants {
-  val MINIKUBE_TEST_BACKEND = "minikube"
-  val GCE_TEST_BACKEND = "gce"
-}
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageBuilder.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala
similarity index 66%
rename from resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageBuilder.scala
rename to resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala
index 0e2fced70c9f7..6b43c142533a3 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageBuilder.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala
@@ -16,24 +16,23 @@
  */
 package org.apache.spark.deploy.k8s.integrationtest.docker
 
-import java.io.File
+import java.io.{File, PrintWriter}
 import java.net.URI
 import java.nio.file.Paths
 
-import scala.collection.JavaConverters._
-
+import com.google.common.base.Charsets
+import com.google.common.io.Files
 import com.spotify.docker.client.{DefaultDockerClient, DockerCertificates, LoggingBuildHandler}
 import org.apache.http.client.utils.URIBuilder
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Seconds, Span}
+import scala.collection.JavaConverters._
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.util.RedirectThread
+import org.apache.spark.util.{RedirectThread, Utils}
 
-
-
-private[spark] class SparkDockerImageBuilder
-  (private val dockerEnv: Map[String, String]) extends Logging{
+private[spark] class SparkDockerImageManager(
+    dockerEnv: Map[String, String], dockerTag: String) extends Logging {
 
   private val DOCKER_BUILD_PATH = Paths.get("target", "docker")
   // Dockerfile paths must be relative to the build path.
@@ -46,7 +45,7 @@ private[spark] class SparkDockerImageBuilder
   private val INIT_CONTAINER_DOCKER_FILE = "dockerfiles/init-container/Dockerfile"
   private val STAGING_SERVER_DOCKER_FILE = "dockerfiles/resource-staging-server/Dockerfile"
   private val STATIC_ASSET_SERVER_DOCKER_FILE =
-    "dockerfiles/integration-test-asset-server/Dockerfile"
+      "dockerfiles/integration-test-asset-server/Dockerfile"
   private val TIMEOUT = PatienceConfiguration.Timeout(Span(2, Minutes))
   private val INTERVAL = PatienceConfiguration.Interval(Span(2, Seconds))
   private val dockerHost = dockerEnv.getOrElse("DOCKER_HOST",
@@ -97,11 +96,54 @@ private[spark] class SparkDockerImageBuilder
     buildImage("spark-integration-test-asset-server", STATIC_ASSET_SERVER_DOCKER_FILE)
   }
 
+  def deleteImages(): Unit = {
+    deleteImage("spark-driver")
+    deleteImage("spark-driver-py")
+    deleteImage("spark-executor")
+    deleteImage("spark-executor-py")
+    deleteImage("spark-shuffle")
+    deleteImage("spark-resource-staging-server")
+    deleteImage("spark-init")
+    deleteImage("spark-integration-test-asset-server")
+    deleteImage("spark-base")
+  }
+
   private def buildImage(name: String, dockerFile: String): Unit = {
-    dockerClient.build(
-      DOCKER_BUILD_PATH,
-      name,
-      dockerFile,
-      new LoggingBuildHandler())
+    log.info(s"Building Docker image - $name:$dockerTag")
+    val dockerFileWithBaseTag = new File(DOCKER_BUILD_PATH.resolve(
+        s"$dockerFile-$dockerTag").toAbsolutePath.toString)
+    dockerFileWithBaseTag.deleteOnExit()
+    try {
+      val originalDockerFileText = Files.readLines(
+          DOCKER_BUILD_PATH.resolve(dockerFile).toFile, Charsets.UTF_8).asScala
+      val dockerFileTextWithProperBaseImage = originalDockerFileText.map(
+          _.replace("FROM spark-base", s"FROM spark-base:$dockerTag"))
+      Utils.tryWithResource(Files.newWriter(dockerFileWithBaseTag, Charsets.UTF_8)) { fileWriter =>
+        Utils.tryWithResource(new PrintWriter(fileWriter)) { printWriter =>
+          for (line <- dockerFileTextWithProperBaseImage) {
+            // scalastyle:off println
+            printWriter.println(line)
+            // scalastyle:on println
+          }
+        }
+      }
+      dockerClient.build(
+          DOCKER_BUILD_PATH,
+          s"$name:$dockerTag",
+          s"$dockerFile-$dockerTag",
+          new LoggingBuildHandler())
+    } finally {
+      dockerFileWithBaseTag.delete()
+    }
+  }
+
+  private def deleteImage(name: String): Unit = {
+    try {
+      dockerClient.removeImage(s"$name:$dockerTag")
+    } catch {
+      case e: RuntimeException =>
+        logWarning(s"Failed to delete image $name:$dockerTag. There may be images leaking in the" +
+          s" docker environment which are now stale and unused.", e)
+    }
   }
 }

From 12287ffa707eac4c361b8b49055e3aca30a74470 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Thu, 5 Oct 2017 17:44:05 -0700
Subject: [PATCH 02/10] Fix docs

---
 resource-managers/kubernetes/README.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/README.md b/resource-managers/kubernetes/README.md
index b726b2509433d..76bc4785d4ceb 100644
--- a/resource-managers/kubernetes/README.md
+++ b/resource-managers/kubernetes/README.md
@@ -45,8 +45,8 @@ Below is a list of the submodules for this cluster manager and what they do.
 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.
+your machine. 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 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

From bbe2a48e63c95ef0fac5242a4ae7304fc36116f5 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Thu, 5 Oct 2017 17:47:36 -0700
Subject: [PATCH 03/10] Add some clarifications for docs.

---
 resource-managers/kubernetes/README.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/README.md b/resource-managers/kubernetes/README.md
index 76bc4785d4ceb..1eff6a6c8b84f 100644
--- a/resource-managers/kubernetes/README.md
+++ b/resource-managers/kubernetes/README.md
@@ -45,8 +45,8 @@ Below is a list of the submodules for this cluster manager and what they do.
 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 8 CPUs and 8GB of memory to the Minikube cluster.
+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 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

From 8e1e3817213a81236df7a5f0aaf060c164ddba06 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Tue, 24 Oct 2017 11:51:48 -0700
Subject: [PATCH 04/10] Fix compilation error

---
 .../spark/deploy/k8s/integrationtest/KubernetesSuite.scala       | 1 -
 1 file changed, 1 deletion(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
index 85cea10a70d76..4a3a82ed8c0e8 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
@@ -33,7 +33,6 @@ 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, 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
 import org.apache.spark.util.Utils

From cd77c2ac12ef914b75bec8c6ddc43c24b27fc5c4 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Tue, 24 Oct 2017 18:33:10 -0700
Subject: [PATCH 05/10] Force containers to halt before removing images.

---
 .../backend/IntegrationTestBackend.scala      |  2 +-
 .../minikube/MinikubeTestBackend.scala        |  8 +--
 ...ala => KubernetesSuiteDockerManager.scala} | 62 ++++++++++++++++---
 3 files changed, 60 insertions(+), 12 deletions(-)
 rename resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/{SparkDockerImageManager.scala => KubernetesSuiteDockerManager.scala} (74%)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
index 155be7528815b..29b91431c9b24 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/IntegrationTestBackend.scala
@@ -21,7 +21,7 @@ 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.SparkDockerImageManager
+import org.apache.spark.deploy.k8s.integrationtest.docker.KubernetesSuiteDockerManager
 
 private[spark] trait IntegrationTestBackend {
   def initialize(): Unit
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
index 6c02eb8c1ce41..ea12b90e4b203 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/MinikubeTestBackend.scala
@@ -22,7 +22,7 @@ import io.fabric8.kubernetes.client.DefaultKubernetesClient
 
 import org.apache.spark.deploy.k8s.config.KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY
 import org.apache.spark.deploy.k8s.integrationtest.backend.IntegrationTestBackend
-import org.apache.spark.deploy.k8s.integrationtest.docker.SparkDockerImageManager
+import org.apache.spark.deploy.k8s.integrationtest.docker.KubernetesSuiteDockerManager
 
 private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
 
@@ -31,7 +31,7 @@ private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
       System.getProperty(KUBERNETES_TEST_DOCKER_TAG_SYSTEM_PROPERTY))
   private val resolvedDockerImageTag =
       userProvidedDockerImageTag.getOrElse(UUID.randomUUID().toString.replaceAll("-", ""))
-  private val dockerImageManager = new SparkDockerImageManager(
+  private val dockerManager = new KubernetesSuiteDockerManager(
       Minikube.getDockerEnv, resolvedDockerImageTag)
 
   override def initialize(): Unit = {
@@ -40,7 +40,7 @@ private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
         s"Minikube must be running before integration tests can execute. Current status" +
             s" is: $minikubeStatus")
     if (userProvidedDockerImageTag.isEmpty) {
-       dockerImageManager.buildSparkDockerImages()
+       dockerManager.buildSparkDockerImages()
     }
     defaultClient = Minikube.getKubernetesClient
   }
@@ -48,7 +48,7 @@ private[spark] object MinikubeTestBackend extends IntegrationTestBackend {
   override def cleanUp(): Unit = {
     super.cleanUp()
     if (userProvidedDockerImageTag.isEmpty) {
-      dockerImageManager.deleteImages()
+      dockerManager.deleteImages()
     }
   }
 
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
similarity index 74%
rename from resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala
rename to resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
index 6b43c142533a3..3cced02856f70 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/SparkDockerImageManager.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
@@ -23,15 +23,18 @@ import java.nio.file.Paths
 import com.google.common.base.Charsets
 import com.google.common.io.Files
 import com.spotify.docker.client.{DefaultDockerClient, DockerCertificates, LoggingBuildHandler}
+import com.spotify.docker.client.DockerClient.{ListContainersParam, RemoveContainerParam}
+import com.spotify.docker.client.messages.Container
 import org.apache.http.client.utils.URIBuilder
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
 import org.scalatest.time.{Minutes, Seconds, Span}
 import scala.collection.JavaConverters._
 
+import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite
 import org.apache.spark.internal.Logging
 import org.apache.spark.util.{RedirectThread, Utils}
 
-private[spark] class SparkDockerImageManager(
+private[spark] class KubernetesSuiteDockerManager(
     dockerEnv: Map[String, String], dockerTag: String) extends Logging {
 
   private val DOCKER_BUILD_PATH = Paths.get("target", "docker")
@@ -62,12 +65,12 @@ private[spark] class SparkDockerImageManager(
       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())
-    .build()
+      .uri(httpsDockerUri)
+      .dockerCertificates(DockerCertificates
+          .builder()
+          .dockerCertPath(Paths.get(dockerCerts))
+          .build().get())
+      .build()
 
   def buildSparkDockerImages(): Unit = {
     Eventually.eventually(TIMEOUT, INTERVAL) { dockerClient.ping() }
@@ -97,6 +100,7 @@ private[spark] class SparkDockerImageManager(
   }
 
   def deleteImages(): Unit = {
+    removeRunningContainers()
     deleteImage("spark-driver")
     deleteImage("spark-driver-py")
     deleteImage("spark-executor")
@@ -137,6 +141,50 @@ private[spark] class SparkDockerImageManager(
     }
   }
 
+  /**
+    * Forces all containers running an image with the configured tag to halt and be removed.
+    */
+  private def removeRunningContainers(): Unit = {
+    Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
+      val runningContainersWithImageTag = stopRunningContainers()
+      require(
+          runningContainersWithImageTag.nonEmpty,
+          s"${runningContainersWithImageTag.size} containers found still running" +
+              s" with the image tag $dockerTag")
+    }
+    dockerClient.listContainers(ListContainersParam.allContainers())
+        .asScala
+        .filter(containerHasImageWithTag(_))
+        .foreach(container => dockerClient.removeContainer(container.id())
+
+  }
+
+  private def stopRunningContainers(): Iterable[Container] = {
+    val runningContainersWithImageTag = getRunningContainersWithImageTag()
+    if (runningContainersWithImageTag.nonEmpty) {
+      log.info(s"Found ${runningContainersWithImageTag.size} containers running with" +
+        s" an image with the tag $dockerTag. Attempting to remove these containers," +
+        s" and then will stall for 2 seconds.")
+      runningContainersWithImageTag.foreach { container =>
+        dockerClient.stopContainer(container.id(), 5)
+      }
+    }
+    runningContainersWithImageTag
+  }
+
+  private def getRunningContainersWithImageTag(): Iterable[Container] = {
+    dockerClient
+        .listContainers(
+            ListContainersParam.allContainers(),
+            ListContainersParam.withStatusRunning())
+        .asScala
+        .filter(containerHasImageWithTag(_)
+  }
+
+  private def containerHasImageWithTag(container: Container): Boolean = {
+    container.image().endsWith(s":$dockerTag")
+  }
+
   private def deleteImage(name: String): Unit = {
     try {
       dockerClient.removeImage(s"$name:$dockerTag")

From ace42185e8ced10b3b470ba3d1e684630859bf85 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Wed, 25 Oct 2017 11:18:13 -0700
Subject: [PATCH 06/10] Fix compilation errors

---
 .../integrationtest/docker/KubernetesSuiteDockerManager.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
index 3cced02856f70..0fe5e96abbf66 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
@@ -155,7 +155,7 @@ private[spark] class KubernetesSuiteDockerManager(
     dockerClient.listContainers(ListContainersParam.allContainers())
         .asScala
         .filter(containerHasImageWithTag(_))
-        .foreach(container => dockerClient.removeContainer(container.id())
+        .foreach(container => dockerClient.removeContainer(container.id()))
 
   }
 
@@ -178,7 +178,7 @@ private[spark] class KubernetesSuiteDockerManager(
             ListContainersParam.allContainers(),
             ListContainersParam.withStatusRunning())
         .asScala
-        .filter(containerHasImageWithTag(_)
+        .filter(containerHasImageWithTag(_))
   }
 
   private def containerHasImageWithTag(container: Container): Boolean = {

From 9a6884ec761f356dc8ef6628c7bf09a02551524f Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Wed, 25 Oct 2017 14:05:01 -0700
Subject: [PATCH 07/10] Invert the requirement check

---
 .../integrationtest/docker/KubernetesSuiteDockerManager.scala   | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
index 0fe5e96abbf66..bc045a2d90fe6 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
@@ -148,7 +148,7 @@ private[spark] class KubernetesSuiteDockerManager(
     Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
       val runningContainersWithImageTag = stopRunningContainers()
       require(
-          runningContainersWithImageTag.nonEmpty,
+          runningContainersWithImageTag.isEmpty,
           s"${runningContainersWithImageTag.size} containers found still running" +
               s" with the image tag $dockerTag")
     }

From 70fee6dc0913f5615f72dc0dc5011eaf8db3a24d Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Wed, 25 Oct 2017 14:51:11 -0700
Subject: [PATCH 08/10] Wait for all containers to disappear before deleting
 images.

---
 .../docker/KubernetesSuiteDockerManager.scala      | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
index bc045a2d90fe6..2fc4306d9f0ae 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
@@ -142,8 +142,8 @@ private[spark] class KubernetesSuiteDockerManager(
   }
 
   /**
-    * Forces all containers running an image with the configured tag to halt and be removed.
-    */
+   * Forces all containers running an image with the configured tag to halt and be removed.
+   */
   private def removeRunningContainers(): Unit = {
     Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
       val runningContainersWithImageTag = stopRunningContainers()
@@ -155,7 +155,15 @@ private[spark] class KubernetesSuiteDockerManager(
     dockerClient.listContainers(ListContainersParam.allContainers())
         .asScala
         .filter(containerHasImageWithTag(_))
-        .foreach(container => dockerClient.removeContainer(container.id()))
+        .foreach(container => dockerClient.removeContainer(
+            container.id(), RemoveContainerParam.forceKill(true)))
+    Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
+      val containersWithImageTag = dockerClient.listContainers(ListContainersParam.allContainers())
+        .asScala
+        .filter(containerHasImageWithTag(_))
+      require(containersWithImageTag.isEmpty, s"${containersWithImageTag.size} containers still" +
+        s" found with image tag $dockerTag.")
+    }
 
   }
 

From 1780d8ed75dde1b75019877d7af8f3e288bb4fa6 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Fri, 1 Dec 2017 16:48:49 -0800
Subject: [PATCH 09/10] Use ImageIds to look up containers and match image ids
 with repo tags

---
 .../docker/KubernetesSuiteDockerManager.scala | 25 ++++++++++---------
 1 file changed, 13 insertions(+), 12 deletions(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
index 2fc4306d9f0ae..fae266e335871 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/docker/KubernetesSuiteDockerManager.scala
@@ -23,7 +23,7 @@ import java.nio.file.Paths
 import com.google.common.base.Charsets
 import com.google.common.io.Files
 import com.spotify.docker.client.{DefaultDockerClient, DockerCertificates, LoggingBuildHandler}
-import com.spotify.docker.client.DockerClient.{ListContainersParam, RemoveContainerParam}
+import com.spotify.docker.client.DockerClient.{ListContainersParam, ListImagesParam, RemoveContainerParam}
 import com.spotify.docker.client.messages.Container
 import org.apache.http.client.utils.URIBuilder
 import org.scalatest.concurrent.{Eventually, PatienceConfiguration}
@@ -145,8 +145,13 @@ private[spark] class KubernetesSuiteDockerManager(
    * Forces all containers running an image with the configured tag to halt and be removed.
    */
   private def removeRunningContainers(): Unit = {
+    val imageIds = dockerClient.listImages(ListImagesParam.allImages())
+        .asScala
+        .filter(image => image.repoTags().asScala.exists(_.endsWith(s":$dockerTag")))
+        .map(_.id())
+        .toSet
     Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
-      val runningContainersWithImageTag = stopRunningContainers()
+      val runningContainersWithImageTag = stopRunningContainers(imageIds)
       require(
           runningContainersWithImageTag.isEmpty,
           s"${runningContainersWithImageTag.size} containers found still running" +
@@ -154,21 +159,21 @@ private[spark] class KubernetesSuiteDockerManager(
     }
     dockerClient.listContainers(ListContainersParam.allContainers())
         .asScala
-        .filter(containerHasImageWithTag(_))
+        .filter(container => imageIds.contains(container.imageId()))
         .foreach(container => dockerClient.removeContainer(
             container.id(), RemoveContainerParam.forceKill(true)))
     Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
       val containersWithImageTag = dockerClient.listContainers(ListContainersParam.allContainers())
         .asScala
-        .filter(containerHasImageWithTag(_))
+        .filter(container => imageIds.contains(container.imageId()))
       require(containersWithImageTag.isEmpty, s"${containersWithImageTag.size} containers still" +
         s" found with image tag $dockerTag.")
     }
 
   }
 
-  private def stopRunningContainers(): Iterable[Container] = {
-    val runningContainersWithImageTag = getRunningContainersWithImageTag()
+  private def stopRunningContainers(imageIds: Set[String]): Iterable[Container] = {
+    val runningContainersWithImageTag = getRunningContainersWithImageIds(imageIds)
     if (runningContainersWithImageTag.nonEmpty) {
       log.info(s"Found ${runningContainersWithImageTag.size} containers running with" +
         s" an image with the tag $dockerTag. Attempting to remove these containers," +
@@ -180,17 +185,13 @@ private[spark] class KubernetesSuiteDockerManager(
     runningContainersWithImageTag
   }
 
-  private def getRunningContainersWithImageTag(): Iterable[Container] = {
+  private def getRunningContainersWithImageIds(imageIds: Set[String]): Iterable[Container] = {
     dockerClient
         .listContainers(
             ListContainersParam.allContainers(),
             ListContainersParam.withStatusRunning())
         .asScala
-        .filter(containerHasImageWithTag(_))
-  }
-
-  private def containerHasImageWithTag(container: Container): Boolean = {
-    container.image().endsWith(s":$dockerTag")
+        .filter(container => imageIds.contains(container.imageId()))
   }
 
   private def deleteImage(name: String): Unit = {

From 31b24e81e42e1a3e161416e9eab062304ec405b9 Mon Sep 17 00:00:00 2001
From: mcheah <mcheah@palantir.com>
Date: Tue, 5 Dec 2017 12:54:30 -0800
Subject: [PATCH 10/10] Fix merge conflict

---
 .../backend/minikube/Minikube.scala           | 36 +------------------
 1 file changed, 1 insertion(+), 35 deletions(-)

diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
index ed5fb3c93ce9f..8c51d358a9314 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/backend/minikube/Minikube.scala
@@ -37,22 +37,10 @@ private[spark] object Minikube extends Logging {
 
   def getMinikubeStatus: MinikubeStatus.Value = synchronized {
     val statusString = executeMinikube("status")
-<<<<<<< HEAD
-      .filter(line => line.contains("minikubeVM: ") || line.contains("minikube"))
-||||||| merged common ancestors
-      .filter(_.contains("minikubeVM: "))
-=======
-      .filter(_.contains("minikube: "))
->>>>>>> origin/branch-2.2-kubernetes
+      .filter(line => line.contains("minikubeVM: ") || line.contains("minikube:"))
       .head
-<<<<<<< HEAD
       .replaceFirst("minikubeVM: ", "")
       .replaceFirst("minikube: ", "")
-||||||| merged common ancestors
-      .replaceFirst("minikubeVM: ", "")
-=======
-      .replaceFirst("minikube: ", "")
->>>>>>> origin/branch-2.2-kubernetes
     MinikubeStatus.unapply(statusString)
         .getOrElse(throw new IllegalStateException(s"Unknown status $statusString"))
   }
@@ -65,28 +53,6 @@ private[spark] object Minikube extends Logging {
         .toMap
   }
 
-<<<<<<< HEAD
-||||||| merged common ancestors
-  def deleteMinikube(): Unit = synchronized {
-    assert(MINIKUBE_EXECUTABLE_DEST.exists, EXPECTED_DOWNLOADED_MINIKUBE_MESSAGE)
-    if (getMinikubeStatus != MinikubeStatus.DOES_NOT_EXIST) {
-      executeMinikube("delete")
-    } else {
-      logInfo("Minikube was already not running.")
-    }
-  }
-
-=======
-  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.")
-    }
-  }
-
->>>>>>> origin/branch-2.2-kubernetes
   def getKubernetesClient: DefaultKubernetesClient = synchronized {
     val kubernetesMaster = s"https://${getMinikubeIp}:8443"
     val userHome = System.getProperty("user.home")