Mount a hadoop secret in the executor pod

Author: kimoonkim

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/HadoopSecretUtil.scala

@@ -0,0 +1,55 @@
+/*
+ * 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.submit
+
+import io.fabric8.kubernetes.api.model.{Container, ContainerBuilder, Pod, PodBuilder}
+
+import org.apache.spark.deploy.kubernetes.constants._
+
+object HadoopSecretUtil {
+
+  def configurePod(secretNameOption: Option[String], pod: Pod) : Pod = {
+    secretNameOption.map { secret =>
+      new PodBuilder(pod)
+        .editOrNewSpec()
+        .addNewVolume()
+        .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
+        .withNewSecret()
+        .withSecretName(secret)
+        .endSecret()
+        .endVolume()
+        .endSpec()
+        .build()
+    }.getOrElse(pod)
+  }
+
+  def configureContainer(secretNameOption: Option[String],
+                         containerSpec: Container) : Container = {
+    secretNameOption.map { secret =>
+      new ContainerBuilder(containerSpec)
+        .addNewVolumeMount()
+        .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
+        .withMountPath(SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR)
+        .endVolumeMount()
+        .addNewEnv()
+        .withName(ENV_HADOOP_TOKEN_FILE_LOCATION)
+        .withValue(SPARK_APP_HADOOP_TOKEN_FILE_PATH)
+        .endEnv()
+        .build()
+    }.getOrElse(containerSpec)
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/DriverHadoopCredentialsStep.scala

@@ -16,46 +16,24 @@
  */
 package org.apache.spark.deploy.kubernetes.submit.submitsteps
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder}
-
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.kubernetes.config._
-import org.apache.spark.deploy.kubernetes.constants._
-
+import org.apache.spark.deploy.kubernetes.submit.HadoopSecretUtil
 
-class DriverHadoopCredentialsStep(submissionSparkConf: SparkConf) extends DriverConfigurationStep {
+private[spark] class DriverHadoopCredentialsStep(submissionSparkConf: SparkConf)
+  extends DriverConfigurationStep {
 
   private val maybeMountedHadoopSecret = submissionSparkConf.getOption(MOUNTED_HADOOP_SECRET_CONF)
 
   override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
-    val driverPodWithMountedHadoopTokens = maybeMountedHadoopSecret.map { secret =>
-      new PodBuilder(driverSpec.driverPod)
-        .editOrNewSpec()
-          .addNewVolume()
-            .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
-            .withNewSecret()
-              .withSecretName(secret)
-            .endSecret()
-          .endVolume()
-        .endSpec()
-        .build()
-    }.getOrElse(driverSpec.driverPod)
-    val driverContainerWithMountedSecretVolume = maybeMountedHadoopSecret.map { secret =>
-      new ContainerBuilder(driverSpec.driverContainer)
-          .addNewVolumeMount()
-            .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
-            .withMountPath(SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR)
-          .endVolumeMount()
-          .addNewEnv()
-            .withName(ENV_HADOOP_TOKEN_FILE_LOCATION)
-            .withValue(SPARK_APP_HADOOP_TOKEN_FILE_PATH)
-          .endEnv()
-        .build()
-    }.getOrElse(driverSpec.driverContainer)
+    val podWithMountedHadoopToken = HadoopSecretUtil.configurePod(maybeMountedHadoopSecret,
+      driverSpec.driverPod)
+    val containerWithMountedHadoopToken = HadoopSecretUtil.configureContainer(
+      maybeMountedHadoopSecret, driverSpec.driverContainer)
     driverSpec.copy(
-      driverPod = driverPodWithMountedHadoopTokens,
+      driverPod = podWithMountedHadoopToken,
       otherKubernetesResources = driverSpec.otherKubernetesResources,
       driverSparkConf = driverSpec.driverSparkConf,
-      driverContainer = driverContainerWithMountedSecretVolume)
+      driverContainer = containerWithMountedHadoopToken)
   }
 }

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala

@@ -37,7 +37,7 @@ import org.apache.spark.{SparkContext, SparkEnv, SparkException}
 import org.apache.spark.deploy.kubernetes.{ConfigurationUtils, InitContainerResourceStagingServerSecretPlugin, PodWithDetachedInitContainer, SparkPodInitContainerBootstrap}
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.constants._
-import org.apache.spark.deploy.kubernetes.submit.InitContainerUtil
+import org.apache.spark.deploy.kubernetes.submit.{HadoopSecretUtil, InitContainerUtil}
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.shuffle.kubernetes.KubernetesExternalShuffleClient
 import org.apache.spark.rpc.{RpcAddress, RpcCallContext, RpcEndpointAddress, RpcEnv}
@@ -130,6 +130,8 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private implicit val requestExecutorContext = ExecutionContext.fromExecutorService(
     ThreadUtils.newDaemonCachedThreadPool("kubernetes-executor-requests"))
 
+  private val maybeMountedHadoopSecret = conf.getOption(MOUNTED_HADOOP_SECRET_CONF)
+
   private val driverPod = try {
     kubernetesClient.pods().inNamespace(kubernetesNamespace).
       withName(kubernetesDriverPodName).get()
@@ -582,9 +584,14 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
     val executorPodWithNodeAffinity = addNodeAffinityAnnotationIfUseful(
         executorPodWithInitContainer, nodeToLocalTaskCount)
-    val resolvedExecutorPod = new PodBuilder(executorPodWithNodeAffinity)
+    val executorPodWithMountedHadoopToken = HadoopSecretUtil.configurePod(maybeMountedHadoopSecret,
+      executorPodWithNodeAffinity)
+    val containerWithMountedHadoopToken = HadoopSecretUtil.configureContainer(
+      maybeMountedHadoopSecret, initBootstrappedExecutorContainer)
+
+    val resolvedExecutorPod = new PodBuilder(executorPodWithMountedHadoopToken)
       .editSpec()
-        .addToContainers(initBootstrappedExecutorContainer)
+        .addToContainers(containerWithMountedHadoopToken)
         .endSpec()
       .build()
     try {