forked from apache/spark
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Download remotely-located resources on driver and executor startup vi…
…a init-container (apache#251) * Download remotely-located resources on driver startup. Use init-container in executors. * FIx owner reference slightly * Clean up config * Don't rely too heavily on conventions that can change * Fix flaky test * Tidy up file resolver * Whitespace arrangement * Indentation change * Fix more indentation * Consolidate init container component providers * Minor method signature and comment changes * Rename class for consistency * Resolve conflicts * Fix flaky test * Add some tests and some refactoring. * Make naming consistent for Staged -> Submitted * Add unit test for the submission client. * Refine expectations * Rename variables and fix typos * Address more comments. Remove redundant SingleKeyConfigMap. * Minor test adjustments. * add another test * Fix conflicts.
- Loading branch information
Showing
46 changed files
with
2,620 additions
and
1,233 deletions.
There are no files selected for viewing
62 changes: 62 additions & 0 deletions
62
...a/org/apache/spark/deploy/kubernetes/InitContainerResourceStagingServerSecretPlugin.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| /* | ||
| * 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.kubernetes | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, Secret} | ||
|
|
||
| import org.apache.spark.deploy.kubernetes.constants._ | ||
|
|
||
| private[spark] trait InitContainerResourceStagingServerSecretPlugin { | ||
|
|
||
| /** | ||
| * Configure the init-container to mount the secret files that allow it to retrieve dependencies | ||
| * from a resource staging server. | ||
| */ | ||
| def mountResourceStagingServerSecretIntoInitContainer( | ||
| initContainer: ContainerBuilder): ContainerBuilder | ||
|
|
||
| /** | ||
| * Configure the pod to attach a Secret volume which hosts secret files allowing the | ||
| * init-container to retrieve dependencies from the resource staging server. | ||
| */ | ||
| def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder | ||
| } | ||
|
|
||
| private[spark] class InitContainerResourceStagingServerSecretPluginImpl( | ||
| initContainerSecretName: String, | ||
| initContainerSecretMountPath: String) | ||
| extends InitContainerResourceStagingServerSecretPlugin { | ||
|
|
||
| override def mountResourceStagingServerSecretIntoInitContainer( | ||
| initContainer: ContainerBuilder): ContainerBuilder = { | ||
| initContainer.addNewVolumeMount() | ||
| .withName(INIT_CONTAINER_SECRET_VOLUME_NAME) | ||
| .withMountPath(initContainerSecretMountPath) | ||
| .endVolumeMount() | ||
| } | ||
|
|
||
| override def addResourceStagingServerSecretVolumeToPod(basePod: PodBuilder): PodBuilder = { | ||
| basePod.editSpec() | ||
| .addNewVolume() | ||
| .withName(INIT_CONTAINER_SECRET_VOLUME_NAME) | ||
| .withNewSecret() | ||
| .withSecretName(initContainerSecretName) | ||
| .endSecret() | ||
| .endVolume() | ||
| .endSpec() | ||
| } | ||
| } |
103 changes: 103 additions & 0 deletions
103
...re/src/main/scala/org/apache/spark/deploy/kubernetes/SparkPodInitContainerBootstrap.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| /* | ||
| * 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.kubernetes | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{ContainerBuilder, EmptyDirVolumeSource, PodBuilder, VolumeMount, VolumeMountBuilder} | ||
|
|
||
| import org.apache.spark.deploy.kubernetes.constants._ | ||
| import org.apache.spark.deploy.kubernetes.submit.v2.{ContainerNameEqualityPredicate, InitContainerUtil} | ||
|
|
||
| private[spark] trait SparkPodInitContainerBootstrap { | ||
| /** | ||
| * Bootstraps an init-container that downloads dependencies to be used by a main container. | ||
| * Note that this primarily assumes that the init-container's configuration is being provided | ||
| * by a ConfigMap that was installed by some other component; that is, the implementation | ||
| * here makes no assumptions about how the init-container is specifically configured. For | ||
| * example, this class is unaware if the init-container is fetching remote dependencies or if | ||
| * it is fetching dependencies from a resource staging server. | ||
| */ | ||
| def bootstrapInitContainerAndVolumes( | ||
| mainContainerName: String, originalPodSpec: PodBuilder): PodBuilder | ||
| } | ||
|
|
||
| private[spark] class SparkPodInitContainerBootstrapImpl( | ||
| initContainerImage: String, | ||
| jarsDownloadPath: String, | ||
| filesDownloadPath: String, | ||
| downloadTimeoutMinutes: Long, | ||
| initContainerConfigMapName: String, | ||
| initContainerConfigMapKey: String, | ||
| resourceStagingServerSecretPlugin: Option[InitContainerResourceStagingServerSecretPlugin]) | ||
| extends SparkPodInitContainerBootstrap { | ||
|
|
||
| override def bootstrapInitContainerAndVolumes( | ||
| mainContainerName: String, | ||
| originalPodSpec: PodBuilder): PodBuilder = { | ||
| val sharedVolumeMounts = Seq[VolumeMount]( | ||
| new VolumeMountBuilder() | ||
| .withName(INIT_CONTAINER_DOWNLOAD_JARS_VOLUME_NAME) | ||
| .withMountPath(jarsDownloadPath) | ||
| .build(), | ||
| new VolumeMountBuilder() | ||
| .withName(INIT_CONTAINER_DOWNLOAD_FILES_VOLUME_NAME) | ||
| .withMountPath(filesDownloadPath) | ||
| .build()) | ||
|
|
||
| val initContainer = new ContainerBuilder() | ||
| .withName(s"spark-init") | ||
| .withImage(initContainerImage) | ||
| .withImagePullPolicy("IfNotPresent") | ||
| .addNewVolumeMount() | ||
| .withName(INIT_CONTAINER_PROPERTIES_FILE_VOLUME) | ||
| .withMountPath(INIT_CONTAINER_PROPERTIES_FILE_DIR) | ||
| .endVolumeMount() | ||
| .addToVolumeMounts(sharedVolumeMounts: _*) | ||
| .addToArgs(INIT_CONTAINER_PROPERTIES_FILE_PATH) | ||
| val resolvedInitContainer = resourceStagingServerSecretPlugin.map { plugin => | ||
| plugin.mountResourceStagingServerSecretIntoInitContainer(initContainer) | ||
| }.getOrElse(initContainer).build() | ||
| val podWithBasicVolumes = InitContainerUtil.appendInitContainer( | ||
| originalPodSpec, resolvedInitContainer) | ||
| .editSpec() | ||
| .addNewVolume() | ||
| .withName(INIT_CONTAINER_PROPERTIES_FILE_VOLUME) | ||
| .withNewConfigMap() | ||
| .withName(initContainerConfigMapName) | ||
| .addNewItem() | ||
| .withKey(initContainerConfigMapKey) | ||
| .withPath(INIT_CONTAINER_PROPERTIES_FILE_NAME) | ||
| .endItem() | ||
| .endConfigMap() | ||
| .endVolume() | ||
| .addNewVolume() | ||
| .withName(INIT_CONTAINER_DOWNLOAD_JARS_VOLUME_NAME) | ||
| .withEmptyDir(new EmptyDirVolumeSource()) | ||
| .endVolume() | ||
| .addNewVolume() | ||
| .withName(INIT_CONTAINER_DOWNLOAD_FILES_VOLUME_NAME) | ||
| .withEmptyDir(new EmptyDirVolumeSource()) | ||
| .endVolume() | ||
| .editMatchingContainer(new ContainerNameEqualityPredicate(mainContainerName)) | ||
| .addToVolumeMounts(sharedVolumeMounts: _*) | ||
| .endContainer() | ||
| .endSpec() | ||
| resourceStagingServerSecretPlugin.map { plugin => | ||
| plugin.addResourceStagingServerSecretVolumeToPod(podWithBasicVolumes) | ||
| }.getOrElse(podWithBasicVolumes) | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.