support python deps

Author: skonto

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -1522,8 +1522,8 @@ class SparkContext(config: SparkConf) extends Logging {
     val schemeCorrectedPath = uri.getScheme match {
       case null => new File(path).getCanonicalFile.toURI.toString
       case "local" =>
-        logWarning("File with 'local' scheme is not supported to add to file server, since " +
-          "it is already available on every node.")
+        logWarning(s"File with 'local' scheme $path is not supported to add to file server, " +
+          s"since it is already available on every node.")
         return
       case _ => path
     }

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -390,6 +390,7 @@ private[spark] class SparkSubmit extends Logging {
           downloadFileList(_, targetDir, sparkConf, hadoopConf, secMgr)
         }.orNull
         args.files = renameResourcesToLocalFS(args.files, localFiles)
+        args.pyFiles = renameResourcesToLocalFS(args.pyFiles, localPyFiles)
       }
     }
 

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

@@ -259,12 +259,19 @@ private[spark] object KubernetesUtils extends Logging {
       isLocalDependency(Utils.resolveURI(resource))
   }
 
-  def renameMainAppResource(resource: String, conf: SparkConf): String = {
+  def renameMainAppResource(
+      resource: String,
+      conf: Option[SparkConf] = None,
+      isJava: Boolean = true): String = {
     if (isLocalAndResolvable(resource)) {
-      SparkLauncher.NO_RESOURCE
+      if (isJava) {
+        SparkLauncher.NO_RESOURCE
+      } else {
+        uploadFileUri(resource, conf)
+      }
     } else {
       resource
-   }
+    }
   }
 
   def uploadFileUri(uri: String, conf: Option[SparkConf] = None): String = {

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

@@ -160,7 +160,7 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
       KUBERNETES_DRIVER_SUBMIT_CHECK.key -> "true",
       MEMORY_OVERHEAD_FACTOR.key -> overheadFactor.toString)
     // try upload local, resolvable files to a hadoop compatible file system
-    Seq(JARS, FILES).foreach { key =>
+    Seq(JARS, FILES, SUBMIT_PYTHON_FILES).foreach { key =>
       val value = conf.get(key).filter(uri => KubernetesUtils.isLocalAndResolvable(uri))
       val resolved = KubernetesUtils.uploadAndTransformFileUris(value, Some(conf.sparkConf))
       if (resolved.nonEmpty) {
@@ -170,4 +170,3 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
     additionalProps.toMap
   }
 }
-

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

@@ -62,7 +62,10 @@ private[spark] class DriverCommandFeatureStep(conf: KubernetesDriverConf)
   }
 
   private def configureForJava(pod: SparkPod, res: String): SparkPod = {
-    val driverContainer = baseDriverContainer(pod, res).build()
+    // re-write primary resource, app jar is also added to spark.jars by default in SparkSubmit
+    // no uploading takes place here
+    val newResName = KubernetesUtils.renameMainAppResource(res)
+    val driverContainer = baseDriverContainer(pod, newResName).build()
     SparkPod(pod.pod, driverContainer)
   }
 
@@ -73,7 +76,9 @@ private[spark] class DriverCommandFeatureStep(conf: KubernetesDriverConf)
           .withValue(conf.get(PYSPARK_MAJOR_PYTHON_VERSION))
         .build())
 
-    val pythonContainer = baseDriverContainer(pod, res)
+    // re-write primary resource to be the remote one and upload the related file
+    val newResName = KubernetesUtils.renameMainAppResource(res, Some(conf.sparkConf), false)
+    val pythonContainer = baseDriverContainer(pod, newResName)
       .addAllToEnv(pythonEnvs.asJava)
       .build()
 
