Skip to content

Commit 04afcf8

Browse files
mccheahfoxish
mccheah
authored and
foxish
committed
Driver submission with mounting dependencies from the staging server (#227)
1 parent 4940eae commit 04afcf8

File tree

38 files changed

+2932
-616
lines changed

38 files changed

+2932
-616
lines changed

‎resource-managers/kubernetes/core/pom.xml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,8 @@
108108
<groupId>com.google.guava</groupId>
109109
<artifactId>guava</artifactId>
110110
</dependency>
111+
<!-- End of shaded deps. -->
112+
111113
<dependency>
112114
<groupId>org.bouncycastle</groupId>
113115
<artifactId>bcpkix-jdk15on</artifactId>
@@ -116,7 +118,11 @@
116118
<groupId>org.bouncycastle</groupId>
117119
<artifactId>bcprov-jdk15on</artifactId>
118120
</dependency>
119-
<!-- End of shaded deps. -->
121+
<dependency>
122+
<groupId>org.mockito</groupId>
123+
<artifactId>mockito-core</artifactId>
124+
<scope>test</scope>
125+
</dependency>
120126

121127
</dependencies>
122128

‎resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/v1/CompressionUtils.scala‎ renamed to ‎resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/CompressionUtils.scala‎

Lines changed: 71 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,9 @@
1414
* See the License for the specific language governing permissions and
1515
* limitations under the License.
1616
*/
17-
package org.apache.spark.deploy.kubernetes.submit.v1
17+
package org.apache.spark.deploy.kubernetes
1818

19-
import java.io.{ByteArrayInputStream, File, FileInputStream, FileOutputStream}
19+
import java.io.{ByteArrayInputStream, File, FileInputStream, FileOutputStream, InputStream, OutputStream}
2020
import java.util.zip.{GZIPInputStream, GZIPOutputStream}
2121

2222
import com.google.common.io.Files
@@ -48,40 +48,7 @@ private[spark] object CompressionUtils extends Logging {
4848
*/
4949
def createTarGzip(paths: Iterable[String]): TarGzippedData = {
5050
val compressedBytesStream = Utils.tryWithResource(new ByteBufferOutputStream()) { raw =>
51-
Utils.tryWithResource(new GZIPOutputStream(raw)) { gzipping =>
52-
Utils.tryWithResource(new TarArchiveOutputStream(
53-
gzipping,
54-
BLOCK_SIZE,
55-
RECORD_SIZE,
56-
ENCODING)) { tarStream =>
57-
val usedFileNames = mutable.HashSet.empty[String]
58-
for (path <- paths) {
59-
val file = new File(path)
60-
if (!file.isFile) {
61-
throw new IllegalArgumentException(s"Cannot add $path to tarball; either does" +
62-
s" not exist or is a directory.")
63-
}
64-
var resolvedFileName = file.getName
65-
val extension = Files.getFileExtension(file.getName)
66-
val nameWithoutExtension = Files.getNameWithoutExtension(file.getName)
67-
var deduplicationCounter = 1
68-
while (usedFileNames.contains(resolvedFileName)) {
69-
val oldResolvedFileName = resolvedFileName
70-
resolvedFileName = s"$nameWithoutExtension-$deduplicationCounter.$extension"
71-
logWarning(s"File with name $oldResolvedFileName already exists. Trying to add" +
72-
s" with file name $resolvedFileName instead.")
73-
deduplicationCounter += 1
74-
}
75-
usedFileNames += resolvedFileName
76-
val tarEntry = new TarArchiveEntry(file, resolvedFileName)
77-
tarStream.putArchiveEntry(tarEntry)
78-
Utils.tryWithResource(new FileInputStream(file)) { fileInput =>
79-
IOUtils.copy(fileInput, tarStream)
80-
}
81-
tarStream.closeArchiveEntry()
82-
}
83-
}
84-
}
51+
writeTarGzipToStream(raw, paths)
8552
raw
8653
}
8754
val compressedAsBase64 = Base64.encodeBase64String(compressedBytesStream.toByteBuffer.array)
@@ -93,6 +60,44 @@ private[spark] object CompressionUtils extends Logging {
9360
)
9461
}
9562

63+
def writeTarGzipToStream(outputStream: OutputStream, paths: Iterable[String]): Unit = {
64+
Utils.tryWithResource(new GZIPOutputStream(outputStream)) { gzipping =>
65+
Utils.tryWithResource(new TarArchiveOutputStream(
66+
gzipping,
67+
BLOCK_SIZE,
68+
RECORD_SIZE,
69+
ENCODING)) { tarStream =>
70+
val usedFileNames = mutable.HashSet.empty[String]
71+
for (path <- paths) {
72+
val file = new File(path)
73+
if (!file.isFile) {
74+
throw new IllegalArgumentException(s"Cannot add $path to tarball; either does" +
75+
s" not exist or is a directory.")
76+
}
77+
var resolvedFileName = file.getName
78+
val extension = Files.getFileExtension(file.getName)
79+
val nameWithoutExtension = Files.getNameWithoutExtension(file.getName)
80+
var deduplicationCounter = 1
81+
while (usedFileNames.contains(resolvedFileName)) {
82+
val oldResolvedFileName = resolvedFileName
83+
resolvedFileName = s"$nameWithoutExtension-$deduplicationCounter.$extension"
84+
logWarning(s"File with name $oldResolvedFileName already exists. Trying to add" +
85+
s" with file name $resolvedFileName instead.")
86+
deduplicationCounter += 1
87+
}
88+
usedFileNames += resolvedFileName
89+
val tarEntry = new TarArchiveEntry(resolvedFileName)
90+
tarEntry.setSize(file.length());
91+
tarStream.putArchiveEntry(tarEntry)
92+
Utils.tryWithResource(new FileInputStream(file)) { fileInput =>
93+
IOUtils.copy(fileInput, tarStream)
94+
}
95+
tarStream.closeArchiveEntry()
96+
}
97+
}
98+
}
99+
}
100+
96101
/**
97102
* Decompresses the provided tar archive to a directory.
98103
* @param compressedData In-memory representation of the compressed data, ideally created via
@@ -104,7 +109,6 @@ private[spark] object CompressionUtils extends Logging {
104109
def unpackAndWriteCompressedFiles(
105110
compressedData: TarGzippedData,
106111
rootOutputDir: File): Seq[String] = {
107-
val paths = mutable.Buffer.empty[String]
108112
val compressedBytes = Base64.decodeBase64(compressedData.dataBase64)
109113
if (!rootOutputDir.exists) {
110114
if (!rootOutputDir.mkdirs) {
@@ -116,24 +120,39 @@ private[spark] object CompressionUtils extends Logging {
116120
s"${rootOutputDir.getAbsolutePath} exists and is not a directory.")
117121
}
118122
Utils.tryWithResource(new ByteArrayInputStream(compressedBytes)) { compressedBytesStream =>
119-
Utils.tryWithResource(new GZIPInputStream(compressedBytesStream)) { gzipped =>
120-
Utils.tryWithResource(new TarArchiveInputStream(
121-
gzipped,
122-
compressedData.blockSize,
123-
compressedData.recordSize,
124-
compressedData.encoding)) { tarInputStream =>
125-
var nextTarEntry = tarInputStream.getNextTarEntry
126-
while (nextTarEntry != null) {
127-
val outputFile = new File(rootOutputDir, nextTarEntry.getName)
128-
Utils.tryWithResource(new FileOutputStream(outputFile)) { fileOutputStream =>
129-
IOUtils.copy(tarInputStream, fileOutputStream)
130-
}
131-
paths += outputFile.getAbsolutePath
132-
nextTarEntry = tarInputStream.getNextTarEntry
123+
unpackTarStreamToDirectory(
124+
compressedBytesStream,
125+
rootOutputDir,
126+
compressedData.blockSize,
127+
compressedData.recordSize,
128+
compressedData.encoding)
129+
}
130+
}
131+
132+
def unpackTarStreamToDirectory(
133+
inputStream: InputStream,
134+
outputDir: File,
135+
blockSize: Int = BLOCK_SIZE,
136+
recordSize: Int = RECORD_SIZE,
137+
encoding: String = ENCODING): Seq[String] = {
138+
val paths = mutable.Buffer.empty[String]
139+
Utils.tryWithResource(new GZIPInputStream(inputStream)) { gzipped =>
140+
Utils.tryWithResource(new TarArchiveInputStream(
141+
gzipped,
142+
blockSize,
143+
recordSize,
144+
encoding)) { tarInputStream =>
145+
var nextTarEntry = tarInputStream.getNextTarEntry
146+
while (nextTarEntry != null) {
147+
val outputFile = new File(outputDir, nextTarEntry.getName)
148+
Utils.tryWithResource(new FileOutputStream(outputFile)) { fileOutputStream =>
149+
IOUtils.copy(tarInputStream, fileOutputStream)
133150
}
151+
paths += outputFile.getAbsolutePath
152+
nextTarEntry = tarInputStream.getNextTarEntry
134153
}
135154
}
136155
}
137-
paths.toSeq
156+
paths
138157
}
139158
}

‎resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala‎

Lines changed: 106 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,13 @@ package org.apache.spark.deploy.kubernetes
1919
import java.util.concurrent.TimeUnit
2020

2121
import org.apache.spark.{SPARK_VERSION => sparkVersion}
22+
import org.apache.spark.deploy.kubernetes.constants._
2223
import org.apache.spark.deploy.kubernetes.submit.v1.NodePortUrisDriverServiceManager
24+
import org.apache.spark.internal.Logging
2325
import org.apache.spark.internal.config.ConfigBuilder
2426
import org.apache.spark.network.util.ByteUnit
2527

26-
package object config {
28+
package object config extends Logging {
2729

2830
private[spark] val KUBERNETES_NAMESPACE =
2931
ConfigBuilder("spark.kubernetes.namespace")
@@ -321,4 +323,107 @@ package object config {
321323
.doc("File containing the key password for the Kubernetes dependency server.")
322324
.stringConf
323325
.createOptional
326+
327+
private[spark] val RESOURCE_STAGING_SERVER_SSL_ENABLED =
328+
ConfigBuilder("spark.ssl.kubernetes.resourceStagingServer.enabled")
329+
.doc("Whether or not to use SSL when communicating with the dependency server.")
330+
.booleanConf
331+
.createOptional
332+
private[spark] val RESOURCE_STAGING_SERVER_TRUSTSTORE_FILE =
333+
ConfigBuilder("spark.ssl.kubernetes.resourceStagingServer.trustStore")
334+
.doc("File containing the trustStore to communicate with the Kubernetes dependency server.")
335+
.stringConf
336+
.createOptional
337+
private[spark] val RESOURCE_STAGING_SERVER_TRUSTSTORE_PASSWORD =
338+
ConfigBuilder("spark.ssl.kubernetes.resourceStagingServer.trustStorePassword")
339+
.doc("Password for the trustStore for talking to the dependency server.")
340+
.stringConf
341+
.createOptional
342+
private[spark] val RESOURCE_STAGING_SERVER_TRUSTSTORE_TYPE =
343+
ConfigBuilder("spark.ssl.kubernetes.resourceStagingServer.trustStoreType")
344+
.doc("Type of trustStore for communicating with the dependency server.")
345+
.stringConf
346+
.createOptional
347+
348+
// Driver and Init-Container parameters for submission v2
349+
private[spark] val RESOURCE_STAGING_SERVER_URI =
350+
ConfigBuilder("spark.kubernetes.resourceStagingServer.uri")
351+
.doc("Base URI for the Spark resource staging server")
352+
.stringConf
353+
.createOptional
354+
355+
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_RESOURCE_IDENTIFIER =
356+
ConfigBuilder("spark.kubernetes.driver.initcontainer.downloadJarsResourceIdentifier")
357+
.doc("Identifier for the jars tarball that was uploaded to the staging service.")
358+
.internal()
359+
.stringConf
360+
.createOptional
361+
362+
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_SECRET_LOCATION =
363+
ConfigBuilder("spark.kubernetes.driver.initcontainer.downloadJarsSecretLocation")
364+
.doc("Location of the application secret to use when the init-container contacts the" +
365+
" resource staging server to download jars.")
366+
.internal()
367+
.stringConf
368+
.createWithDefault(INIT_CONTAINER_DOWNLOAD_JARS_SECRET_PATH)
369+
370+
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_RESOURCE_IDENTIFIER =
371+
ConfigBuilder("spark.kubernetes.driver.initcontainer.downloadFilesResourceIdentifier")
372+
.doc("Identifier for the files tarball that was uploaded to the staging service.")
373+
.internal()
374+
.stringConf
375+
.createOptional
376+
377+
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_SECRET_LOCATION =
378+
ConfigBuilder("spark.kubernetes.driver.initcontainer.downloadFilesSecretLocation")
379+
.doc("Location of the application secret to use when the init-container contacts the" +
380+
" resource staging server to download files.")
381+
.internal()
382+
.stringConf
383+
.createWithDefault(INIT_CONTAINER_DOWNLOAD_FILES_SECRET_PATH)
384+
385+
private[spark] val INIT_CONTAINER_DOCKER_IMAGE =
386+
ConfigBuilder("spark.kubernetes.driver.initcontainer.docker.image")
387+
.doc("Image for the driver's init-container that downloads mounted dependencies.")
388+
.stringConf
389+
.createWithDefault(s"spark-driver-init:$sparkVersion")
390+
391+
private[spark] val DRIVER_LOCAL_JARS_DOWNLOAD_LOCATION =
392+
ConfigBuilder("spark.kubernetes.driver.mountdependencies.jarsDownloadDir")
393+
.doc("Location to download local jars to in the driver. When using spark-submit, this" +
394+
" directory must be empty and will be mounted as an empty directory volume on the" +
395+
" driver pod.")
396+
.stringConf
397+
.createWithDefault("/var/spark-data/spark-local-jars")
398+
399+
private[spark] val DRIVER_LOCAL_FILES_DOWNLOAD_LOCATION =
400+
ConfigBuilder("spark.kubernetes.driver.mountdependencies.filesDownloadDir")
401+
.doc("Location to download local files to in the driver. When using spark-submit, this" +
402+
" directory must be empty and will be mounted as an empty directory volume on the" +
403+
" driver pod.")
404+
.stringConf
405+
.createWithDefault("/var/spark-data/spark-local-files")
406+
407+
private[spark] val DRIVER_MOUNT_DEPENDENCIES_INIT_TIMEOUT =
408+
ConfigBuilder("spark.kubernetes.mountdependencies.mountTimeout")
409+
.doc("Timeout before aborting the attempt to download and unpack local dependencies from" +
410+
" the dependency staging server when initializing the driver pod.")
411+
.timeConf(TimeUnit.MINUTES)
412+
.createWithDefault(5)
413+
414+
private[spark] def resolveK8sMaster(rawMasterString: String): String = {
415+
if (!rawMasterString.startsWith("k8s://")) {
416+
throw new IllegalArgumentException("Master URL should start with k8s:// in Kubernetes mode.")
417+
}
418+
val masterWithoutK8sPrefix = rawMasterString.replaceFirst("k8s://", "")
419+
if (masterWithoutK8sPrefix.startsWith("http://")
420+
|| masterWithoutK8sPrefix.startsWith("https://")) {
421+
masterWithoutK8sPrefix
422+
} else {
423+
val resolvedURL = s"https://$masterWithoutK8sPrefix"
424+
logDebug(s"No scheme specified for kubernetes master URL, so defaulting to https. Resolved" +
425+
s" URL is $resolvedURL")
426+
resolvedURL
427+
}
428+
}
324429
}

‎resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala‎

Lines changed: 36 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,9 @@ package object constants {
3030
private[spark] val SUBMISSION_APP_SECRET_PREFIX = "spark-submission-server-secret"
3131
private[spark] val SUBMISSION_APP_SECRET_VOLUME_NAME = "spark-submission-secret-volume"
3232
private[spark] val SUBMISSION_SSL_KEY_PASSWORD_SECRET_NAME =
33-
"spark-submission-server-key-password"
33+
"spark-submission-server-key-password"
3434
private[spark] val SUBMISSION_SSL_KEYSTORE_PASSWORD_SECRET_NAME =
35-
"spark-submission-server-keystore-password"
35+
"spark-submission-server-keystore-password"
3636
private[spark] val SUBMISSION_SSL_KEYSTORE_SECRET_NAME = "spark-submission-server-keystore"
3737
private[spark] val SUBMISSION_SSL_SECRETS_PREFIX = "spark-submission-server-ssl"
3838
private[spark] val SUBMISSION_SSL_SECRETS_VOLUME_NAME = "spark-submission-server-ssl-secrets"
@@ -55,9 +55,9 @@ package object constants {
5555
private[spark] val ENV_SUBMISSION_SERVER_PORT = "SPARK_SUBMISSION_SERVER_PORT"
5656
private[spark] val ENV_SUBMISSION_KEYSTORE_FILE = "SPARK_SUBMISSION_KEYSTORE_FILE"
5757
private[spark] val ENV_SUBMISSION_KEYSTORE_PASSWORD_FILE =
58-
"SPARK_SUBMISSION_KEYSTORE_PASSWORD_FILE"
58+
"SPARK_SUBMISSION_KEYSTORE_PASSWORD_FILE"
5959
private[spark] val ENV_SUBMISSION_KEYSTORE_KEY_PASSWORD_FILE =
60-
"SPARK_SUBMISSION_KEYSTORE_KEY_PASSWORD_FILE"
60+
"SPARK_SUBMISSION_KEYSTORE_KEY_PASSWORD_FILE"
6161
private[spark] val ENV_SUBMISSION_KEYSTORE_TYPE = "SPARK_SUBMISSION_KEYSTORE_TYPE"
6262
private[spark] val ENV_SUBMISSION_KEY_PEM_FILE = "SPARK_SUBMISSION_KEY_PEM_FILE"
6363
private[spark] val ENV_SUBMISSION_CERT_PEM_FILE = "SPARK_SUBMISSION_CERT_PEM_FILE"
@@ -70,17 +70,47 @@ package object constants {
7070
private[spark] val ENV_EXECUTOR_ID = "SPARK_EXECUTOR_ID"
7171
private[spark] val ENV_EXECUTOR_POD_IP = "SPARK_EXECUTOR_POD_IP"
7272
private[spark] val ENV_DRIVER_MEMORY = "SPARK_DRIVER_MEMORY"
73+
private[spark] val ENV_UPLOADED_JARS_DIR = "SPARK_UPLOADED_JARS_DIR"
74+
private[spark] val ENV_SUBMIT_EXTRA_CLASSPATH = "SPARK_SUBMIT_EXTRA_CLASSPATH"
75+
private[spark] val ENV_MOUNTED_CLASSPATH = "SPARK_MOUNTED_CLASSPATH"
76+
private[spark] val ENV_DRIVER_MAIN_CLASS = "SPARK_DRIVER_CLASS"
77+
private[spark] val ENV_DRIVER_ARGS = "SPARK_DRIVER_ARGS"
78+
private[spark] val ENV_DRIVER_JAVA_OPTS = "SPARK_DRIVER_JAVA_OPTS"
7379

7480
// Annotation keys
7581
private[spark] val ANNOTATION_PROVIDE_EXTERNAL_URI =
76-
"spark-job.alpha.apache.org/provideExternalUri"
82+
"spark-job.alpha.apache.org/provideExternalUri"
7783
private[spark] val ANNOTATION_RESOLVED_EXTERNAL_URI =
78-
"spark-job.alpha.apache.org/resolvedExternalUri"
84+
"spark-job.alpha.apache.org/resolvedExternalUri"
7985

8086
// Miscellaneous
8187
private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
8288
private[spark] val DRIVER_SUBMIT_SSL_NAMESPACE = "kubernetes.driversubmitserver"
8389
private[spark] val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
8490
private[spark] val MEMORY_OVERHEAD_FACTOR = 0.10
8591
private[spark] val MEMORY_OVERHEAD_MIN = 384L
92+
93+
// V2 submission init container
94+
private[spark] val INIT_CONTAINER_ANNOTATION = "pod.beta.kubernetes.io/init-containers"
95+
private[spark] val INIT_CONTAINER_SECRETS_VOLUME_NAME = "dependency-secret"
96+
private[spark] val INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH = "/mnt/secrets/spark-init"
97+
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_SECRET_KEY = "downloadJarsSecret"
98+
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_SECRET_KEY = "downloadFilesSecret"
99+
private[spark] val INIT_CONTAINER_TRUSTSTORE_SECRET_KEY = "trustStore"
100+
private[spark] val INIT_CONTAINER_DOWNLOAD_JARS_SECRET_PATH =
101+
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_DOWNLOAD_JARS_SECRET_KEY"
102+
private[spark] val INIT_CONTAINER_DOWNLOAD_FILES_SECRET_PATH =
103+
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_DOWNLOAD_FILES_SECRET_KEY"
104+
private[spark] val INIT_CONTAINER_TRUSTSTORE_PATH =
105+
s"$INIT_CONTAINER_SECRETS_VOLUME_MOUNT_PATH/$INIT_CONTAINER_TRUSTSTORE_SECRET_KEY"
106+
private[spark] val INIT_CONTAINER_DOWNLOAD_CREDENTIALS_PATH =
107+
"/mnt/secrets/kubernetes-credentials"
108+
private[spark] val INIT_CONTAINER_CONFIG_MAP_KEY = "init-driver"
109+
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_VOLUME = "init-container-properties"
110+
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_MOUNT_PATH = "/etc/spark-init/"
111+
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_NAME = "init-driver.properties"
112+
private[spark] val INIT_CONTAINER_PROPERTIES_FILE_PATH =
113+
s"$INIT_CONTAINER_PROPERTIES_FILE_MOUNT_PATH/$INIT_CONTAINER_PROPERTIES_FILE_NAME"
114+
private[spark] val DOWNLOAD_JARS_VOLUME_NAME = "download-jars"
115+
private[spark] val DOWNLOAD_FILES_VOLUME_NAME = "download-files"
86116
}

0 commit comments

Comments
 (0)