Skip to content
This repository was archived by the owner on Jan 9, 2020. It is now read-only.

Download remotely-located resources on driver startup. #240

Closed
wants to merge 1 commit into from
Closed

Download remotely-located resources on driver startup. #240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 38 additions & 6 deletions ...e-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,7 +365,7 @@ package object config extends Logging {
" resource staging server to download jars.")
.internal()
.stringConf
.createWithDefault(INIT_CONTAINER_DOWNLOAD_JARS_SECRET_PATH)
.createWithDefault(INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_JARS_SECRET_PATH)

private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_RESOURCE_IDENTIFIER =
ConfigBuilder("spark.kubernetes.driver.initcontainer.downloadFilesResourceIdentifier")
Expand All @@ -380,30 +380,62 @@ package object config extends Logging {
" resource staging server to download files.")
.internal()
.stringConf
.createWithDefault(INIT_CONTAINER_DOWNLOAD_FILES_SECRET_PATH)
.createWithDefault(INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_FILES_SECRET_PATH)

private[spark] val INIT_CONTAINER_REMOTE_JARS =
ConfigBuilder("spark.kubernetes.driver.initcontainer.remoteJars")
.doc("Comma-separated list of jar URIs to download in the init-container. This is inferred" +
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: inferred -> calculated

" from spark.jars.")
.internal()
.stringConf
.createOptional

private[spark] val INIT_CONTAINER_REMOTE_FILES =
ConfigBuilder("spark.kubernetes.driver.initcontainer.remoteFiles")
.doc("Comma-separated list of file URIs to download in the init-container. This is inferred" +
" from spark.files.")
.internal()
.stringConf
.createOptional

private[spark] val INIT_CONTAINER_DOCKER_IMAGE =
ConfigBuilder("spark.kubernetes.driver.initcontainer.docker.image")
.doc("Image for the driver's init-container that downloads mounted dependencies.")
.stringConf
.createWithDefault(s"spark-driver-init:$sparkVersion")

private[spark] val DRIVER_LOCAL_JARS_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.jarsDownloadDir")
private[spark] val DRIVER_SUBMITTED_JARS_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.submittedJars.downloadDir")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The config string is getting long and unwieldy. spark.kubernetes.driver.jars and spark.kubernetes.driver.files?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We probably want to indicate that this is where the jars are being downloaded to. spark.kubernetes.driver.jars is ambiguous in the sense that it could be jars that need to be uploaded or downloaded or added to its classpath, etc.

.doc("Location to download local jars to in the driver. When using spark-submit, this" +
" directory must be empty and will be mounted as an empty directory volume on the" +
" driver pod.")
.stringConf
.createWithDefault("/var/spark-data/spark-local-jars")

private[spark] val DRIVER_LOCAL_FILES_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.filesDownloadDir")
private[spark] val DRIVER_SUBMITTED_FILES_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.submittedFiles.downloadDir")
.doc("Location to download local files to in the driver. When using spark-submit, this" +
" directory must be empty and will be mounted as an empty directory volume on the" +
" driver pod.")
.stringConf
.createWithDefault("/var/spark-data/spark-local-files")

private[spark] val DRIVER_REMOTE_JARS_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.remoteJars.downloadDir")
.doc("Location to download remotely-located (e.g. HDFS) jars to in the driver. When" +
" using spark-submit, this directory must be empty and will be mounted as an empty" +
" directory volume on the driver pod.")
.stringConf
.createWithDefault("/var/spark-data/spark-remote-jars")

private[spark] val DRIVER_REMOTE_FILES_DOWNLOAD_LOCATION =
ConfigBuilder("spark.kubernetes.driver.mountdependencies.remoteFiles.downloadDir")
.doc("Location to download remotely-located (e.g. HDFS) files to in the driver. When" +
" using spark-submit, this directory must be empty and will be mounted as an empty" +
" directory volume on the driver pod.")
.stringConf
.createWithDefault("/var/spark-data/spark-remote-files")