@@ -86,17 +91,11 @@ private[spark] class DriverCommandFeatureStep(conf: KubernetesDriverConf)
   }
 
   private def baseDriverContainer(pod: SparkPod, resource: String): ContainerBuilder = {
-    // re-write primary resource, app jar is also added to spark.jars by default in SparkSubmit
-    val resolvedResource = if (conf.mainAppResource.isInstanceOf[JavaMainAppResource]) {
-      KubernetesUtils.renameMainAppResource(resource, conf.sparkConf)
-    } else {
-      resource
-    }
     new ContainerBuilder(pod.container)
       .addToArgs("driver")
       .addToArgs("--properties-file", SPARK_CONF_PATH)
       .addToArgs("--class", conf.mainClass)
-      .addToArgs(resolvedResource)
+      .addToArgs(resource)
       .addToArgs(conf.appArgs: _*)
   }
 }

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

@@ -125,16 +125,18 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
       .endSpec()
       .build()
 
-    kubernetesTestComponents
+    // try until the service from a previous test is deleted
+    Eventually.eventually(TIMEOUT, INTERVAL) (kubernetesTestComponents
       .kubernetesClient
       .services()
-      .create(cephService)
+      .create(cephService))
 
-    kubernetesTestComponents
+    // try until the stateful set of a previous test is deleted
+    Eventually.eventually(TIMEOUT, INTERVAL) (kubernetesTestComponents
       .kubernetesClient
       .apps()
       .statefulSets()
-      .create(cephStatefulSet)
+      .create(cephStatefulSet))
   }
 
  private def deleteCephStorage(): Unit = {
@@ -143,24 +145,26 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
       .apps()
       .statefulSets()
       .withName(cName)
+      .withGracePeriod(0)
       .delete()
 
     kubernetesTestComponents
       .kubernetesClient
       .services()
       .withName(svcName)
+      .withGracePeriod(0)
       .delete()
   }
 
