Added docs in the security guide. Changed some code to ensure that the renewer objects are created only if required.

Author: harishreedharan

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

@@ -17,62 +17,61 @@
 package org.apache.spark.deploy
 
 import java.util.concurrent.{Executors, TimeUnit}
-import java.util.{Comparator, Arrays}
 
-import com.google.common.primitives.Longs
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{PathFilter, FileStatus, Path, FileSystem}
+import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 import org.apache.hadoop.security.{Credentials, UserGroupInformation}
 
-import org.apache.spark.util.Utils
 import org.apache.spark.{Logging, SparkConf}
+import org.apache.spark.util.Utils
 
 private[spark] class ExecutorDelegationTokenUpdater(
     sparkConf: SparkConf,
     hadoopConf: Configuration) extends Logging {
 
   @volatile private var lastCredentialsFileSuffix = 0
 
-  private lazy val delegationTokenRenewer =
+  private val credentialsFile = sparkConf.get("spark.yarn.credentials.file")
+
+  private val delegationTokenRenewer =
     Executors.newSingleThreadScheduledExecutor(
       Utils.namedThreadFactory("Delegation Token Refresh Thread"))
 
   // On the executor, this thread wakes up and picks up new tokens from HDFS, if any.
-  private lazy val executorUpdaterRunnable =
+  private val executorUpdaterRunnable =
     new Runnable {
       override def run(): Unit = Utils.logUncaughtExceptions(updateCredentialsIfRequired())
     }
 
   def updateCredentialsIfRequired(): Unit = {
     try {
-      sparkConf.getOption("spark.yarn.credentials.file").foreach { credentialsFile =>
-        val credentials = UserGroupInformation.getCurrentUser.getCredentials
-        val credentialsFilePath = new Path(credentialsFile)
-        val remoteFs = FileSystem.get(hadoopConf)
-        SparkHadoopUtil.get.listFilesSorted(
-          remoteFs, credentialsFilePath.getParent, credentialsFilePath.getName, ".tmp")
-          .lastOption.foreach { credentialsStatus =>
-          val suffix = getSuffixForCredentialsPath(credentialsStatus)
-          if (suffix > lastCredentialsFileSuffix) {
-            logInfo("Reading new delegation tokens from " + credentialsStatus.getPath)
-            val newCredentials = getCredentialsFromHDFSFile(remoteFs, credentialsStatus.getPath)
-            lastCredentialsFileSuffix = suffix
-            UserGroupInformation.getCurrentUser.addCredentials(newCredentials)
-            val totalValidity = SparkHadoopUtil.get.getLatestTokenValidity(credentials) -
-              credentialsStatus.getModificationTime
-            val timeToRunRenewal =
-              credentialsStatus.getModificationTime + (0.8 * totalValidity).toLong
-            val timeFromNowToRenewal = timeToRunRenewal - System.currentTimeMillis()
-            logInfo("Updated delegation tokens, will check for new tokens in " +
-              timeFromNowToRenewal + " millis")
-            delegationTokenRenewer.schedule(
-              executorUpdaterRunnable, timeFromNowToRenewal, TimeUnit.MILLISECONDS)
-          } else {
-            // Check every hour to see if new credentials arrived.
-            logInfo("Updated delegation tokens were expected, but the driver has not updated the " +
-              "tokens yet, will check again in an hour.")
-            delegationTokenRenewer.schedule(executorUpdaterRunnable, 1, TimeUnit.HOURS)
-          }
+      val credentials = UserGroupInformation.getCurrentUser.getCredentials
+      val credentialsFilePath = new Path(credentialsFile)
+      val remoteFs = FileSystem.get(hadoopConf)
+      SparkHadoopUtil.get.listFilesSorted(
+        remoteFs, credentialsFilePath.getParent, credentialsFilePath.getName, ".tmp")
+        .lastOption
+        .foreach { credentialsStatus =>
+        val suffix = getSuffixForCredentialsPath(credentialsStatus)
+        if (suffix > lastCredentialsFileSuffix) {
+          logInfo("Reading new delegation tokens from " + credentialsStatus.getPath)
+          val newCredentials = getCredentialsFromHDFSFile(remoteFs, credentialsStatus.getPath)
+          lastCredentialsFileSuffix = suffix
+          UserGroupInformation.getCurrentUser.addCredentials(newCredentials)
+          val totalValidity = SparkHadoopUtil.get.getLatestTokenValidity(credentials) -
+            credentialsStatus.getModificationTime
+          val timeToRunRenewal =
+            credentialsStatus.getModificationTime + (0.8 * totalValidity).toLong
+          val timeFromNowToRenewal = timeToRunRenewal - System.currentTimeMillis()
+          logInfo("Updated delegation tokens, will check for new tokens in " +
+            timeFromNowToRenewal + " millis")
+          delegationTokenRenewer.schedule(
+            executorUpdaterRunnable, timeFromNowToRenewal, TimeUnit.MILLISECONDS)
+        } else {
+          // Check every hour to see if new credentials arrived.
+          logInfo("Updated delegation tokens were expected, but the driver has not updated the " +
+            "tokens yet, will check again in an hour.")
+          delegationTokenRenewer.schedule(executorUpdaterRunnable, 1, TimeUnit.HOURS)
         }
       }
     } catch {

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

@@ -20,11 +20,11 @@ package org.apache.spark.deploy
 import java.io.{ByteArrayInputStream, DataInputStream}
 import java.lang.reflect.Method
 import java.security.PrivilegedExceptionAction
-import java.util.{Comparator, Arrays}
+import java.util.{Arrays, Comparator}
 
 import com.google.common.primitives.Longs
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{PathFilter, FileStatus, FileSystem, Path}
+import org.apache.hadoop.fs.{FileStatus, FileSystem, Path, PathFilter}
 import org.apache.hadoop.fs.FileSystem.Statistics
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier
 import org.apache.hadoop.mapred.JobConf
@@ -44,7 +44,7 @@ import scala.collection.JavaConversions._
  */
 @DeveloperApi
 class SparkHadoopUtil extends Logging {
-  val sparkConf = new SparkConf()
+  private val sparkConf = new SparkConf()
   val conf: Configuration = newConfiguration(sparkConf)
   UserGroupInformation.setConfiguration(conf)
 
@@ -209,12 +209,12 @@ class SparkHadoopUtil extends Logging {
 
   /**
    * Lists all the files in a directory with the specified prefix, and does not end with the
-   * given suffix.
+   * given suffix. The returned {{FileStatus}} instances are sorted by the modification times of
+   * the respective files.
    * @param remoteFs
    * @param prefix
    * @return
    */
-
   def listFilesSorted(
       remoteFs: FileSystem,
       dir: Path,

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

@@ -501,7 +501,10 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
         |  --principal PRINCIPAL       Principal to be used to login to KDC, while running on
         |                              secure HDFS.
         |  --keytab KEYTAB             The full path to the file that contains the keytab for the
-        |                              principal specified above.
+        |                              principal specified above. This keytab will be copied to
+        |                              the node running the Application Master via the Secure
+        |                              Distributed Cache, for renewing the login tickets and the
+        |                              delegation tokens periodically.
       """.stripMargin
     )
     SparkSubmit.exitFn()

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

@@ -174,9 +174,14 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
           driverConf.set(key, value)
         }
       }
-      // Periodically update the credentials for this user to ensure HDFS tokens get updated.
-      val tokenUpdater = new ExecutorDelegationTokenUpdater(driverConf, SparkHadoopUtil.get.conf)
-      tokenUpdater.updateCredentialsIfRequired()
+      var tokenUpdaterOption: Option[ExecutorDelegationTokenUpdater] = None
+      if(driverConf.contains("spark.yarn.credentials.file")) {
+        // Periodically update the credentials for this user to ensure HDFS tokens get updated.
+        tokenUpdaterOption =
+          Some(new ExecutorDelegationTokenUpdater(driverConf, SparkHadoopUtil.get.conf))
+        tokenUpdaterOption.get.updateCredentialsIfRequired()
+      }
+
       val env = SparkEnv.createExecutorEnv(
         driverConf, executorId, hostname, port, cores, isLocal = false)
 
@@ -192,7 +197,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
         env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url))
       }
       env.actorSystem.awaitTermination()
-      tokenUpdater.stop()
+      tokenUpdaterOption.foreach(_.stop())
       env.rpcEnv.awaitTermination()
     }
   }

docs/security.md

@@ -31,6 +31,7 @@ SSL must be configured on each node and configured for each component involved i
 
 ### YARN mode
 The key-store can be prepared on the client side and then distributed and used by the executors as the part of the application. It is possible because the user is able to deploy files before the application is started in YARN by using `spark.yarn.dist.files` or `spark.yarn.dist.archives` configuration settings. The responsibility for encryption of transferring these files is on YARN side and has nothing to do with Spark.
+For long-running apps like Spark Streaming apps be able to write to HDFS, it is possible to pass a principal and keytab to `spark-submit` via the `--principal` and `--keytab` parameters respectively. The keytab passed in will be copied over to the machine running the Application Master securely via the Hadoop Distributed Cache. The Kerberos login will be periodically renewed using this principal and keytab and the delegation tokens required for HDFS will be generated periodically so the application can continue writing to HDFS.
 
 ### Standalone mode
 The user needs to provide key-stores and configuration options for master and workers. They have to be set by attaching appropriate Java system properties in `SPARK_MASTER_OPTS` and in `SPARK_WORKER_OPTS` environment variables, or just in `SPARK_DAEMON_JAVA_OPTS`. In this mode, the user may allow the executors to use the SSL settings inherited from the worker which spawned that executor. It can be accomplished by setting `spark.ssl.useNodeLocalConf` to `true`. If that parameter is set, the settings provided by user on the client side, are not used by the executors.

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

@@ -16,16 +16,14 @@
  */
 package org.apache.spark.deploy.yarn
 
-import java.io.{DataOutputStream, ByteArrayOutputStream}
+import java.io.{ByteArrayOutputStream, DataOutputStream}
 import java.nio.ByteBuffer
 import java.util.concurrent.{Executors, TimeUnit}
 
-import akka.actor.ActorSelection
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 import org.apache.hadoop.security.UserGroupInformation
 
-import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.rpc.RpcEndpointRef
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.NewTokens
 import org.apache.spark.{Logging, SparkConf}
@@ -50,18 +48,20 @@ import org.apache.spark.util.{SerializableBuffer, Utils}
  * appeared, it will read the credentials and update the currently running UGI with it. This
  * process happens again once 80% of the validity of this has expired.
  */
-class AMDelegationTokenRenewer(sparkConf: SparkConf, hadoopConf: Configuration) extends Logging {
+private[yarn] class AMDelegationTokenRenewer(
+    sparkConf: SparkConf,
+    hadoopConf: Configuration) extends Logging {
 
   private var lastCredentialsFileSuffix = 0
 
-  private lazy val delegationTokenRenewer =
+  private val delegationTokenRenewer =
     Executors.newSingleThreadScheduledExecutor(
       Utils.namedThreadFactory("Delegation Token Refresh Thread"))
 
   private var loggedInViaKeytab = false
   var driverEndPoint: RpcEndpointRef = null
 
-  private lazy val hadoopUtil = YarnSparkHadoopUtil.get
+  private val hadoopUtil = YarnSparkHadoopUtil.get
 
   /**
    * Schedule a login from the keytab and principal set using the --principal and --keytab
@@ -70,62 +70,51 @@ class AMDelegationTokenRenewer(sparkConf: SparkConf, hadoopConf: Configuration)
    * to do the login. This method is a no-op in non-YARN mode.
    */
   private[spark] def scheduleLoginFromKeytab(): Unit = {
-    sparkConf.getOption("spark.yarn.principal").foreach { principal =>
-      val keytab = sparkConf.get("spark.yarn.keytab")
-
-      def getRenewalInterval: Long = {
-        import scala.concurrent.duration._
-        val credentials = UserGroupInformation.getCurrentUser.getCredentials
-        val interval = (0.75 * (hadoopUtil.getLatestTokenValidity(credentials) -
-          System.currentTimeMillis())).toLong
-        // If only 6 hours left, then force a renewal immediately. This is to avoid tokens with
-        // very less validity being used on AM restart.
-        if ((interval millis).toHours <= 6) {
-          0L
-        } else {
-          interval
-        }
+    val principal = sparkConf.get("spark.yarn.principal")
+    val keytab = sparkConf.get("spark.yarn.keytab")
+
+    def getRenewalInterval: Long = {
+      import scala.concurrent.duration._
+      val credentials = UserGroupInformation.getCurrentUser.getCredentials
+      val interval = (0.75 * (hadoopUtil.getLatestTokenValidity(credentials) -
+        System.currentTimeMillis())).toLong
+      // If only 6 hours left, then force a renewal immediately. This is to avoid tokens with
+      // very less validity being used on AM restart.
+      if ((interval millis).toHours <= 6) {
+        0L
+      } else {
+        interval
       }
+    }
 
-      def scheduleRenewal(runnable: Runnable): Unit = {
-        val renewalInterval = getRenewalInterval
-        logInfo(s"Scheduling login from keytab in $renewalInterval millis.")
-        delegationTokenRenewer.schedule(runnable, renewalInterval, TimeUnit.MILLISECONDS)
-      }
+    def scheduleRenewal(runnable: Runnable): Unit = {
+      val renewalInterval = getRenewalInterval
+      logInfo(s"Scheduling login from keytab in $renewalInterval millis.")
+      delegationTokenRenewer.schedule(runnable, renewalInterval, TimeUnit.MILLISECONDS)
+    }
 
-      // This thread periodically runs on the driver to update the delegation tokens on HDFS.
-      val driverTokenRenewerRunnable =
-        new Runnable {
-          override def run(): Unit = {
-            var wroteNewFiles = false
-            try {
-              writeNewTokensToHDFS(principal, keytab)
-              wroteNewFiles = true
-              cleanupOldFiles()
-            } catch {
-              case e: Exception =>
-                // If the exception was due to some issue deleting files, don't worry about it -
-                // just try to clean up next time. Else, reschedule for an hour later so new
-                // tokens get written out.
-                if (!wroteNewFiles) {
-                  logWarning("Failed to write out new credentials to HDFS, will try again in an " +
-                    "hour! If this happens too often tasks will fail.", e)
-                  delegationTokenRenewer.schedule(this, 1, TimeUnit.HOURS)
-                  return
-                } else {
-                  logWarning("Error while attempting to clean up old delegation token files. " +
-                    "Cleanup will be reattempted the next time new tokens are being written.")
-                }
-            }
-            scheduleRenewal(this)
+    // This thread periodically runs on the driver to update the delegation tokens on HDFS.
+    val driverTokenRenewerRunnable =
+      new Runnable {
+        override def run(): Unit = {
+          try {
+            writeNewTokensToHDFS(principal, keytab)
+            cleanupOldFiles()
+          } catch {
+            case e: Exception =>
+              // Log the error and try to write new tokens back in an hour
+              logWarning("Failed to write out new credentials to HDFS, will try again in an " +
+                "hour! If this happens too often tasks will fail.", e)
+              delegationTokenRenewer.schedule(this, 1, TimeUnit.HOURS)
+              return
           }
+          scheduleRenewal(this)
         }
-      // Schedule update of credentials. This handles the case of updating the tokens right now
-      // as well, since the renenwal interval will be 0, and the thread will get scheduled
-      // immediately.
-      scheduleRenewal(driverTokenRenewerRunnable)
-
-    }
+      }
+    // Schedule update of credentials. This handles the case of updating the tokens right now
+    // as well, since the renenwal interval will be 0, and the thread will get scheduled
+    // immediately.
+    scheduleRenewal(driverTokenRenewerRunnable)
   }
 
   // Keeps only files that are newer than 30 days, and deletes everything else. At least 5 files
@@ -135,9 +124,11 @@ class AMDelegationTokenRenewer(sparkConf: SparkConf, hadoopConf: Configuration)
     try {
       val remoteFs = FileSystem.get(hadoopConf)
       val credentialsPath = new Path(sparkConf.get("spark.yarn.credentials.file"))
+      val thresholdTime = System.currentTimeMillis() - (30 days).toMillis
       hadoopUtil.listFilesSorted(
-        remoteFs, credentialsPath.getParent, credentialsPath.getName, ".tmp").dropRight(5)
-        .takeWhile(_.getModificationTime < System.currentTimeMillis() - (30 days).toMillis)
+        remoteFs, credentialsPath.getParent, credentialsPath.getName, ".tmp")
+        .dropRight(5)
+        .takeWhile(_.getModificationTime < thresholdTime)
         .foreach(x => remoteFs.delete(x.getPath, true))
     } catch {
       // Such errors are not fatal, so don't throw. Make sure they are logged though

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

@@ -76,7 +76,7 @@ private[spark] class ApplicationMaster(
   // Fields used in cluster mode.
   private val sparkContextRef = new AtomicReference[SparkContext](null)
 
-  private lazy val delegationTokenRenewer = new AMDelegationTokenRenewer(sparkConf, yarnConf)
+  private var delegationTokenRenewerOption: Option[AMDelegationTokenRenewer] = None
 
   final def run(): Int = {
     try {
@@ -140,6 +140,12 @@ private[spark] class ApplicationMaster(
       // doAs in order for the credentials to be passed on to the executor containers.
       val securityMgr = new SecurityManager(sparkConf)
 
+      // If the credentials file config is present, we must periodically renew tokens. So create
+      // a new AMDelegationTokenRenewer
+      if(sparkConf.contains("spark.yarn.credentials.file")) {
+        delegationTokenRenewerOption = Some(new AMDelegationTokenRenewer(sparkConf, yarnConf))
+      }
+
       if (isClusterMode) {
         runDriver(securityMgr)
       } else {
@@ -204,7 +210,7 @@ private[spark] class ApplicationMaster(
           logDebug("shutting down user thread")
           userClassThread.interrupt()
         }
-        if (!inShutdown) delegationTokenRenewer.stop()
+        if (!inShutdown) delegationTokenRenewerOption.foreach(_.stop())
       }
     }
   }
@@ -254,7 +260,7 @@ private[spark] class ApplicationMaster(
       SparkEnv.driverActorSystemName,
       RpcAddress(host, port.toInt),
       YarnSchedulerBackend.ENDPOINT_NAME)
-    delegationTokenRenewer.driverEndPoint = driverEndpoint
+    delegationTokenRenewerOption.foreach(_.driverEndPoint = driverEndpoint)
     amEndpoint =
       rpcEnv.setupEndpoint("YarnAM", new AMEndpoint(rpcEnv, driverEndpoint, isClusterMode))
   }
@@ -281,7 +287,7 @@ private[spark] class ApplicationMaster(
       registerAM(sc.ui.map(_.appUIAddress).getOrElse(""), securityMgr)
       // If a principal and keytab have been set, use that to create new credentials for executors
       // periodically
-      delegationTokenRenewer.scheduleLoginFromKeytab()
+      delegationTokenRenewerOption.foreach(_.scheduleLoginFromKeytab())
       userClassThread.join()
     }
   }
@@ -292,7 +298,7 @@ private[spark] class ApplicationMaster(
     addAmIpFilter()
     // If a principal and keytab have been set, use that to create new credentials for executors
     // periodically
-    delegationTokenRenewer.scheduleLoginFromKeytab()
+    delegationTokenRenewerOption.foreach(_.scheduleLoginFromKeytab())
     registerAM(sparkConf.get("spark.driver.appUIAddress", ""), securityMgr)
 
     // In client mode the actor will stop the reporter thread.