Integration Stages 1,2 and 3

Author: ifilonenko

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

@@ -0,0 +1,72 @@
+/*
+ * 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}
+
+import org.apache.spark.deploy.kubernetes.constants._
+import org.apache.spark.internal.Logging
+
+
+ /**
+  * This is separated out from the HadoopConf steps API because this component can be reused to
+  * mounted the DT secret for executors as well.
+  */
+private[spark] trait KerberosTokenBootstrapConf {
+  /**
+    * Bootstraps a main container with the Secret mounted as volumes and an ENV variable
+    * pointing to the mounted file containing the DT for Secure HDFS interaction
+    */
+  def bootstrapMainContainerAndVolumes(
+  originalPodWithMainContainer: PodWithMainContainer)
+  : PodWithMainContainer
+}
+
+private[spark] class KerberosTokenConfBootstrapImpl(
+  secretName: String,
+  secretLabel: String) extends KerberosTokenBootstrapConf with Logging{
+
+  override def bootstrapMainContainerAndVolumes(
+  originalPodWithMainContainer: PodWithMainContainer)
+  : PodWithMainContainer = {
+    logInfo("Mounting HDFS DT from Secret for Secure HDFS")
+    val dtMountedPod = new PodBuilder(originalPodWithMainContainer.pod)
+      .editOrNewSpec()
+        .addNewVolume()
+          .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
+          .withNewSecret()
+            .withSecretName(secretName)
+            .endSecret()
+          .endVolume()
+        .endSpec()
+      .build()
+    val mainContainerWithMountedKerberos = new ContainerBuilder(
+      originalPodWithMainContainer.mainContainer)
+      .addNewVolumeMount()
+        .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
+        .withMountPath(SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR)
+        .endVolumeMount()
+      .addNewEnv()
+        .withName(ENV_HADOOP_TOKEN_FILE_LOCATION)
+        .withValue(s"$SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR/$secretLabel")
+        .endEnv()
+      .build()
+    originalPodWithMainContainer.copy(
+      pod = dtMountedPod,
+      mainContainer = mainContainerWithMountedKerberos)
+  }
+}

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