-  test("Launcher client dependencies", k8sTestTag, MinikubeTag) {
+  test("Launcher java client dependencies", k8sTestTag, MinikubeTag) {
     val fileName = Utils.createTempFile(FILE_CONTENTS, HOST_PATH)
     try {
       setupCephStorage()
       val cephUrlStr = getServiceUrl(svcName)
       val cephUrl = new URL(cephUrlStr)
       val cephHost = cephUrl.getHost
       val cephPort = cephUrl.getPort
-      val examplesJar = Utils.getExamplesJarAbsolutePath(sparkHomeDir)
+      val examplesJar = Utils.getTestFileAbsolutePath(Utils.getExamplesJarName(), sparkHomeDir)
       val (accessKey, secretKey) = getCephCredentials()
       sparkAppConf
         .set("spark.hadoop.fs.s3a.access.key", accessKey)
@@ -183,6 +187,62 @@ private[spark] trait DepsTestsSuite { k8sSuite: KubernetesSuite =>
     }
   }
 
+  test("Launcher python client dependencies using py", k8sTestTag, MinikubeTag) {
+    val depsFile = Utils.getTestFileAbsolutePath("py_container_checks.py", sparkHomeDir)
+    testPythonDeps(depsFile)
+  }
+
+  test("Launcher python client dependencies using a zip file", k8sTestTag, MinikubeTag) {
+    val inDepsFile = Utils.getTestFileAbsolutePath("py_container_checks.py", sparkHomeDir)
+    val outDepsFile = s"${inDepsFile.substring(0, inDepsFile.lastIndexOf("."))}.zip"
+    Utils.createZipFile(inDepsFile, outDepsFile)
+    testPythonDeps(outDepsFile)
+  }
+
+  private def testPythonDeps(depsFile: String): Unit = {
+    try {
+      setupCephStorage()
+      val cephUrlStr = getServiceUrl(svcName)
+      val cephUrl = new URL(cephUrlStr)
+      val cephHost = cephUrl.getHost
+      val cephPort = cephUrl.getPort
+      val examplesJar = Utils.getTestFileAbsolutePath(Utils.getExamplesJarName(), sparkHomeDir)
+
+      val (accessKey, secretKey) = getCephCredentials()
+      sparkAppConf
+        .set("spark.kubernetes.container.image", pyImage)
+        .set("spark.kubernetes.pyspark.pythonVersion", "3")
+        .set("spark.hadoop.fs.s3a.access.key", accessKey)
+        .set("spark.hadoop.fs.s3a.secret.key", secretKey)
+        .set("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
+        .set("spark.hadoop.fs.s3a.endpoint", s"$cephHost:$cephPort")
+        .set("spark.kubernetes.file.upload.path", s"s3a://$bucket")
+        .set("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
+        .set("spark.jars.packages", "com.amazonaws:aws-java-sdk:" +
+          "1.7.4,org.apache.hadoop:hadoop-aws:2.7.6")
+        .set("spark.driver.extraJavaOptions", "-Divy.cache.dir=/tmp -Divy.home=/tmp")
+      createS3Bucket(accessKey, secretKey, cephUrlStr)
+      val pySparkFiles = Utils.getTestFileAbsolutePath("pyfiles.py", sparkHomeDir )
+      val pyContainerChecks = depsFile
+      runSparkApplicationAndVerifyCompletion(
+        appResource = pySparkFiles,
+        mainClass = "",
+        expectedLogOnCompletion = Seq(
+          "Python runtime version check is: True",
+          "Python environment version check is: True",
+          "Python runtime version check for executor is: True"),
+        appArgs = Array("python"),
+        driverPodChecker = doBasicDriverPyPodCheck,
+        executorPodChecker = doBasicExecutorPyPodCheck,
+        appLocator = appLocator,
+        isJVM = false,
+        pyFiles = Some(pyContainerChecks))
+    } finally {
+      // make sure this always runs
+      deleteCephStorage()
+    }
+  }
+
   // There isn't a cleaner way to get the credentials
   // when ceph-nano runs on k8s
   private def getCephCredentials(): (String, String) = {

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

@@ -16,14 +16,16 @@
  */
 package org.apache.spark.deploy.k8s.integrationtest
 
-import java.io.{Closeable, File, PrintWriter}
+import java.io.{Closeable, File, FileInputStream, FileOutputStream, PrintWriter}
 import java.nio.file.{Files, Path}
 import java.util.concurrent.CountDownLatch
+import java.util.zip.{ZipEntry, ZipOutputStream}
 
 import scala.collection.JavaConverters._
 
 import io.fabric8.kubernetes.client.dsl.ExecListener
 import okhttp3.Response
+import org.apache.commons.compress.utils.IOUtils
 import org.apache.commons.io.output.ByteArrayOutputStream
 
 import org.apache.spark.{SPARK_VERSION, SparkException}
@@ -131,4 +133,41 @@ object Utils extends Logging {
         s"under spark home test dir ${sparkHomeDir.toAbsolutePath}!")
     }
   }
+
+  def getTestFileAbsolutePath(fileName: String, sparkHomeDir: Path): String = {
+    val filePathsFound = Files
+      .walk(sparkHomeDir)
+      .filter(Files.isRegularFile(_))
+      .filter((f: Path) => {f.toFile.getName == fileName})
+    // we should not have more than one here under current test build dir
+    // we only need one though
+    val filePath = filePathsFound
+      .iterator()
+      .asScala
+      .map(_.toAbsolutePath.toString)
+      .toArray
+      .headOption
+    filePath match {
+      case Some(file) => file
+      case _ => throw new SparkException(s"No valid $fileName file was found " +
+        s"under spark home test dir ${sparkHomeDir.toAbsolutePath}!")
+    }
+  }
+
+  def createZipFile(inFile: String, outFile: String): Unit = {
+    try {
+      val fileToZip = new File(inFile)
+      val fis = new FileInputStream(fileToZip)
+      val fos = new FileOutputStream(outFile)
+      val zipOut = new ZipOutputStream(fos)
+      val zipEntry = new ZipEntry(fileToZip.getName)
+      zipOut.putNextEntry(zipEntry)
+      IOUtils.copy(fis, zipOut)
+      IOUtils.closeQuietly(fis)
+      IOUtils.closeQuietly(zipOut)
+    } catch {
+      case e: Exception => log.error(s"Failed to zip file: $inFile"); throw e
+    }
+  }
 }
+