Merge pull request apache-spark-on-k8s#258 from palantir/resync-kube

ash211 · web-flow · commit a72151fc7ae2 · 2017-09-04T23:10:29.000-07:00
[NOSQUASH] Resync kube
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/MountSmallLocalFilesStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/MountSmallLocalFilesStep.scala
@@ -34,7 +34,8 @@ private[spark] class MountSmallLocalFilesStep(
 
   import MountSmallLocalFilesStep._
   override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
-    val localFiles = KubernetesFileUtils.getOnlySubmitterLocalFiles(sparkFiles).map(new File(_))
+    val localFiles = KubernetesFileUtils.getOnlySubmitterLocalFiles(sparkFiles)
+        .map(localFileUri => new File(Utils.resolveURI(localFileUri).getPath))
     val totalSizeBytes = localFiles.map(_.length()).sum
     val totalSizeBytesString = Utils.bytesToString(totalSizeBytes)
     require(totalSizeBytes < MAX_SECRET_BUNDLE_SIZE_BYTES,
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
@@ -457,7 +457,16 @@ private[spark] class KubernetesClusterSchedulerBackend(
         .withValue(cp)
         .build()
     }
-    val requiredEnv = (Seq(
+    val executorExtraJavaOptionsEnv = conf
+        .get(org.apache.spark.internal.config.EXECUTOR_JAVA_OPTIONS)
+        .map { opts =>
+          val delimitedOpts = Utils.splitCommandString(opts)
+          delimitedOpts.zipWithIndex.map {
+            case (opt, index) =>
+              new EnvVarBuilder().withName(s"$ENV_JAVA_OPT_PREFIX$index").withValue(opt).build()
+          }
+        }.getOrElse(Seq.empty[EnvVar])
+    val executorEnv = (Seq(
       (ENV_EXECUTOR_PORT, executorPort.toString),
       (ENV_DRIVER_URL, driverUrl),
       // Executor backend expects integral value for executor cores, so round it up to an int.
@@ -477,7 +486,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
           .withNewFieldRef("v1", "status.podIP")
           .build())
         .build()
-      )
+      ) ++ executorExtraJavaOptionsEnv ++ executorExtraClasspathEnv.toSeq
     val requiredPorts = Seq(
       (EXECUTOR_PORT_NAME, executorPort),
       (BLOCK_MANAGER_PORT_NAME, blockmanagerPort))
@@ -497,8 +506,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
         .addToLimits("memory", executorMemoryLimitQuantity)
         .addToRequests("cpu", executorCpuQuantity)
       .endResources()
-      .addAllToEnv(requiredEnv.asJava)
-      .addToEnv(executorExtraClasspathEnv.toSeq: _*)
+      .addAllToEnv(executorEnv.asJava)
       .withPorts(requiredPorts.asJava)
       .build()
 
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/MountSmallLocalFilesStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/MountSmallLocalFilesStepSuite.scala
@@ -32,7 +32,7 @@ import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.deploy.kubernetes.submit.MountSmallFilesBootstrap
 import org.apache.spark.util.Utils
 
-private[spark] class MountSmallLocalFilesStepTest extends SparkFunSuite with BeforeAndAfter {
+private[spark] class MountSmallLocalFilesStepSuite extends SparkFunSuite with BeforeAndAfter {
 
   private val FIRST_TEMP_FILE_NAME = "file1.txt"
   private val SECOND_TEMP_FILE_NAME = "file2.txt"
@@ -56,11 +56,11 @@ private[spark] class MountSmallLocalFilesStepTest extends SparkFunSuite with Bef
 
   test("Local files should be added to the secret.") {
     val firstTempFile = createTempFileWithContents(
-      tempFolder, FIRST_TEMP_FILE_NAME, FIRST_TEMP_FILE_CONTENTS)
+        tempFolder, FIRST_TEMP_FILE_NAME, FIRST_TEMP_FILE_CONTENTS)
     val secondTempFile = createTempFileWithContents(
         tempFolder, SECOND_TEMP_FILE_NAME, SECOND_TEMP_FILE_CONTENTS)
     val sparkFiles = Seq(
-        firstTempFile.getAbsolutePath,
+        s"file://${firstTempFile.getAbsolutePath}",
         secondTempFile.getAbsolutePath,
         REMOTE_FILE_URI)
     val configurationStep = new MountSmallLocalFilesStep(
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile
@@ -38,10 +38,11 @@ ENV PYSPARK_PYTHON python
 ENV PYSPARK_DRIVER_PYTHON python
 ENV PYTHONPATH ${SPARK_HOME}/python/:${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${PYTHONPATH}
 
-# TODO support spark.executor.extraClassPath
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_EXECUTOR_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH}+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXECUTOR_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXECUTOR_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_FROM_SECRET_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_FROM_SECRET_DIR/." .; fi && \
-    ${JAVA_HOME}/bin/java -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
+    ${JAVA_HOME}/bin/java "${SPARK_EXECUTOR_JAVA_OPTS[@]}" -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
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
@@ -23,11 +23,12 @@ FROM spark-base
 
 COPY examples /opt/spark/examples
 
-# TODO support spark.executor.extraClassPath
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_EXECUTOR_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH}+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXECUTOR_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXECUTOR_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_FROM_SECRET_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_FROM_SECRET_DIR/." .; fi && \
-    ${JAVA_HOME}/bin/java -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
+    ${JAVA_HOME}/bin/java "${SPARK_EXECUTOR_JAVA_OPTS[@]}" -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
diff --git a/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/JavaOptionsTest.scala b/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/JavaOptionsTest.scala
@@ -29,28 +29,51 @@ private[spark] object JavaOptionsTest {
 
   def main(args: Array[String]): Unit = {
     // scalastyle:off println
-    if (args.length != 1) {
+    if (args.length != 2) {
       println(s"Invalid arguments: ${args.mkString(",")}." +
-        s"Usage: JavaOptionsTest <driver-java-options-list-file>")
+        s"Usage: JavaOptionsTest <driver-java-options-list-file> <executor-java-options-list-file>")
       System.exit(1)
     }
     val expectedDriverJavaOptions = loadPropertiesFromFile(args(0))
+    val expectedExecutorJavaOptions = loadPropertiesFromFile(args(1))
     val nonMatchingDriverOptions = expectedDriverJavaOptions.filter {
       case (optKey, optValue) => System.getProperty(optKey) != optValue
     }
     if (nonMatchingDriverOptions.nonEmpty) {
       println(s"The driver's JVM options did not match. Expected $expectedDriverJavaOptions." +
         s" But these options did not match: $nonMatchingDriverOptions.")
       val sysProps = Maps.fromProperties(System.getProperties).asScala
-      println("System properties are:")
+      println("Driver system properties are:")
       for (prop <- sysProps) {
         println(s"Key: ${prop._1}, Value: ${prop._2}")
       }
       System.exit(1)
     }
 
-    // TODO support spark.executor.extraJavaOptions and test here.
-    println(s"All expected JVM options were present on the driver and executors.")
+    val spark = SparkSession.builder().getOrCreate().sparkContext
+    try {
+      val nonMatchingExecutorOptions = spark.parallelize(Seq(0)).flatMap { _ =>
+        expectedExecutorJavaOptions.filter {
+          case (optKey, optValue) => System.getProperty(optKey) != optValue
+        }
+      }.collectAsMap()
+      if (nonMatchingExecutorOptions.nonEmpty) {
+        val executorSysProps = spark.parallelize(Seq(0)).flatMap { _ =>
+          Maps.fromProperties(System.getProperties).asScala
+        }.collectAsMap()
+        println(s"The executor's JVM options did not match. Expected" +
+          s" $expectedExecutorJavaOptions. But these options did not" +
+          s" match: $nonMatchingExecutorOptions.")
+        println("Executor system properties are:")
+        for (prop <- executorSysProps) {
+          println(s"Key: ${prop._1}, Value: ${prop._2}")
+        }
+      } else {
+        println("All expected JVM options were present on the driver and executors.")
+      }
+    } finally {
+      spark.stop()
+    }
     // scalastyle:on println
   }
 
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
@@ -234,18 +234,28 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
     launchStagingServer(SSLOptions(), None)
     val driverJvmOptionsFile = storeJvmOptionsInTempFile(
         Map("simpleDriverConf" -> "simpleDriverConfValue",
-          "driverconfwithspaces" -> "driver conf with spaces value"),
+            "driverconfwithspaces" -> "driver conf with spaces value"),
         "driver-jvm-options.properties",
         "JVM options that should be set on the driver.")
+    val executorJvmOptionsFile = storeJvmOptionsInTempFile(
+        Map("simpleExecutorConf" -> "simpleExecutorConfValue",
+            "executor conf with spaces" -> "executor conf with spaces value"),
+        "executor-jvm-options.properties",
+        "JVM options that should be set on the executors.")
     sparkConf.set(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS,
         "-DsimpleDriverConf=simpleDriverConfValue" +
             " -Ddriverconfwithspaces='driver conf with spaces value'")
-    sparkConf.set("spark.files", driverJvmOptionsFile.getAbsolutePath)
+    sparkConf.set(SparkLauncher.EXECUTOR_EXTRA_JAVA_OPTIONS,
+        "-DsimpleExecutorConf=simpleExecutorConfValue" +
+            " -D\'executor conf with spaces\'=\'executor conf with spaces value\'")
+    sparkConf.set("spark.files",
+        Seq(driverJvmOptionsFile.getAbsolutePath, executorJvmOptionsFile.getAbsolutePath)
+            .mkString(","))
     runSparkApplicationAndVerifyCompletion(
         JavaMainAppResource(SUBMITTER_LOCAL_MAIN_APP_RESOURCE),
         JAVA_OPTIONS_MAIN_CLASS,
         Seq(s"All expected JVM options were present on the driver and executors."),
-        Array(driverJvmOptionsFile.getName),
+        Array(driverJvmOptionsFile.getName, executorJvmOptionsFile.getName),
         Seq.empty[String])
   }
 