@@ -45,9 +45,6 @@ package object constants {
 
   // Hadoop credentials secrets for the Spark app.
   private[spark] val SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR = "/mnt/secrets/hadoop-credentials"
-  private[spark] val SPARK_APP_HADOOP_TOKEN_FILE_SECRET_NAME = "hadoop-token-file"
-  private[spark] val SPARK_APP_HADOOP_TOKEN_FILE_PATH =
-    s"$SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR/$SPARK_APP_HADOOP_TOKEN_FILE_SECRET_NAME"
   private[spark] val SPARK_APP_HADOOP_SECRET_VOLUME_NAME = "hadoop-secret"
 
   // Default and fixed ports
@@ -110,12 +107,12 @@ package object constants {
   // Kerberos Configuration
   private[spark] val HADOOP_KERBEROS_SECRET_NAME =
     "spark.kubernetes.kerberos.dt"
-  private[spark] val KERBEROS_SPARK_CONF_NAME =
-    "spark.kubernetes.kerberos.secretlabelname"
+  private[spark] val HADOOP_KERBEROS_CONF_SECRET =
+    "spark.kubernetes.kerberos.secretname"
+  private[spark] val HADOOP_KERBEROS_CONF_LABEL =
+    "spark.kubernetes.kerberos.labelname"
   private[spark] val KERBEROS_SECRET_LABEL_PREFIX =
     "hadoop-tokens"
-  private[spark] val ENV_KERBEROS_SECRET_LABEL =
-    "KERBEROS_SECRET_LABEL"
 
   // Miscellaneous
   private[spark] val ANNOTATION_EXECUTOR_NODE_AFFINITY = "scheduler.alpha.kubernetes.io/affinity"

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

@@ -16,8 +16,6 @@
  */
 package org.apache.spark.deploy.kubernetes.submit
 
-import java.io.File
-
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.kubernetes.ConfigurationUtils
 import org.apache.spark.deploy.kubernetes.config._
@@ -99,14 +97,6 @@ private[spark] class DriverConfigurationStepsOrchestrator(
         submissionSparkConf)
     val kubernetesCredentialsStep = new DriverKubernetesCredentialsStep(
         submissionSparkConf, kubernetesResourceNamePrefix)
-    // CHANGES
-    val hadoopCredentialsStep = new DriverHadoopCredentialsStep(submissionSparkConf)
-    val hadoopConfigurations2 =
-      sys.env.get("HADOOP_CONF_DIR").map{ conf => getHadoopConfFiles(conf)}
-          .getOrElse(Array.empty[File])
-    // CHANGES
-    val hadoopConfigurations = hadoopConfDir.map(conf => getHadoopConfFiles(conf))
-      .getOrElse(Array.empty[File])
     val hadoopConfigSteps =
       if (hadoopConfDir.isEmpty) {
         Option.empty[DriverConfigurationStep]
@@ -157,7 +147,6 @@ private[spark] class DriverConfigurationStepsOrchestrator(
     Seq(
       initialSubmissionStep,
       kubernetesCredentialsStep,
-      hadoopCredentialsStep,
       dependencyResolutionStep) ++
       initContainerBootstrapStep.toSeq ++
       hadoopConfigSteps.toSeq ++

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

@@ -40,7 +40,9 @@ private[spark] class HadoopConfigBootstrapStep(
       driverContainer = driverSpec.driverContainer,
       configMapProperties = Map.empty[String, String],
       additionalDriverSparkConf = Map.empty[String, String],
-      dtSecret = None)
+      dtSecret = None,
+      dtSecretName = HADOOP_KERBEROS_SECRET_NAME,
+      dtSecretLabel = "")
     for (nextStep <- hadoopConfigurationSteps) {
       currentHadoopSpec = nextStep.configureContainers(currentHadoopSpec)
     }

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

@@ -29,10 +29,14 @@ import io.fabric8.kubernetes.api.model.{Container, Pod, Secret}
   * - The properties that will be stored into the config map which have (key, value)
   *   pairs of (path, data)
   * - The secret containing a DT, either previously specified or built on the fly
+  * - The name of the secret where the DT will be stored
+  * - The label on the secret which correlates with where the current DT data is stored
   */
 private[spark] case class HadoopConfigSpec(
   additionalDriverSparkConf: Map[String, String],
   driverPod: Pod,
   driverContainer: Container,
   configMapProperties: Map[String, String],
-  dtSecret: Option[Secret])
+  dtSecret: Option[Secret],
+  dtSecretName: String,
+  dtSecretLabel: String)

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

@@ -31,9 +31,9 @@ import org.apache.hadoop.security.token.{Token, TokenIdentifier}
 import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier
 
 import org.apache.spark.SparkConf
-import org.apache.spark.deploy.kubernetes.{KerberosConfBootstrapImpl, PodWithMainContainer}
-import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.deploy.kubernetes.{KerberosTokenConfBootstrapImpl, PodWithMainContainer}
+import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.internal.Logging
 
  /**
@@ -101,29 +101,32 @@ private[spark] class HadoopKerberosKeytabResolverStep(
     if (renewedTokens.isEmpty) logError("Did not obtain any Delegation Tokens")
     val data = serialize(renewedCredentials)
     val renewalTime = getTokenRenewalInterval(renewedTokens, hadoopConf).getOrElse(Long.MaxValue)
-    val delegationToken = HDFSDelegationToken(data, renewalTime)
     val currentTime: Long = System.currentTimeMillis()
     val initialTokenLabelName = s"$KERBEROS_SECRET_LABEL_PREFIX-$currentTime-$renewalTime"
-    logInfo(s"Storing dt in $initialTokenLabelName")
     val secretDT =
       new SecretBuilder()
         .withNewMetadata()
           .withName(HADOOP_KERBEROS_SECRET_NAME)
           .endMetadata()
-          .addToData(initialTokenLabelName, Base64.encodeBase64String(delegationToken.bytes))
+          .addToData(initialTokenLabelName, Base64.encodeBase64String(data))
       .build()
-    val bootstrapKerberos = new KerberosConfBootstrapImpl(initialTokenLabelName)
+    val bootstrapKerberos = new KerberosTokenConfBootstrapImpl(
+      HADOOP_KERBEROS_SECRET_NAME,
+      initialTokenLabelName)
     val withKerberosEnvPod = bootstrapKerberos.bootstrapMainContainerAndVolumes(
       PodWithMainContainer(
         hadoopConfigSpec.driverPod,
         hadoopConfigSpec.driverContainer))
     hadoopConfigSpec.copy(
       additionalDriverSparkConf =
         hadoopConfigSpec.additionalDriverSparkConf ++ Map(
-          KERBEROS_SPARK_CONF_NAME -> initialTokenLabelName),
+          HADOOP_KERBEROS_CONF_LABEL -> initialTokenLabelName,
+          HADOOP_KERBEROS_CONF_SECRET -> HADOOP_KERBEROS_SECRET_NAME),
       driverPod = withKerberosEnvPod.pod,
       driverContainer = withKerberosEnvPod.mainContainer,
-      dtSecret = Some(secretDT))
+      dtSecret = Some(secretDT),
+      dtSecretName = HADOOP_KERBEROS_SECRET_NAME,
+      dtSecretLabel = initialTokenLabelName)
   }
 
   // Functions that should be in Core with Rebase to 2.3