Using locks and monitors to ensure proper configs for setting up kerberized cluster in integration tests

Author: ifilonenko

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

@@ -17,19 +17,26 @@
 package org.apache.spark.deploy.kubernetes.submit.submitsteps.hadoopsteps
 
 import java.io._
-import java.security.PrivilegedExceptionAction
+
+import scala.collection.JavaConverters._
+import scala.util.Try
 
 import io.fabric8.kubernetes.api.model.SecretBuilder
 import org.apache.commons.codec.binary.Base64
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileSystem
 import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+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.SparkHadoopUtil
 import org.apache.spark.deploy.kubernetes.{KerberosConfBootstrapImpl, PodWithMainContainer}
 import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.internal.Logging
 
+
+
  /**
   * Step that configures the ConfigMap + Volumes for the driver
   */
@@ -59,20 +66,26 @@ private[spark] class HadoopKerberosKeytabResolverStep(
       }
     // In the case that keytab is not specified we will read from Local Ticket Cache
     val jobUserUGI = maybeJobUserUGI.getOrElse(UserGroupInformation.getCurrentUser)
-    logInfo(s"Primary group name: ${jobUserUGI.getPrimaryGroupName}")
-    val credentials: Credentials = jobUserUGI.getCredentials
-    val credentialsManager = newHadoopTokenManager(submissionSparkConf, hadoopConf)
-    var renewalTime = Long.MaxValue
-    jobUserUGI.doAs(new PrivilegedExceptionAction[Void] {
-      override def run(): Void = {
-        renewalTime = Math.min(
-          obtainCredentials(credentialsManager, hadoopConf, credentials),
-          renewalTime)
-        null
-      }
-    })
-    if (credentials.getAllTokens.isEmpty) logError("Did not obtain any Delegation Tokens")
-    val data = serialize(credentials)
+    logInfo(s"Retrieved Job User UGI: $jobUserUGI")
+    val originalCredentials: Credentials = jobUserUGI.getCredentials
+    logInfo(s"Original tokens: ${originalCredentials.toString}")
+    logInfo(s"All tokens: ${originalCredentials.getAllTokens}")
+    logInfo(s"All secret keys: ${originalCredentials.getAllSecretKeys}")
+    val dfs: FileSystem = FileSystem.get(hadoopConf)
+    // This is not necessary with [Spark-20328] since we would be using
+    // Spark core providers to handle delegation token renewal
+    val renewer: String = jobUserUGI.getShortUserName
+    logInfo(s"Renewer is: $renewer")
+    val renewedCredentials: Credentials = new Credentials(originalCredentials)
+    dfs.addDelegationTokens(renewer, renewedCredentials)
+    val renewedTokens = renewedCredentials.getAllTokens.asScala
+    logInfo(s"Renewed tokens: ${renewedCredentials.toString}")
+    logInfo(s"All renewed tokens: ${renewedTokens}")
+    logInfo(s"All renewed secret keys: ${renewedCredentials.getAllSecretKeys}")
+    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 initialTokenLabelName = s"$KERBEROS_SECRET_LABEL_PREFIX-1-$renewalTime"
     logInfo(s"Storing dt in $initialTokenLabelName")
@@ -97,24 +110,24 @@ private[spark] class HadoopKerberosKeytabResolverStep(
       dtSecret = Some(secretDT))
   }
 
-  // Functions that should be in SparkHadoopUtil with Rebase to 2.2
+  // Functions that should be in Core with Rebase to 2.3
   @deprecated("Moved to core in 2.2", "2.2")
