/
SecretsTestsSuite.scala
128 lines (116 loc) · 4.95 KB
/
SecretsTestsSuite.scala
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
/*
* 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.k8s.integrationtest
import java.util.Locale
import scala.collection.JavaConverters._
import io.fabric8.kubernetes.api.model.{Pod, SecretBuilder}
import org.apache.commons.codec.binary.Base64
import org.apache.commons.io.output.ByteArrayOutputStream
import org.scalatest.concurrent.Eventually
import org.apache.spark.deploy.k8s.integrationtest.KubernetesSuite._
private[spark] trait SecretsTestsSuite { k8sSuite: KubernetesSuite =>
import SecretsTestsSuite._
private def createTestSecret(): Unit = {
val sb = new SecretBuilder()
sb.withNewMetadata()
.withName(ENV_SECRET_NAME)
.endMetadata()
val secUsername = Base64.encodeBase64String(ENV_SECRET_VALUE_1.getBytes())
val secPassword = Base64.encodeBase64String(ENV_SECRET_VALUE_2.getBytes())
val envSecretData = Map(ENV_SECRET_KEY_1 -> secUsername, ENV_SECRET_KEY_2 -> secPassword)
sb.addToData(envSecretData.asJava)
val envSecret = sb.build()
val sec = kubernetesTestComponents
.kubernetesClient
.secrets()
.createOrReplace(envSecret)
}
private def deleteTestSecret(): Unit = {
kubernetesTestComponents
.kubernetesClient
.secrets()
.withName(ENV_SECRET_NAME)
.delete()
}
test("Run SparkPi with env and mount secrets.", k8sTestTag) {
createTestSecret()
sparkAppConf
.set(s"spark.kubernetes.driver.secrets.$ENV_SECRET_NAME", SECRET_MOUNT_PATH)
.set(
s"spark.kubernetes.driver.secretKeyRef.${ENV_SECRET_KEY_1_CAP}",
s"$ENV_SECRET_NAME:${ENV_SECRET_KEY_1}")
.set(
s"spark.kubernetes.driver.secretKeyRef.${ENV_SECRET_KEY_2_CAP}",
s"$ENV_SECRET_NAME:${ENV_SECRET_KEY_2}")
.set(s"spark.kubernetes.executor.secrets.$ENV_SECRET_NAME", SECRET_MOUNT_PATH)
.set(s"spark.kubernetes.executor.secretKeyRef.${ENV_SECRET_KEY_1_CAP}",
s"${ENV_SECRET_NAME}:$ENV_SECRET_KEY_1")
.set(s"spark.kubernetes.executor.secretKeyRef.${ENV_SECRET_KEY_2_CAP}",
s"${ENV_SECRET_NAME}:$ENV_SECRET_KEY_2")
try {
runSparkPiAndVerifyCompletion(
driverPodChecker = (driverPod: Pod) => {
doBasicDriverPodCheck(driverPod)
checkSecrets(driverPod)
},
executorPodChecker = (executorPod: Pod) => {
doBasicExecutorPodCheck(executorPod)
checkSecrets(executorPod)
},
appArgs = Array("1000") // give it enough time for all execs to be visible
)
} finally {
// make sure this always run
deleteTestSecret()
}
}
private def checkSecrets(pod: Pod): Unit = {
logDebug(s"Checking secrets for ${pod}")
// Wait for the pod to become ready & have secrets provisioned
implicit val podName: String = pod.getMetadata.getName
implicit val components: KubernetesTestComponents = kubernetesTestComponents
val env = Eventually.eventually(TIMEOUT, INTERVAL) {
logDebug(s"Checking env of ${pod.getMetadata().getName()} ....")
val env = Utils.executeCommand("env")
assert(!env.isEmpty)
env
}
env.toString should include (s"${ENV_SECRET_KEY_1_CAP}=$ENV_SECRET_VALUE_1")
env.toString should include (s"${ENV_SECRET_KEY_2_CAP}=$ENV_SECRET_VALUE_2")
// Make sure our secret files are mounted correctly
val files = Utils.executeCommand("ls", s"$SECRET_MOUNT_PATH")
files should include (ENV_SECRET_KEY_1)
files should include (ENV_SECRET_KEY_2)
// Validate the contents
val fileUsernameContents = Utils
.executeCommand("cat", s"$SECRET_MOUNT_PATH/$ENV_SECRET_KEY_1")
fileUsernameContents.toString.trim should equal(ENV_SECRET_VALUE_1)
val filePasswordContents = Utils
.executeCommand("cat", s"$SECRET_MOUNT_PATH/$ENV_SECRET_KEY_2")
filePasswordContents.toString.trim should equal(ENV_SECRET_VALUE_2)
}
}
private[spark] object SecretsTestsSuite {
val ENV_SECRET_NAME = "mysecret"
val SECRET_MOUNT_PATH = "/etc/secret"
val ENV_SECRET_KEY_1 = "username"
val ENV_SECRET_KEY_2 = "password"
val ENV_SECRET_KEY_1_CAP = ENV_SECRET_KEY_1.toUpperCase(Locale.ROOT)
val ENV_SECRET_KEY_2_CAP = ENV_SECRET_KEY_2.toUpperCase(Locale.ROOT)
val ENV_SECRET_VALUE_1 = "secretusername"
val ENV_SECRET_VALUE_2 = "secretpassword"
}