Make sure principal and keytab are set before CoarseGrainedSchedulerBackend is started.

Also schedule re-logins in CoarseGrainedSchedulerBackend#start()

Author: harishreedharan

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -17,20 +17,14 @@
 
 package org.apache.spark.deploy
 
-import java.io.{ByteArrayInputStream, DataInputStream, DataOutputStream, ByteArrayOutputStream}
 import java.lang.reflect.Method
-import java.net.URI
-import java.nio.ByteBuffer
 import java.security.PrivilegedExceptionAction
-import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.{TimeUnit, ThreadFactory, Executors}
 
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileUtil, FileStatus, FileSystem, Path}
+import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 import org.apache.hadoop.fs.FileSystem.Statistics
-import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier
 import org.apache.hadoop.mapred.JobConf
-import org.apache.hadoop.mapreduce.{JobContext, TaskAttemptContext}
+import org.apache.hadoop.mapreduce.JobContext
 import org.apache.hadoop.security.Credentials
 import org.apache.hadoop.security.UserGroupInformation
 

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -108,7 +108,9 @@ private[spark] class CoarseGrainedExecutorBackend(
       context.stop(self)
       context.system.shutdown()
 
+    // Add new credentials received from the driver to the current user.
     case UpdateCredentials(newCredentials) =>
+      logInfo("New credentials received from driver, adding the credentials to the current user")
       val credentials = new Credentials()
       credentials.readTokenStorageStream(
         new DataInputStream(new ByteArrayInputStream(newCredentials.value.array())))

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala

@@ -51,7 +51,8 @@ private[spark] object CoarseGrainedClusterMessages {
   case class StatusUpdate(executorId: String, taskId: Long, state: TaskState,
     data: SerializableBuffer) extends CoarseGrainedClusterMessage
 
-  // Driver to all executors.
+  // When the delegation tokens are about expire, the driver creates new tokens and sends them to
+  // the executors via this message.
   case class UpdateCredentials(newCredentials: SerializableBuffer)
     extends CoarseGrainedClusterMessage
 

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -76,10 +76,6 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val actorSyste
     override protected def log = CoarseGrainedSchedulerBackend.this.log
     private val addressToExecutorId = new HashMap[Address, String]
 
-    // If a principal and keytab have been set, use that to create new credentials for executors
-    // periodically
-    SparkHadoopUtil.get.scheduleLoginFromKeytab(sendNewCredentialsToExecutors _)
-
     override def preStart() {
       // Listen for remote client disconnection events, since they don't go through Akka's watch()
       context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])
@@ -90,6 +86,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val actorSyste
       context.system.scheduler.schedule(0.millis, reviveInterval.millis, self, ReviveOffers)
     }
 
+    /**
+     * Send new credentials to executors. This is the method that is called when the scheduled
+     * login completes, so the new credentials can be sent to the executors.
+     * @param credentials
+     */
     def sendNewCredentialsToExecutors(credentials: SerializableBuffer): Unit = {
       executorDataMap.values.foreach{ x =>
         x.executorActor ! UpdateCredentials(credentials)
@@ -245,9 +246,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val actorSyste
         properties += ((key, value))
       }
     }
+
+    val driver = new DriverActor(properties)
     // TODO (prashant) send conf instead of properties
     driverActor = actorSystem.actorOf(
-      Props(new DriverActor(properties)), name = CoarseGrainedSchedulerBackend.ACTOR_NAME)
+      Props(driver), name = CoarseGrainedSchedulerBackend.ACTOR_NAME)
+
+    // If a principal and keytab have been set, use that to create new credentials for executors
+    // periodically
+    SparkHadoopUtil.get.scheduleLoginFromKeytab(driver.sendNewCredentialsToExecutors _)
   }
 
   def stopExecutors() {

yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -256,6 +256,12 @@ private[spark] class ApplicationMaster(
 
   private def runDriver(securityMgr: SecurityManager): Unit = {
     addAmIpFilter()
+
+    // This must be done before SparkContext is initialized, since the CoarseGrainedSchedulerBackend
+    // is started at that time. That is what schedules the re-logins. It is scheduled only if the
+    // principal is actually setup. So we make sure it is available.
+    SparkHadoopUtil.get.setPrincipalAndKeytabForLogin(
+      System.getenv("SPARK_PRINCIPAL"), System.getenv("SPARK_KEYTAB"))
     userClassThread = startUserApplication()
 
     // This a bit hacky, but we need to wait until the spark.driver.port property has
@@ -576,11 +582,6 @@ object ApplicationMaster extends Logging {
       master = new ApplicationMaster(amArgs, new YarnRMClient(amArgs))
       System.exit(master.run())
     }
-    // At this point, we have tokens that will expire only after a while, so we now schedule a
-    // login for some time before the tokens expire. Since the SparkContext has already started,
-    // we can now get access to the driver actor as well.
-    SparkHadoopUtil.get.setPrincipalAndKeytabForLogin(
-      System.getenv("SPARK_PRINCIPAL"), System.getenv("SPARK_KEYTAB"))
   }
 
   private[spark] def sparkContextInitialized(sc: SparkContext) = {

yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -568,11 +568,14 @@ private[spark] class Client(
           case Some(keytabPath) =>
             // Generate a file name that can be used for the keytab file, that does not conflict
             // with any user file.
+            logInfo("Attempting to login to the Kerberos" +
+              s" using principal: $principal and keytab: $keytabPath")
             val f = new File(keytabPath)
             keytabFileName = f.getName + "-" + System.currentTimeMillis()
             val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytabPath)
             credentials = ugi.getCredentials
             loginFromKeytab = true
+            logInfo("Successfully logged into Kerberos.")
           case None =>
             throw new SparkException("Keytab must be specified when principal is specified.")
         }

yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala

@@ -92,7 +92,7 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
     if (credentials != null) credentials.getSecretKey(new Text(key)) else null
   }
 
-  override def setPrincipalAndKeytabForLogin(principal: String, keytab: String): Unit ={
+  override def setPrincipalAndKeytabForLogin(principal: String, keytab: String): Unit = {
     loginPrincipal = Option(principal)
     keytabFile = Option(keytab)
   }