Merge remote-tracking branch 'apache-spark-on-k8s/k8s-support-alternate-incremental' into nodeport-upload

mccheah · mccheah · commit f7ca5553eac9 · 2017-01-27T14:17:42.000-08:00
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -464,8 +464,6 @@ object SparkSubmit extends CommandLineUtils {
         sysProp = "spark.kubernetes.namespace"),
       OptionAssigner(args.kubernetesUploadJars, KUBERNETES, CLUSTER,
         sysProp = "spark.kubernetes.driver.uploads.jars"),
-      OptionAssigner(args.kubernetesUploadDriverExtraClasspath, KUBERNETES, CLUSTER,
-        sysProp = "spark.kubernetes.driver.uploads.driverExtraClasspath"),
 
         // Other options
       OptionAssigner(args.executorCores, STANDALONE | YARN, ALL_DEPLOY_MODES,
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
@@ -74,7 +74,6 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
   // Kubernetes only
   var kubernetesNamespace: String = null
   var kubernetesUploadJars: String = null
-  var kubernetesUploadDriverExtraClasspath: String = null
 
   // Standalone cluster mode only
   var supervise: Boolean = false
@@ -197,9 +196,6 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
     kubernetesUploadJars = Option(kubernetesUploadJars)
       .orElse(sparkProperties.get("spark.kubernetes.driver.uploads.jars"))
       .orNull
-    kubernetesUploadDriverExtraClasspath = Option(kubernetesUploadDriverExtraClasspath)
-      .orElse(sparkProperties.get("spark.kubernetes.driver.uploads.driverExtraClasspath"))
-      .orNull
 
     // Try to set main class from JAR if no --class argument is given
     if (mainClass == null && !isPython && !isR && primaryResource != null) {
@@ -444,9 +440,6 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       case KUBERNETES_UPLOAD_JARS =>
         kubernetesUploadJars = value
 
-      case KUBERNETES_UPLOAD_DRIVER_EXTRA_CLASSPATH =>
-        kubernetesUploadDriverExtraClasspath = value
-
       case HELP =>
         printUsageAndExit(0)
 
diff --git a/launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java b/launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java
@@ -80,8 +80,6 @@ class SparkSubmitOptionParser {
   protected final String KUBERNETES_MASTER = "--kubernetes-master";
   protected final String KUBERNETES_NAMESPACE = "--kubernetes-namespace";
   protected final String KUBERNETES_UPLOAD_JARS = "--upload-jars";
-  protected final String KUBERNETES_UPLOAD_DRIVER_EXTRA_CLASSPATH =
-          "--upload-driver-extra-classpath";
 
   /**
    * This is the canonical list of spark-submit options. Each entry in the array contains the
@@ -124,8 +122,7 @@ class SparkSubmitOptionParser {
     { TOTAL_EXECUTOR_CORES },
     { KUBERNETES_MASTER },
     { KUBERNETES_NAMESPACE },
-    { KUBERNETES_UPLOAD_JARS },
-    { KUBERNETES_UPLOAD_DRIVER_EXTRA_CLASSPATH }
+    { KUBERNETES_UPLOAD_JARS }
   };
 
   /**
diff --git a/pom.xml b/pom.xml
@@ -223,6 +223,7 @@
     <PermGen>64m</PermGen>
     <MaxPermGen>512m</MaxPermGen>
     <CodeCacheSize>512m</CodeCacheSize>
+    <extraScalaTestArgs></extraScalaTestArgs>
   </properties>
   <repositories>
     <repository>
@@ -2096,7 +2097,7 @@
             <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
             <junitxml>.</junitxml>
             <filereports>SparkTestSuite.txt</filereports>
-            <argLine>-ea -Xmx3g -XX:MaxPermSize=${MaxPermGen} -XX:ReservedCodeCacheSize=${CodeCacheSize}</argLine>
+            <argLine>-ea -Xmx3g -XX:MaxPermSize=${MaxPermGen} -XX:ReservedCodeCacheSize=${CodeCacheSize} ${extraScalaTestArgs}</argLine>
             <stderr/>
             <environmentVariables>
               <!--
diff --git a/resource-managers/kubernetes/README.md b/resource-managers/kubernetes/README.md
@@ -51,6 +51,20 @@ Afterwards, the integration tests can be executed with Maven or your IDE. Note t
 `pre-integration-test` phase must be run every time the Spark main code changes. When running tests from the
 command line, the `pre-integration-test` phase should automatically be invoked if the `integration-test` phase is run.
 
+# Preserve the Minikube VM
+
+The integration tests make use of [Minikube](https://github.com/kubernetes/minikube), which fires up a virtual machine
+and setup a single-node kubernetes cluster within it. By default the vm is destroyed after the tests are finished.
+If you want to preserve the vm, e.g. to reduce the running time of tests during development, you can pass the property
+`spark.docker.test.persistMinikube` to the test process:
+
+```sh
+build/mvn integration-test \
+    -Pkubernetes -Pkubernetes-integration-tests \
+    -pl resource-managers/kubernetes/integration-tests -am \
+    -DextraScalaTestArgs=-Dspark.docker.test.persistMinikube=true
+```
+
 # Usage Guide
 
 See the [usage guide](../../docs/running-on-kubernetes.md) for more information.
diff --git a/resource-managers/kubernetes/core/pom.xml b/resource-managers/kubernetes/core/pom.xml
@@ -29,7 +29,7 @@
   <name>Spark Project Kubernetes</name>
   <properties>
     <sbt.project.name>kubernetes</sbt.project.name>
-    <kubernetes.client.version>1.4.17</kubernetes.client.version>
+    <kubernetes.client.version>1.4.34</kubernetes.client.version>
   </properties>
 
   <dependencies>
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala
@@ -51,7 +51,7 @@ private[spark] class Client(
   private val appName = sparkConf.getOption("spark.app.name")
     .orElse(sparkConf.getOption("spark.app.id"))
     .getOrElse("spark")
-  private val kubernetesAppId = s"$appName-$launchTime"
+  private val kubernetesAppId = s"$appName-$launchTime".toLowerCase.replaceAll("\\.", "-")
   private val secretName = s"spark-submission-server-secret-$kubernetesAppId"
   private val driverLauncherSelectorValue = s"driver-launcher-$launchTime"
   private val driverDockerImage = sparkConf.get(
@@ -97,7 +97,7 @@ private[spark] class Client(
 
     val k8ClientConfig = k8ConfBuilder.build
     Utils.tryWithResource(new DefaultKubernetesClient(k8ClientConfig))(kubernetesClient => {
-      val secret = kubernetesClient.secrets().createNew()
+      val applicationSubmitSecret = kubernetesClient.secrets().createNew()
         .withNewMetadata()
         .withName(secretName)
         .endMetadata()
@@ -114,7 +114,7 @@ private[spark] class Client(
         val secretDirectory = s"$SPARK_SUBMISSION_SECRET_BASE_DIR/$kubernetesAppId"
         val submitPending = new AtomicBoolean(false)
         val podWatcher = new DriverPodWatcher(submitCompletedFuture, submitPending,
-          kubernetesClient, driverKubernetesSelectors)
+          kubernetesClient, applicationSubmitSecret, driverKubernetesSelectors)
         Utils.tryWithResource(kubernetesClient
             .pods()
             .withLabels(driverKubernetesSelectors)
@@ -128,7 +128,9 @@ private[spark] class Client(
               .withRestartPolicy("OnFailure")
               .addNewVolume()
                 .withName(s"spark-submission-secret-volume")
-                .withNewSecret().withSecretName(secret.getMetadata.getName).endSecret()
+                .withNewSecret()
+                  .withSecretName(applicationSubmitSecret.getMetadata.getName)
+                  .endSecret()
                 .endVolume
               .withServiceAccount(serviceAccount)
               .addNewContainer()
@@ -173,7 +175,7 @@ private[spark] class Client(
           }
         }
       } finally {
-        kubernetesClient.secrets().delete(secret)
+        kubernetesClient.secrets().delete(applicationSubmitSecret)
       }
     })
   }
@@ -182,6 +184,7 @@ private[spark] class Client(
       submitCompletedFuture: SettableFuture[Boolean],
       submitPending: AtomicBoolean,
       kubernetesClient: KubernetesClient,
+      applicationSubmitSecret: Secret,
       driverKubernetesSelectors: java.util.Map[String, String]) extends Watcher[Pod] {
     override def eventReceived(action: Action, pod: Pod): Unit = {
       if ((action == Action.ADDED || action == Action.MODIFIED)
@@ -194,6 +197,17 @@ private[spark] class Client(
             .find(status =>
               status.getName == DRIVER_LAUNCHER_CONTAINER_NAME && status.getReady) match {
             case Some(_) =>
+              val ownerRefs = Seq(new OwnerReferenceBuilder()
+                .withName(pod.getMetadata.getName)
+                .withUid(pod.getMetadata.getUid)
+                .withApiVersion(pod.getApiVersion)
+                .withKind(pod.getKind)
+                .withController(true)
+                .build())
+
+              applicationSubmitSecret.getMetadata.setOwnerReferences(ownerRefs.asJava)
+              kubernetesClient.secrets().createOrReplace(applicationSubmitSecret)
+
               val driverLauncherServicePort = new ServicePortBuilder()
                 .withName(DRIVER_LAUNCHER_SERVICE_PORT_NAME)
                 .withPort(DRIVER_LAUNCHER_SERVICE_INTERNAL_PORT)
@@ -203,6 +217,7 @@ private[spark] class Client(
                 .withNewMetadata()
                   .withName(kubernetesAppId)
                   .withLabels(driverKubernetesSelectors)
+                  .withOwnerReferences(ownerRefs.asJava)
                   .endMetadata()
                 .withNewSpec()
                   .withType("NodePort")
@@ -213,6 +228,7 @@ private[spark] class Client(
               try {
                 sparkConf.set("spark.kubernetes.driver.service.name",
                   service.getMetadata.getName)
+                sparkConf.set("spark.kubernetes.driver.pod.name", kubernetesAppId)
                 sparkConf.setIfMissing("spark.driver.port", DEFAULT_DRIVER_PORT.toString)
                 sparkConf.setIfMissing("spark.blockmanager.port",
                   DEFAULT_BLOCKMANAGER_PORT.toString)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/KubernetesSparkRestServer.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/KubernetesSparkRestServer.scala
@@ -135,11 +135,6 @@ private[spark] class KubernetesSparkRestServer(
               } else {
                 val tempDir = Utils.createTempDir()
                 val appResourcePath = resolvedAppResource(appResource, tempDir)
-                val driverClasspathDirectory = new File(tempDir, "driver-extra-classpath")
-                if (!driverClasspathDirectory.mkdir) {
-                  throw new IllegalStateException("Failed to create driver extra classpath" +
-                    s" dir at ${driverClasspathDirectory.getAbsolutePath}")
-                }
                 val jarsDirectory = new File(tempDir, "jars")
                 if (!jarsDirectory.mkdir) {
                   throw new IllegalStateException("Failed to create jars dir at" +
@@ -173,6 +168,10 @@ private[spark] class KubernetesSparkRestServer(
                 val driverMemory = resolvedSparkProperties.getOrElse("spark.driver.memory", "1g")
                 command += s"-Xms$driverMemory"
                 command += s"-Xmx$driverMemory"
+                val extraJavaOpts = resolvedSparkProperties.get("spark.driver.extraJavaOptions")
+                  .map(Utils.splitCommandString)
+                  .getOrElse(Seq.empty)
+                command ++= extraJavaOpts
                 command += mainClass
                 command ++= appArgs
                 val pb = new ProcessBuilder(command: _*).inheritIO()
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala
@@ -60,6 +60,11 @@ private[spark] class KubernetesClusterSchedulerBackend(
     .getOrElse(
       throw new SparkException("Must specify the service name the driver is running with"))
 
+  private val kubernetesDriverPodName = conf
+    .getOption("spark.kubernetes.driver.pod.name")
+    .getOrElse(
+      throw new SparkException("Must specify the driver pod name"))
+
   private val executorMemory = conf.getOption("spark.executor.memory").getOrElse("1g")
   private val executorMemoryBytes = Utils.byteStringAsBytes(executorMemory)
 
@@ -82,6 +87,15 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private val kubernetesClient = KubernetesClientBuilder
     .buildFromWithinPod(kubernetesMaster, kubernetesNamespace)
 
+  val driverPod = try {
+    kubernetesClient.pods().inNamespace(kubernetesNamespace).
+      withName(kubernetesDriverPodName).get()
+  } catch {
+    case throwable: Throwable =>
+      logError(s"Executor cannot find driver pod.", throwable)
+      throw new SparkException(s"Executor cannot find driver pod", throwable)
+  }
+
   override val minRegisteredRatio =
     if (conf.getOption("spark.scheduler.minRegisteredResourcesRatio").isEmpty) {
       0.8
@@ -199,7 +213,15 @@ private[spark] class KubernetesClusterSchedulerBackend(
       .withNewMetadata()
         .withName(name)
         .withLabels(selectors)
-        .endMetadata()
+        .withOwnerReferences()
+        .addNewOwnerReference()
+          .withController(true)
+          .withApiVersion(driverPod.getApiVersion)
+          .withKind(driverPod.getKind)
+          .withName(driverPod.getMetadata.getName)
+          .withUid(driverPod.getMetadata.getUid)
+        .endOwnerReference()
+      .endMetadata()
       .withNewSpec()
         .addNewContainer()
           .withName(s"exec-${applicationId()}-container")
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile
@@ -1,9 +1,11 @@
-FROM anapsix/alpine-java:8
+FROM openjdk:8-alpine
 
 # If this docker file is being used in the context of building your images from a Spark distribution, the docker build
 # command should be invoked from the top level directory of the Spark distribution. E.g.:
 # docker build -t spark-driver:latest -f dockerfiles/driver/Dockerfile .
 
+RUN apk upgrade --update
+RUN apk add --update bash
 RUN mkdir -p /opt/spark
 RUN touch /opt/spark/RELEASE
 
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile
@@ -1,9 +1,11 @@
-FROM anapsix/alpine-java:8
+FROM openjdk:8-alpine
 
 # If this docker file is being used in the context of building your images from a Spark distribution, the docker build
 # command should be invoked from the top level directory of the Spark distribution. E.g.:
 # docker build -t spark-executor:latest -f dockerfiles/executor/Dockerfile .
 
+RUN apk upgrade --update
+RUN apk add --update bash
 RUN mkdir -p /opt/spark
 RUN touch /opt/spark/RELEASE
 
diff --git a/resource-managers/kubernetes/integration-tests/src/test/resources/log4j.properties b/resource-managers/kubernetes/integration-tests/src/test/resources/log4j.properties
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+# Set everything to be logged to the file target/integration-tests.log
+log4j.rootCategory=INFO, file
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.append=true
+log4j.appender.file.file=target/integration-tests.log
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss.SSS} %t %p %c{1}: %m%n
+
+# Ignore messages below warning level from a few verbose libraries.
+log4j.logger.com.sun.jersey=WARN
+log4j.logger.org.apache.hadoop=WARN
+log4j.logger.org.eclipse.jetty=WARN
+log4j.logger.org.mortbay=WARN
+log4j.logger.org.spark_project.jetty=WARN
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala
@@ -36,6 +36,7 @@ import org.apache.spark.deploy.kubernetes.Client
 import org.apache.spark.deploy.kubernetes.integrationtest.docker.SparkDockerImageBuilder
 import org.apache.spark.deploy.kubernetes.integrationtest.minikube.Minikube
 import org.apache.spark.deploy.kubernetes.integrationtest.restapis.SparkRestApiV1
+import org.apache.spark.internal.Logging
 import org.apache.spark.status.api.v1.{ApplicationStatus, StageStatus}
 import org.apache.spark.util.Utils
 
@@ -82,8 +83,15 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
 
   before {
     Eventually.eventually(TIMEOUT, INTERVAL) {
-      assert(minikubeKubernetesClient.pods().list().getItems.isEmpty)
-      assert(minikubeKubernetesClient.services().list().getItems.isEmpty)
+      val podsList = minikubeKubernetesClient.pods().list()
+      assert(podsList == null
+        || podsList.getItems == null
+        || podsList.getItems.isEmpty
+      )
+      val servicesList = minikubeKubernetesClient.services().list()
+      assert(servicesList == null
+        || servicesList.getItems == null
+        || servicesList.getItems.isEmpty)
     }
   }
 
@@ -139,6 +147,9 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   test("Run a simple example") {
+    // We'll make assertions based on spark rest api, so we need to turn on
+    // spark.ui.enabled explicitly since the scalatest-maven-plugin would set it
+    // to false by default.
     val sparkConf = new SparkConf(true)
       .setMaster(s"k8s://https://${Minikube.getMinikubeIp}:8443")
       .set("spark.kubernetes.submit.caCertFile", clientConfig.getCaCertFile)
@@ -152,6 +163,8 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
       .set("spark.executor.cores", "1")
       .set("spark.executors.instances", "1")
       .set("spark.app.id", "spark-pi")
+      .set("spark.ui.enabled", "true")
+      .set("spark.testing", "false")
     val mainAppResource = s"file://$EXAMPLES_JAR"
 
     new Client(
@@ -174,6 +187,8 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
       "--num-executors", "1",
       "--upload-jars", HELPER_JAR,
       "--class", MAIN_CLASS,
+      "--conf", "spark.ui.enabled=true",
+      "--conf", "spark.testing=false",
       "--conf", s"spark.kubernetes.submit.caCertFile=${clientConfig.getCaCertFile}",
       "--conf", s"spark.kubernetes.submit.clientKeyFile=${clientConfig.getClientKeyFile}",
       "--conf", s"spark.kubernetes.submit.clientCertFile=${clientConfig.getClientCertFile}",
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/docker/SparkDockerImageBuilder.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/docker/SparkDockerImageBuilder.scala
@@ -55,5 +55,4 @@ private[spark] class SparkDockerImageBuilder(private val dockerEnv: Map[String,
     dockerClient.build(Paths.get("target", "docker", "driver"), "spark-driver")
     dockerClient.build(Paths.get("target", "docker", "executor"), "spark-executor")
   }
-
-}
+}
