Adding logging, fixing minor formatting and ordering issues.

Author: harishreedharan

launcher/src/main/java/org/apache/spark/launcher/SparkSubmitOptionParser.java

@@ -57,8 +57,6 @@ class SparkSubmitOptionParser {
   protected final String REPOSITORIES = "--repositories";
   protected final String STATUS = "--status";
   protected final String TOTAL_EXECUTOR_CORES = "--total-executor-cores";
-  protected final String PRINCIPAL = "--principal";
-  protected final String KEYTAB = "--keytab";
 
   // Options that do not take arguments.
   protected final String HELP = "--help";
@@ -71,8 +69,10 @@ class SparkSubmitOptionParser {
   // YARN-only options.
   protected final String ARCHIVES = "--archives";
   protected final String EXECUTOR_CORES = "--executor-cores";
-  protected final String QUEUE = "--queue";
+  protected final String KEYTAB = "--keytab";
   protected final String NUM_EXECUTORS = "--num-executors";
+  protected final String PRINCIPAL = "--principal";
+  protected final String QUEUE = "--queue";
 
   /**
    * This is the canonical list of spark-submit options. Each entry in the array contains the
@@ -98,20 +98,20 @@ class SparkSubmitOptionParser {
     { EXECUTOR_MEMORY },
     { FILES },
     { JARS },
+    { KEYTAB },
     { KILL_SUBMISSION },
     { MASTER },
     { NAME },
     { NUM_EXECUTORS },
     { PACKAGES },
+    { PRINCIPAL },
     { PROPERTIES_FILE },
     { PROXY_USER },
     { PY_FILES },
     { QUEUE },
     { REPOSITORIES },
     { STATUS },
     { TOTAL_EXECUTOR_CORES },
-    { PRINCIPAL},
-    { KEYTAB}
   };
 
   /**

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

@@ -26,7 +26,7 @@ import scala.collection.JavaConversions._
 import scala.collection.mutable.{ArrayBuffer, HashMap, ListBuffer, Map}
 import scala.util.{Try, Success, Failure}
 
-import com.google.common.base.{Objects, Preconditions}
+import com.google.common.base.Objects
 import org.apache.hadoop.io.DataOutputBuffer
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs._
@@ -238,6 +238,8 @@ private[spark] class Client(
     // If we passed in a keytab, make sure we copy the keytab to the staging directory on
     // HDFS, and setup the relevant environment vars, so the AM can login again.
     if (loginFromKeytab) {
+      logInfo("To enable the AM to login from keytab, credentials are being copied over to the AM" +
+        " via the YARN Secure Distributed Cache.")
       val localUri = new URI(args.keytab)
       val localPath = getQualifiedLocalPath(localUri, hadoopConf)
       val destinationPath = copyFileToRemote(dst, localPath, replication)
@@ -333,6 +335,7 @@ private[spark] class Client(
       val credentialsFile = "credentials-" + UUID.randomUUID().toString
       sparkConf.set(
         "spark.yarn.credentials.file", new Path(stagingDirPath, credentialsFile).toString)
+      logInfo(s"Credentials file set to: $credentialsFile")
     }
 
     // Set the environment variables to be passed on to the executors.
@@ -562,8 +565,7 @@ private[spark] class Client(
 
   def setupCredentials(): Unit = {
     if (args.principal != null) {
-      Preconditions.checkNotNull(
-        args.keytab, "Keytab must be specified when principal is specified.", Array.empty[Any])
+      require(args.keytab != null, "Keytab must be specified when principal is specified.")
       logInfo("Attempting to login to the Kerberos" +
         s" using principal: ${args.principal} and keytab: ${args.keytab}")
       val f = new File(args.keytab)

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

@@ -126,6 +126,13 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
   private[spark] override def scheduleLoginFromKeytab(): Unit = {
     sparkConf.getOption("spark.yarn.principal").foreach { principal =>
       val keytab = sparkConf.get("spark.yarn.keytab")
+
+      def scheduleRenewal(runnable: Runnable) = {
+        val renewalInterval = (0.75 * (getLatestValidity - System.currentTimeMillis())).toLong
+        logInfo("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 {
@@ -139,23 +146,21 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
                 delegationTokenRenewer.schedule(this, 1, TimeUnit.HOURS)
                 return
             }
-            delegationTokenRenewer.schedule(
-              this, (0.75 * (getLatestValidity - System.currentTimeMillis())).toLong,
-              TimeUnit.MILLISECONDS)
+            scheduleRenewal(this)
           }
         }
-      val timeToRenewal = (0.75 * (getLatestValidity - System.currentTimeMillis())).toLong
-      delegationTokenRenewer.schedule(
-        driverTokenRenewerRunnable, timeToRenewal, TimeUnit.MILLISECONDS)
+      scheduleRenewal(driverTokenRenewerRunnable)
     }
   }
 
   private def renewCredentials(principal: String, keytab: String): Unit = {
     if (!loggedInViaKeytab) {
       // Keytab is copied by YARN to the working directory of the AM, so full path is
       // not needed.
+      logInfo(s"Attempting to login to KDC using principal: $principal")
       loggedInUGI = UserGroupInformation.loginUserFromKeytabAndReturnUGI(
         principal, keytab)
+      logInfo("Successfully logged into KDC.")
       loggedInViaKeytab = true
     }
     val nns = getNameNodesToAccess(sparkConf)
@@ -167,13 +172,16 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
       sparkConf.get("spark.yarn.credentials.file") + "-" + nextSuffix
     val tokenPath = new Path(tokenPathStr)
     val tempTokenPath = new Path(tokenPathStr + ".tmp")
+    logInfo("Writing out delegation tokens to " + tempTokenPath.toString)
     val stream = Option(remoteFs.create(tempTokenPath, true))
     try {
       stream.foreach { s =>
         newCredentials.writeTokenStorageToStream(s)
         s.hflush()
         s.close()
+        logInfo(s"Delegation Tokens written out successfully. Renaming file to $tokenPathStr")
         remoteFs.rename(tempTokenPath, tokenPath)
+        logInfo("Delegation token file rename complete.")
       }
     } finally {
       stream.foreach(_.close())
@@ -205,16 +213,23 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
           val credentials = credentialsStatus.getPath
           val suffix = credentials.getName.substring(credentials.getName.lastIndexOf("-") + 1).toInt
           if (suffix > lastCredentialsFileSuffix) {
+            logInfo("Reading new delegation tokens from " + credentials.toString)
             val newCredentials = getCredentialsFromHDFSFile(remoteFs, credentials)
+            lastCredentialsFileSuffix = suffix
             UserGroupInformation.getCurrentUser.addCredentials(newCredentials)
+
             val totalValidity = getLatestValidity - credentialsStatus.getModificationTime
             val timeToRunRenewal =
               credentialsStatus.getModificationTime + (0.8 * totalValidity).toLong
             val timeFromNowToRenewal = timeToRunRenewal - System.currentTimeMillis()
-            delegationTokenRenewer.schedule(executorUpdaterRunnable,
-              timeFromNowToRenewal, TimeUnit.MILLISECONDS)
+            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)
           }
         }
@@ -223,8 +238,7 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
       // Since the file may get deleted while we are reading it, catch the Exception and come
       // back in an hour to try again
       case e: Exception =>
-        logWarning(
-          "Error encountered while trying to update credentials, will try again in 1 hour", e)
+        logWarning("Error while trying to update credentials, will try again in 1 hour", e)
         delegationTokenRenewer.schedule(executorUpdaterRunnable, 1, TimeUnit.HOURS)
     }
   }