[SPARK-23257][K8S] Kerberos Support for Spark on K8S

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

@@ -72,7 +72,7 @@ private[spark] case class KubernetesConf[T <: KubernetesRoleSpecificConf](
     roleEnvs: Map[String, String],
     roleVolumes: Iterable[KubernetesVolumeSpec[_ <: KubernetesVolumeSpecificConf]],
     sparkFiles: Seq[String],
-    hadoopConfDir: Option[HadoopConfSpec]) {
+    hadoopConfSpec: Option[HadoopConfSpec]) {
 
   def hadoopConfigMapName: String = s"$appResourceNamePrefix-hadoop-config"
 

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/HadoopConfExecutorFeatureStep.scala

@@ -34,12 +34,12 @@ private[spark] class HadoopConfExecutorFeatureStep(
 
    override def configurePod(pod: SparkPod): SparkPod = {
      val sparkConf = kubernetesConf.sparkConf
-     val maybeHadoopConfDir = sparkConf.getOption(HADOOP_CONFIG_MAP_NAME)
-     require(maybeHadoopConfDir.isDefined,
+     val hadoopConfDirCMapName = sparkConf.getOption(HADOOP_CONFIG_MAP_NAME)
+     require(hadoopConfDirCMapName.isDefined,
        "Ensure that the env `HADOOP_CONF_DIR` is defined either in the client or " +
          " using pre-existing ConfigMaps")
      logInfo("HADOOP_CONF_DIR defined")
-     HadoopBootstrapUtil.bootstrapHadoopConfDir(None, None, maybeHadoopConfDir, pod)
+     HadoopBootstrapUtil.bootstrapHadoopConfDir(None, None, hadoopConfDirCMapName, pod)
   }
 
   override def getAdditionalPodSystemProperties(): Map[String, String] = Map.empty

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala

@@ -34,9 +34,9 @@ private[spark] class KerberosConfDriverFeatureStep(
    kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf])
    extends KubernetesFeatureConfigStep with Logging {
 
-   require(kubernetesConf.hadoopConfDir.isDefined,
+   require(kubernetesConf.hadoopConfSpec.isDefined,
      "Ensure that HADOOP_CONF_DIR is defined either via env or a pre-defined ConfigMap")
-   private val hadoopConfDirSpec = kubernetesConf.hadoopConfDir.get
+   private val hadoopConfDirSpec = kubernetesConf.hadoopConfSpec.get
    private val conf = kubernetesConf.sparkConf
    private val principal = conf.get(org.apache.spark.internal.config.PRINCIPAL)
    private val keytab = conf.get(org.apache.spark.internal.config.KEYTAB)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/hadooputils/HadoopKerberosLogin.scala

@@ -48,10 +48,8 @@ private[spark] object HadoopKerberosLogin {
        submissionSparkConf,
        hadoopConf)
      require(tokenData.nonEmpty, "Did not obtain any delegation tokens")
-     val currentTime = tokenManager.getCurrentTime
-     val initialTokenDataKeyName = s"$KERBEROS_SECRET_KEY_PREFIX-$currentTime"
-     val newSecretName =
-       s"$kubernetesResourceNamePrefix-$KERBEROS_DELEGEGATION_TOKEN_SECRET_NAME.$currentTime"
+     val initialTokenDataKeyName = KERBEROS_SECRET_KEY_PREFIX
+     val newSecretName = s"$kubernetesResourceNamePrefix-$KERBEROS_DELEGEGATION_TOKEN_SECRET_NAME"
      val secretDT =
        new SecretBuilder()
          .withNewMetadata()

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/security/KubernetesHadoopDelegationTokenManager.scala

@@ -25,7 +25,6 @@ import org.apache.spark.SparkConf
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.security.HadoopDelegationTokenManager
 import org.apache.spark.internal.Logging
-import org.apache.spark.util.{Clock, SystemClock}
 
  /**
   * The KubernetesHadoopDelegationTokenManager fetches Hadoop delegation tokens
@@ -37,14 +36,12 @@ private[spark] class KubernetesHadoopDelegationTokenManager(
    tokenManager: HadoopDelegationTokenManager) extends Logging {
 
    // HadoopUGI Util methods
-   private val clock: Clock = new SystemClock()
    def getCurrentUser: UserGroupInformation = UserGroupInformation.getCurrentUser
    def getShortUserName : String = getCurrentUser.getShortUserName
    def getFileSystem(hadoopConf: Configuration) : FileSystem = FileSystem.get(hadoopConf)
    def isSecurityEnabled: Boolean = UserGroupInformation.isSecurityEnabled
    def loginUserFromKeytabAndReturnUGI(principal: String, keytab: String): UserGroupInformation =
      UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytab)
-   def getCurrentTime: Long = clock.getTimeMillis()
    def serializeCreds(creds: Credentials): Array[Byte] = SparkHadoopUtil.get.serialize(creds)
    def nextRT(rt: Long, conf: SparkConf): Long = SparkHadoopUtil.nextCredentialRenewalTime(rt, conf)
 

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

@@ -85,7 +85,7 @@ private[spark] class KubernetesDriverBuilder(
       .getOrElse(provideJavaStep(kubernetesConf))
 
     val maybeHadoopConfigStep =
-      kubernetesConf.hadoopConfDir.map { _ =>
+      kubernetesConf.hadoopConfSpec.map { _ =>
           provideHadoopGlobalStep(kubernetesConf)}
 
     val allFeatures: Seq[KubernetesFeatureConfigStep] =

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala

@@ -78,7 +78,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
       DRIVER_ENVS,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val featureStep = new BasicDriverFeatureStep(kubernetesConf)
     val basePod = SparkPod.initialPod()
@@ -157,7 +157,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
       DRIVER_ENVS,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val pythonKubernetesConf = KubernetesConf(
       pythonSparkConf,
@@ -175,7 +175,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
       DRIVER_ENVS,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     val javaFeatureStep = new BasicDriverFeatureStep(javaKubernetesConf)
     val pythonFeatureStep = new BasicDriverFeatureStep(pythonKubernetesConf)
     val basePod = SparkPod.initialPod()
@@ -205,7 +205,7 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
       DRIVER_ENVS,
       Nil,
       allFiles,
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new BasicDriverFeatureStep(kubernetesConf)
     val additionalProperties = step.getAdditionalPodSystemProperties()

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala

@@ -92,7 +92,7 @@ class BasicExecutorFeatureStepSuite
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     val executor = step.configurePod(SparkPod.initialPod())
 
     // The executor pod name and default labels.
@@ -133,7 +133,7 @@ class BasicExecutorFeatureStepSuite
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     assert(step.configurePod(SparkPod.initialPod()).pod.getSpec.getHostname.length === 63)
   }
 
@@ -155,7 +155,7 @@ class BasicExecutorFeatureStepSuite
         Map("qux" -> "quux"),
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     val executor = step.configurePod(SparkPod.initialPod())
 
     checkEnv(executor,
@@ -183,7 +183,7 @@ class BasicExecutorFeatureStepSuite
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     val executor = step.configurePod(SparkPod.initialPod())
     // This is checking that basic executor + executorMemory = 1408 + 42 = 1450
     assert(executor.container.getResources.getRequests.get("memory").getAmount === "1450Mi")

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStepSuite.scala

@@ -63,7 +63,7 @@ class DriverKubernetesCredentialsFeatureStepSuite extends SparkFunSuite with Bef
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     val kubernetesCredentialsStep = new DriverKubernetesCredentialsFeatureStep(kubernetesConf)
     assert(kubernetesCredentialsStep.configurePod(BASE_DRIVER_POD) === BASE_DRIVER_POD)
     assert(kubernetesCredentialsStep.getAdditionalPodSystemProperties().isEmpty)
@@ -96,7 +96,7 @@ class DriverKubernetesCredentialsFeatureStepSuite extends SparkFunSuite with Bef
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val kubernetesCredentialsStep = new DriverKubernetesCredentialsFeatureStep(kubernetesConf)
     assert(kubernetesCredentialsStep.configurePod(BASE_DRIVER_POD) === BASE_DRIVER_POD)
@@ -136,7 +136,7 @@ class DriverKubernetesCredentialsFeatureStepSuite extends SparkFunSuite with Bef
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     val kubernetesCredentialsStep = new DriverKubernetesCredentialsFeatureStep(kubernetesConf)
     val resolvedProperties = kubernetesCredentialsStep.getAdditionalPodSystemProperties()
     val expectedSparkConf = Map(

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala

@@ -69,7 +69,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     assert(configurationStep.configurePod(SparkPod.initialPod()) === SparkPod.initialPod())
     assert(configurationStep.getAdditionalKubernetesResources().size === 1)
     assert(configurationStep.getAdditionalKubernetesResources().head.isInstanceOf[Service])
@@ -102,7 +102,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     val expectedServiceName = SHORT_RESOURCE_NAME_PREFIX +
       DriverServiceFeatureStep.DRIVER_SVC_POSTFIX
     val expectedHostName = s"$expectedServiceName.my-namespace.svc"
@@ -125,7 +125,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None))
+        hadoopConfSpec = None))
     val resolvedService = configurationStep
       .getAdditionalKubernetesResources()
       .head
@@ -157,7 +157,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
         Map.empty,
         Nil,
         Seq.empty[String],
-        hadoopConfDir = None),
+        hadoopConfSpec = None),
       clock)
     val driverService = configurationStep
       .getAdditionalKubernetesResources()
@@ -186,7 +186,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
           Map.empty,
           Nil,
           Seq.empty[String],
-          hadoopConfDir = None),
+          hadoopConfSpec = None),
         clock)
       fail("The driver bind address should not be allowed.")
     } catch {
@@ -213,7 +213,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
           Map.empty,
           Nil,
           Seq.empty[String],
-          hadoopConfDir = None),
+          hadoopConfSpec = None),
         clock)
       fail("The driver host address should not be allowed.")
     } catch {

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/EnvSecretsFeatureStepSuite.scala

@@ -47,7 +47,7 @@ class EnvSecretsFeatureStepSuite extends SparkFunSuite{
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new EnvSecretsFeatureStep(kubernetesConf)
     val driverContainerWithEnvSecrets = step.configurePod(baseDriverPod).container

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala

@@ -49,7 +49,7 @@ class LocalDirsFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
   }
 
   test("Resolve to default local dir if neither env nor configuration are set") {

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountSecretsFeatureStepSuite.scala

@@ -45,7 +45,7 @@ class MountSecretsFeatureStepSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new MountSecretsFeatureStep(kubernetesConf)
     val driverPodWithSecretsMounted = step.configurePod(baseDriverPod).pod

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountVolumesFeatureStepSuite.scala

@@ -37,7 +37,7 @@ class MountVolumesFeatureStepSuite extends SparkFunSuite {
     roleEnvs = Map.empty,
     roleVolumes = Nil,
     sparkFiles = Nil,
-    hadoopConfDir = None)
+    hadoopConfSpec = None)
 
   test("Mounts hostPath volumes") {
     val volumeConf = KubernetesVolumeSpec(

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/bindings/JavaDriverFeatureStepSuite.scala

@@ -44,7 +44,7 @@ class JavaDriverFeatureStepSuite extends SparkFunSuite {
       roleEnvs = Map.empty,
       roleVolumes = Nil,
       sparkFiles = Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new JavaDriverFeatureStep(kubernetesConf)
     val driverPod = step.configurePod(baseDriverPod).pod

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/bindings/PythonDriverFeatureStepSuite.scala

@@ -54,7 +54,7 @@ class PythonDriverFeatureStepSuite extends SparkFunSuite {
       roleEnvs = Map.empty,
       roleVolumes = Nil,
       sparkFiles = Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new PythonDriverFeatureStep(kubernetesConf)
     val driverPod = step.configurePod(baseDriverPod).pod
@@ -92,7 +92,7 @@ class PythonDriverFeatureStepSuite extends SparkFunSuite {
       roleEnvs = Map.empty,
       roleVolumes = Nil,
       sparkFiles = Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     val step = new PythonDriverFeatureStep(kubernetesConf)
     val driverContainerwithPySpark = step.configurePod(baseDriverPod).container
     val args = driverContainerwithPySpark

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/bindings/RDriverFeatureStepSuite.scala

@@ -48,7 +48,7 @@ class RDriverFeatureStepSuite extends SparkFunSuite {
       roleEnvs = Map.empty,
       roleVolumes = Seq.empty,
       sparkFiles = Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
 
     val step = new RDriverFeatureStep(kubernetesConf)
     val driverContainerwithR = step.configurePod(baseDriverPod).container

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/ClientSuite.scala

@@ -143,7 +143,7 @@ class ClientSuite extends SparkFunSuite with BeforeAndAfter {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     when(driverBuilder.buildFromFeatures(kubernetesConf)).thenReturn(BUILT_KUBERNETES_SPEC)
     when(kubernetesClient.pods()).thenReturn(podOperations)
     when(podOperations.withName(POD_NAME)).thenReturn(namedPods)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/KubernetesDriverBuilderSuite.scala

@@ -99,7 +99,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -126,7 +126,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -155,7 +155,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -182,7 +182,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -214,7 +214,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       volumeSpec :: Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -242,7 +242,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = None)
+      hadoopConfSpec = None)
     validateStepTypesApplied(
       builderUnderTest.buildFromFeatures(conf),
       BASIC_STEP_TYPE,
@@ -269,7 +269,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = Some(
+      hadoopConfSpec = Some(
         HadoopConfSpec(
           Some("/var/hadoop-conf"),
           None)))
@@ -300,7 +300,7 @@ class KubernetesDriverBuilderSuite extends SparkFunSuite {
       Map.empty,
       Nil,
       Seq.empty[String],
-      hadoopConfDir = Some(
+      hadoopConfSpec = Some(
         HadoopConfSpec(
           None,
           Some("pre-defined-configMapName"))))