-  private def obtainCredentials(instance: Any, args: AnyRef*): Long = {
-    val method = Class
-      .forName("org.apache.spark.deploy.yarn.security.ConfigurableCredentialManager")
-      .getMethod("obtainCredentials", classOf[Configuration], classOf[Configuration])
-    method.setAccessible(true)
-    method.invoke(instance, args: _*).asInstanceOf[Long]
+  private def getTokenRenewalInterval(
+    renewedTokens: Iterable[Token[_ <: TokenIdentifier]],
+    hadoopConf: Configuration): Option[Long] = {
+      val renewIntervals = renewedTokens.filter {
+        _.decodeIdentifier().isInstanceOf[AbstractDelegationTokenIdentifier]}
+        .flatMap { token =>
+        Try {
+          val newExpiration = token.renew(hadoopConf)
+          val identifier = token.decodeIdentifier().asInstanceOf[AbstractDelegationTokenIdentifier]
+          val interval = newExpiration - identifier.getIssueDate
+          logInfo(s"Renewal interval is $interval for token ${token.getKind.toString}")
+          interval
+        }.toOption}
+      if (renewIntervals.isEmpty) None else Some(renewIntervals.min)
   }
-   @deprecated("Moved to core in 2.2", "2.2")
-   // This method will instead be using HadoopDelegationTokenManager from Spark 2.2
-   private def newHadoopTokenManager(args: AnyRef*): Any = {
-     val constructor = Class
-       .forName("org.apache.spark.deploy.yarn.security.ConfigurableCredentialManager")
-       .getConstructor(classOf[SparkConf], classOf[Configuration])
-     constructor.setAccessible(true)
-     constructor.newInstance(args: _*)
-   }
+
   @deprecated("Moved to core in 2.2", "2.2")
   private def serialize(creds: Credentials): Array[Byte] = {
     val byteStream = new ByteArrayOutputStream

resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/kerberos-test/Dockerfile

@@ -21,25 +21,7 @@ FROM spark-base
 # command should be invoked from the top level directory of the Spark distribution. E.g.:
 # docker build -t kerberos-test:latest -f dockerfiles/kerberos-test/Dockerfile .
 
-COPY examples /opt/spark/examples
 RUN apk add --no-cache --update krb5 krb5-libs
-COPY hconf/krb5.conf /etc/krb5.conf
-COPY test-scripts/test-env.sh /
-
-CMD /opt/spark/bin/spark-submit \
-      --deploy-mode cluster \
-      --class ${CLASS_NAME} \
-      --master k8s://${MASTER_URL} \
-      --kubernetes-namespace ${NAMESPACE} \
-      --conf spark.executor.instances=1 \
-      --conf spark.app.name=spark-hdfs \
-      --conf spark.kubernetes.driver.docker.image=spark-driver:latest \
-      --conf spark.kubernetes.executor.docker.image=spark-executor:latest \
-      --conf spark.kubernetes.initcontainer.docker.image=spark-init:latest \
-      --conf spark.kubernetes.kerberos=true \
-      --conf spark.kubernetes.kerberos.keytab=/var/keytabs/hdfs.keytab \
-      --conf spark.kubernetes.kerberos.principal=hdfs/nn.${NAMESPACE}[email protected] \
-      --conf spark.kubernetes.driver.labels=spark-app-locator=${APP_LOCATOR_LABEL} \
-      --files local:///etc/hadoop/core-site.xml,local:///etc/hadoop/hdfs-site.xml,local:///etc/hadoop/yarn-site.xml \
-      ${SUBMIT_RESOURCE} \
-      hdfs://nn.${NAMESPACE}.svc.cluster.local:9000/user/ifilonenko/wordcount.txt
+COPY examples /opt/spark/examples
+COPY test-scripts/test-env.sh /opt/spark/
+COPY hconf /opt/spark/hconf

resource-managers/kubernetes/integration-tests/kerberos-yml/data-populator-deployment.yml

@@ -11,6 +11,7 @@ spec:
       labels:
         name: hdfs-data-populator
         kerberosService: data-populator
+        job: kerberos-test
     spec:
       containers:
       - command:

resource-managers/kubernetes/integration-tests/kerberos-yml/data-populator-service.yml

@@ -5,6 +5,7 @@ metadata:
     service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
   labels:
     kerberosService: data-populator
+    job: kerberos-test
   name: data-populator
 spec:
   clusterIP: None

resource-managers/kubernetes/integration-tests/kerberos-yml/dn1-deployment.yml

@@ -11,6 +11,7 @@ spec:
       labels:
         name: hdfs-dn1
         kerberosService: dn1
+        job: kerberos-test
     spec:
       containers:
       - command:

resource-managers/kubernetes/integration-tests/kerberos-yml/dn1-service.yml

@@ -5,6 +5,7 @@ metadata:
     service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
   labels:
     kerberosService: dn1
+    job: kerberos-test
   name: dn1
 spec:
   clusterIP: None

resource-managers/kubernetes/integration-tests/kerberos-yml/kerberos-deployment.yml

@@ -11,6 +11,7 @@ spec:
       labels:
         name: hdfs-kerberos
         kerberosService: kerberos
+        job: kerberos-test
     spec:
       containers:
       - command:

resource-managers/kubernetes/integration-tests/kerberos-yml/kerberos-service.yml

@@ -5,6 +5,7 @@ metadata:
     service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
   labels:
     kerberosService: kerberos
+    job: kerberos-test
   name: kerberos
 spec:
   clusterIP: None

resource-managers/kubernetes/integration-tests/kerberos-yml/namenode-hadoop-pv.yml

@@ -7,7 +7,7 @@ metadata:
     job: kerberostest
 spec:
   capacity:
-    storage: 10Gi
+    storage: 1Gi
   accessModes:
     - ReadWriteOnce
   hostPath:

resource-managers/kubernetes/integration-tests/kerberos-yml/namenode-hadoop.yml

@@ -2,6 +2,8 @@ apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
   name: nn-hadoop
+  labels:
+    job: kerberostest
 spec:
   accessModes:
   - ReadWriteOnce