private[spark] val DRIVER_MOUNT_DEPENDENCIES_INIT_TIMEOUT =
ConfigBuilder("spark.kubernetes.mountdependencies.mountTimeout")
.doc("Timeout before aborting the attempt to download and unpack local dependencies from" +
Expand Down
77 changes: 55 additions & 22 deletions ...anagers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,6 @@ package object constants {
private[spark] val ENV_EXECUTOR_ID = "SPARK_EXECUTOR_ID"
private[spark] val ENV_EXECUTOR_POD_IP = "SPARK_EXECUTOR_POD_IP"
private[spark] val ENV_DRIVER_MEMORY = "SPARK_DRIVER_MEMORY"
private[spark] val ENV_UPLOADED_JARS_DIR = "SPARK_UPLOADED_JARS_DIR"
private[spark] val ENV_SUBMIT_EXTRA_CLASSPATH = "SPARK_SUBMIT_EXTRA_CLASSPATH"
private[spark] val ENV_MOUNTED_CLASSPATH = "SPARK_MOUNTED_CLASSPATH"
private[spark] val ENV_DRIVER_MAIN_CLASS = "SPARK_DRIVER_CLASS"
Expand All @@ -92,25 +91,59 @@ package object constants {

// V2 submission init container
private[spark] val INIT_CONTAINER_ANNOTATION = "pod.beta.kubernetes.io/init-containers"
private[spark] val INIT_CONTAINER_SECRETS_VOLUME_NAME = "dependency-secret"
private[spark] val INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH = "/mnt/secrets/spark-init"
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_SECRET_KEY = "downloadJarsSecret"
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_SECRET_KEY = "downloadFilesSecret"
private[spark] val INIT_CONTAINER_TRUSTSTORE_SECRET_KEY = "trustStore"
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_SECRET_PATH =
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_DOWNLOAD_JARS_SECRET_KEY"
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_SECRET_PATH =
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_DOWNLOAD_FILES_SECRET_KEY"
private[spark] val INIT_CONTAINER_TRUSTSTORE_PATH =
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_TRUSTSTORE_SECRET_KEY"
private[spark] val INIT_CONTAINER_DOWNLOAD_CREDENTIALS_PATH =
"/mnt/secrets/kubernetes-credentials"
private[spark] val INIT_CONTAINER_CONFIG_MAP_KEY = "init-driver"
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_VOLUME = "init-container-properties"
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_MOUNT_PATH = "/etc/spark-init/"
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_NAME = "init-driver.properties"
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_PATH =
s"$INIT_CONTAINER_PROPERTIES_FILE_MOUNT_PATH/$INIT_CONTAINER_PROPERTIES_FILE_NAME"
private[spark] val DOWNLOAD_JARS_VOLUME_NAME = "download-jars"
private[spark] val DOWNLOAD_FILES_VOLUME_NAME = "download-files"

// Init container for downloading submitted files from the staging server.
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_CONTAINER_NAME =
"spark-driver-download-submitted-files"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_SECRETS_VOLUME_NAME =
"resource-staging-server-secret"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_SECRETS_VOLUME_MOUNT_PATH =
"/mnt/secrets/spark-init"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_JARS_SECRET_KEY =
"downloadSubmittedJarsSecret"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_FILES_SECRET_KEY =
"downloadSubmittedFilesSecret"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_TRUSTSTORE_SECRET_KEY = "trustStore"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_JARS_SECRET_PATH =
s"$INIT_CONTAINER_SUBMITTED_FILES_SECRETS_VOLUME_MOUNT_PATH/" +
s"$INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_JARS_SECRET_KEY"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_FILES_SECRET_PATH =
s"$INIT_CONTAINER_SUBMITTED_FILES_SECRETS_VOLUME_MOUNT_PATH/" +
s"$INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_FILES_SECRET_KEY"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_TRUSTSTORE_PATH =
s"$INIT_CONTAINER_SUBMITTED_FILES_SECRETS_VOLUME_MOUNT_PATH/" +
s"$INIT_CONTAINER_SUBMITTED_FILES_TRUSTSTORE_SECRET_KEY"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_CONFIG_MAP_KEY =
"download-submitted-files"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_VOLUME =
"download-submitted-files-properties"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_MOUNT_PATH =
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's a lot of growth in complexity here in terms of mount paths and other parameters. Is it possible for us to use fewer init-containers, or group these better?

Copy link
Author

@mccheah mccheah Apr 27, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could download all of the jars from all locations to the same directory (resource staging server and remote) as well as the files, but I'm concerned about file name conflicts and how to deduplicate those. Perhaps we should explicitly forbid that multiple URIs end with the same file name. @aash for thoughts.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Having identically named jars coming from different places sounds like an anti-pattern. I don't think prohibiting that is all that bad.

There's already the contract that file names must be unique as they're all downloaded into the cwd

"/etc/spark-init/download-submitted-files"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_NAME =
"init-driver-download-submitted-files.properties"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_PATH =
s"$INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_MOUNT_PATH/" +
s"$INIT_CONTAINER_SUBMITTED_FILES_PROPERTIES_FILE_NAME"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_JARS_VOLUME_NAME =
"download-submitted-jars"
private[spark] val INIT_CONTAINER_SUBMITTED_FILES_DOWNLOAD_FILES_VOLUME_NAME =
"download-submitted-files"

// Init container for fetching remote dependencies.
private[spark] val INIT_CONTAINER_REMOTE_FILES_CONTAINER_NAME =
"spark-driver-download-remote-files"
private[spark] val INIT_CONTAINER_REMOTE_FILES_CONFIG_MAP_KEY =
"download-remote-files"
private[spark] val INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_VOLUME =
"download-remote-files-properties"
private[spark] val INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_MOUNT_PATH =
"/etc/spark-init/download-remote-files"
private[spark] val INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_NAME =
"init-driver-download-remote-files.properties"
private[spark] val INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_PATH =
s"$INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_MOUNT_PATH/" +
s"$INIT_CONTAINER_REMOTE_FILES_PROPERTIES_FILE_NAME"
private[spark] val INIT_CONTAINER_REMOTE_FILES_DOWNLOAD_JARS_VOLUME_NAME = "download-remote-jars"
private[spark] val INIT_CONTAINER_REMOTE_FILES_DOWNLOAD_FILES_VOLUME_NAME =
"download-remote-files"
}
6 changes: 5 additions & 1 deletion ...t/kubernetes/v1/KubernetesFileUtils.scala → ...bernetes/submit/KubernetesFileUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.spark.deploy.rest.kubernetes.v1
package org.apache.spark.deploy.kubernetes.submit

import org.apache.spark.util.Utils

Expand All @@ -41,4 +41,8 @@ private[spark] object KubernetesFileUtils {
Option(Utils.resolveURI(uri).getScheme).getOrElse("file") == "file"
}

def getOnlyRemoteFiles(uris: Iterable[String]): Iterable[String] = {
filterUriStringsByScheme(uris, scheme => scheme != "file" && scheme != "local")
}

}
3 changes: 2 additions & 1 deletion .../kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/v1/Client.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,8 @@ import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.deploy.kubernetes.CompressionUtils
import org.apache.spark.deploy.kubernetes.config._
import org.apache.spark.deploy.kubernetes.constants._
import org.apache.spark.deploy.rest.kubernetes.v1.{AppResource, ContainerAppResource, HttpClientUtil, KubernetesCreateSubmissionRequest, KubernetesCredentials, KubernetesFileUtils, KubernetesSparkRestApi, RemoteAppResource, UploadedAppResource}
import org.apache.spark.deploy.kubernetes.submit.KubernetesFileUtils
import org.apache.spark.deploy.rest.kubernetes.v1.{AppResource, ContainerAppResource, HttpClientUtil, KubernetesCreateSubmissionRequest, KubernetesCredentials, KubernetesSparkRestApi, RemoteAppResource, UploadedAppResource}
import org.apache.spark.internal.Logging
import org.apache.spark.util.{ShutdownHookManager, Utils}

Expand Down
3 changes: 2 additions & 1 deletion ...a/org/apache/spark/deploy/kubernetes/submit/v1/DriverSubmitSslConfigurationProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ import scala.collection.JavaConverters._
import org.apache.spark.{SecurityManager => SparkSecurityManager, SparkConf, SparkException, SSLOptions}
import org.apache.spark.deploy.kubernetes.config._
import org.apache.spark.deploy.kubernetes.constants._
import org.apache.spark.deploy.rest.kubernetes.v1.{KubernetesFileUtils, PemsToKeyStoreConverter}
import org.apache.spark.deploy.kubernetes.submit.KubernetesFileUtils
import org.apache.spark.deploy.rest.kubernetes.v1.PemsToKeyStoreConverter
import org.apache.spark.util.Utils

/**
Expand Down
68 changes: 46 additions & 22 deletions .../kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/v2/Client.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,14 @@
package org.apache.spark.deploy.kubernetes.submit.v2

import java.io.File
import java.net.URI
import java.util.Collections

import io.fabric8.kubernetes.api.model.{ContainerBuilder, EnvVarBuilder, HasMetadata, OwnerReferenceBuilder, PodBuilder}
import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.{SecurityManager => SparkSecurityManager, SparkConf, SparkException}
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.deploy.kubernetes.config._
import org.apache.spark.deploy.kubernetes.constants._
import org.apache.spark.internal.Logging
Expand All @@ -50,7 +51,8 @@ private[spark] class Client(
appArgs: Array[String],
mainAppResource: String,
kubernetesClientProvider: SubmissionKubernetesClientProvider,
mountedDependencyManagerProvider: MountedDependencyManagerProvider) extends Logging {
submittedDependencyManagerProvider: SubmittedDependencyManagerProvider,
remoteDependencyManagerProvider: DownloadRemoteDependencyManagerProvider) extends Logging {

private val namespace = sparkConf.get(KUBERNETES_NAMESPACE)
private val master = resolveK8sMaster(sparkConf.get("spark.master"))
Expand Down Expand Up @@ -132,27 +134,33 @@ private[spark] class Client(
.endSpec()

val nonDriverPodKubernetesResources = mutable.Buffer[HasMetadata]()

// resolvedJars is all of the original Spark jars, but files on the local disk that
// were uploaded to the resource staging server are converted to the paths they would
// have been downloaded at. Similarly for resolvedFiles.
// If the resource staging server isn't being used, then resolvedJars = spark.jars.
val resolvedJars = mutable.Buffer[String]()
val resolvedFiles = mutable.Buffer[String]()
val driverPodWithMountedDeps = maybeStagingServerUri.map { stagingServerUri =>
val mountedDependencyManager = mountedDependencyManagerProvider.getMountedDependencyManager(
kubernetesAppId,
stagingServerUri,
allLabels,
namespace,
sparkJars,
sparkFiles)
val jarsResourceIdentifier = mountedDependencyManager.uploadJars()
val filesResourceIdentifier = mountedDependencyManager.uploadFiles()
val initContainerKubernetesSecret = mountedDependencyManager.buildInitContainerSecret(
val submittedDependencyManager = submittedDependencyManagerProvider
.getSubmittedDependencyManager(
kubernetesAppId,
stagingServerUri,
allLabels,
namespace,
sparkJars,
sparkFiles)
val jarsResourceIdentifier = submittedDependencyManager.uploadJars()
val filesResourceIdentifier = submittedDependencyManager.uploadFiles()
val initContainerKubernetesSecret = submittedDependencyManager.buildInitContainerSecret(
jarsResourceIdentifier.resourceSecret, filesResourceIdentifier.resourceSecret)
val initContainerConfigMap = mountedDependencyManager.buildInitContainerConfigMap(
val initContainerConfigMap = submittedDependencyManager.buildInitContainerConfigMap(
jarsResourceIdentifier.resourceId, filesResourceIdentifier.resourceId)
resolvedJars ++= mountedDependencyManager.resolveSparkJars()
resolvedFiles ++= mountedDependencyManager.resolveSparkFiles()
resolvedJars ++= submittedDependencyManager.resolveSparkJars()
resolvedFiles ++= submittedDependencyManager.resolveSparkFiles()
nonDriverPodKubernetesResources += initContainerKubernetesSecret
nonDriverPodKubernetesResources += initContainerConfigMap
mountedDependencyManager.configurePodToMountLocalDependencies(
submittedDependencyManager.configurePodToMountLocalDependencies(
driverContainer.getName, initContainerKubernetesSecret, initContainerConfigMap, basePod)
}.getOrElse {
sparkJars.map(Utils.resolveURI).foreach { jar =>
Expand Down Expand Up @@ -186,19 +194,35 @@ private[spark] class Client(
resolvedSparkConf.get(KUBERNETES_DRIVER_OAUTH_TOKEN).foreach { _ =>
resolvedSparkConf.set(KUBERNETES_DRIVER_OAUTH_TOKEN.key, "<present_but_redacted>")
}
val remoteDependencyManager = remoteDependencyManagerProvider
.getDownloadRemoteDependencyManager(
kubernetesAppId,
resolvedJars,
resolvedFiles)
val downloadRemoteDependenciesConfigMap = remoteDependencyManager
.buildInitContainerConfigMap()
nonDriverPodKubernetesResources += downloadRemoteDependenciesConfigMap
val driverPodWithMountedAndDownloadedDeps = remoteDependencyManager
.configurePodToDownloadRemoteDependencies(
downloadRemoteDependenciesConfigMap, driverContainer.getName, driverPodWithMountedDeps)

val mountedClassPath = resolvedJars.map(Utils.resolveURI).filter { jarUri =>
val scheme = Option.apply(jarUri.getScheme).getOrElse("file")
scheme == "local" || scheme == "file"
}.map(_.getPath).mkString(File.pathSeparator)
// The resolved local classpath should *only* contain local file URIs. It consists of the
// driver's classpath (minus spark.driver.extraClassPath which was handled above) with the
// assumption that the remote dependency manager has downloaded all of the remote
// dependencies through its init-container, and thus replaces all the remote URIs with the
// local paths they were downloaded to.
val resolvedLocalClassPath = remoteDependencyManager.resolveLocalClasspath()
resolvedLocalClassPath.foreach { classPathEntry =>
require(Option(URI.create(classPathEntry).getScheme).isEmpty)
}
val resolvedDriverJavaOpts = resolvedSparkConf.getAll.map { case (confKey, confValue) =>
s"-D$confKey=$confValue"
}.mkString(" ") + driverJavaOptions.map(" " + _).getOrElse("")
val resolvedDriverPod = driverPodWithMountedDeps.editSpec()
val resolvedDriverPod = driverPodWithMountedAndDownloadedDeps.editSpec()
.editMatchingContainer(new ContainerNameEqualityPredicate(driverContainer.getName))
.addNewEnv()
.withName(ENV_MOUNTED_CLASSPATH)
.withValue(mountedClassPath)
.withValue(resolvedLocalClassPath.mkString(File.pathSeparator))
.endEnv()
.addNewEnv()
.withName(ENV_DRIVER_JAVA_OPTS)
Expand Down
Loading