Add different severity levels

With this commit, the HistoryNotFoundPage behaves differently for
each of the following scenarios:
  (1) User did not enable event logging -> point them to the configs
  (2) Event logs are not found for some reason -> tell them that
  (3) Exception is thrown in replaying -> show them the stack trace

core/src/main/scala/org/apache/spark/TaskEndReason.scala

@@ -20,6 +20,7 @@ package org.apache.spark
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.storage.BlockManagerId
+import org.apache.spark.util.Utils
 
 /**
  * :: DeveloperApi ::
@@ -88,11 +89,7 @@ case class ExceptionFailure(
     stackTrace: Array[StackTraceElement],
     metrics: Option[TaskMetrics])
   extends TaskFailedReason {
-  override def toErrorString: String = {
-    val stackTraceString =
-      if (stackTrace == null) "null" else stackTrace.map("        " + _).mkString("\n")
-    s"$className ($description)\n$stackTraceString"
-  }
+  override def toErrorString: String = Utils.exceptionString(className, description, stackTrace)
 }
 
 /**

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.deploy.master
 
+import java.net.URLEncoder
 import java.text.SimpleDateFormat
 import java.util.Date
 
@@ -30,7 +31,6 @@ import akka.actor._
 import akka.pattern.ask
 import akka.remote.{DisassociatedEvent, RemotingLifecycleEvent}
 import akka.serialization.SerializationExtension
-import org.apache.hadoop.fs.FileSystem
 
 import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkException}
 import org.apache.spark.deploy.{ApplicationDescription, DriverDescription, ExecutorState}
@@ -642,10 +642,7 @@ private[spark] class Master(
       waitingApps -= app
 
       // If application events are logged, use them to rebuild the UI
-      if (!rebuildSparkUI(app)) {
-        // Avoid broken links if the UI is not reconstructed
-        app.desc.appUiUrl = "/history/not-found"
-      }
+      rebuildSparkUI(app)
 
       for (exec <- app.executors.values) {
         exec.worker.removeExecutor(exec)
@@ -667,29 +664,47 @@ private[spark] class Master(
    */
   def rebuildSparkUI(app: ApplicationInfo): Boolean = {
     val appName = app.desc.name
-    val eventLogDir = app.desc.eventLogDir.getOrElse { return false }
+    val eventLogDir = app.desc.eventLogDir.getOrElse {
+      // Event logging is not enabled for this application
+      app.desc.appUiUrl = "/history/not-found"
+      return false
+    }
     val fileSystem = Utils.getHadoopFileSystem(eventLogDir)
     val eventLogInfo = EventLoggingListener.parseLoggingInfo(eventLogDir, fileSystem)
     val eventLogPaths = eventLogInfo.logPaths
     val compressionCodec = eventLogInfo.compressionCodec
-    if (!eventLogPaths.isEmpty) {
-      try {
-        val replayBus = new ReplayListenerBus(eventLogPaths, fileSystem, compressionCodec)
-        val ui = new SparkUI(
-          new SparkConf, replayBus, appName + " (completed)", "/history/" + app.id)
-        replayBus.replay()
-        app.desc.appUiUrl = ui.basePath
-        appIdToUI(app.id) = ui
-        webUi.attachSparkUI(ui)
-        return true
-      } catch {
-        case e: Exception =>
-          logError("Exception in replaying log for application %s (%s)".format(appName, app.id), e)
-      }
-    } else {
-      logWarning("Application %s (%s) has no valid logs: %s".format(appName, app.id, eventLogDir))
+
+    if (eventLogPaths.isEmpty) {
+      // Event logging is enabled for this application, but no event logs are found
+      val title = s"Application history not found (${app.id})"
+      var msg = s"No event logs found for application $appName in $eventLogDir."
+      logWarning(msg)
+      msg += " Did you specify the correct logging directory?"
+      msg = URLEncoder.encode(msg, "UTF-8")
+      app.desc.appUiUrl = s"/history/not-found?msg=$msg&title=$title"
+      return false
+    }
+
+    try {
+      val replayBus = new ReplayListenerBus(eventLogPaths, fileSystem, compressionCodec)
+      val ui = new SparkUI(new SparkConf, replayBus, appName + " (completed)", "/history/" + app.id)
+      replayBus.replay()
+      appIdToUI(app.id) = ui
+      webUi.attachSparkUI(ui)
+      // Application UI is successfully rebuilt, so link the Master UI to it
+      app.desc.appUiUrl = ui.basePath
+      true
+    } catch {
+      case e: Exception =>
+        // Relay exception message to application UI page
+        val title = s"Application history load error (${app.id})"
+        val exception = URLEncoder.encode(Utils.exceptionString(e), "UTF-8")
+        var msg = s"Exception in replaying log for application $appName!"
+        logError(msg, e)
+        msg = URLEncoder.encode(msg, "UTF-8")
+        app.desc.appUiUrl = s"/history/not-found?msg=$msg&exception=$exception&title=$title"
+        false
     }
-    false
   }
 
   /** Generate a new app ID given a app's submission date */

core/src/main/scala/org/apache/spark/deploy/master/ui/HistoryNotFoundPage.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.deploy.master.ui
 
+import java.net.URLDecoder
 import javax.servlet.http.HttpServletRequest
 
 import scala.xml.Node
@@ -26,17 +27,47 @@ import org.apache.spark.ui.{UIUtils, WebUIPage}
 private[spark] class HistoryNotFoundPage(parent: MasterWebUI)
   extends WebUIPage("history/not-found") {
 
+  /**
+   * Render a page that conveys failure in loading application history.
+   *
+   * This accepts 3 HTTP parameters:
+   *   msg = message to display to the user
+   *   title = title of the page
+   *   exception = detailed description of the exception in loading application history (if any)
+   *
+   * Parameters "msg" and "exception" are assumed to be UTF-8 encoded.
+   */
   def render(request: HttpServletRequest): Seq[Node] = {
-    val content =
+    val titleParam = request.getParameter("title")
+    val msgParam = request.getParameter("msg")
+    val exceptionParam = request.getParameter("exception")
+
+    // If no parameters are specified, assume the user did not enable event logging
+    val defaultTitle = "Event logging is not enabled"
+    val defaultContent =
       <div class="row-fluid">
         <div class="span12" style="font-size:14px">
-          No event logs were found for this application. To
+          No event logs were found for this application! To
           <a href="http://spark.apache.org/docs/latest/monitoring.html">enable event logging</a>,
-          please set <span style="font-style:italic">spark.eventLog.enabled</span> to true and
+          set <span style="font-style:italic">spark.eventLog.enabled</span> to true and
           <span style="font-style:italic">spark.eventLog.dir</span> to the directory to which your
           event logs are written.
         </div>
       </div>
-    UIUtils.basicSparkPage(content, "Application history not found")
+
+    val title = Option(titleParam).getOrElse(defaultTitle)
+    val content = Option(msgParam)
+      .map { msg => URLDecoder.decode(msg, "UTF-8") }
+      .map { msg =>
+        <div class="row-fluid">
+          <div class="span12" style="font-size:14px">{msg}</div>
+        </div> ++
+        Option(exceptionParam)
+          .map { e => URLDecoder.decode(e, "UTF-8") }
+          .map { e => <pre>{e}</pre> }
+          .getOrElse(Seq.empty)
+      }.getOrElse(defaultContent)
+
+    UIUtils.basicSparkPage(content, title)
   }
 }

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -44,7 +44,7 @@ import org.apache.spark.executor.ExecutorUncaughtExceptionHandler
 import org.apache.spark.serializer.{DeserializationStream, SerializationStream, SerializerInstance}
 
 /** CallSite represents a place in user code. It can have a short and a long form. */
-private[spark] case class CallSite(val short: String, val long: String)
+private[spark] case class CallSite(short: String, long: String)
 
 /**
  * Various utility methods used by Spark.
@@ -1291,4 +1291,19 @@ private[spark] object Utils extends Logging {
     }
   }
 
+  /** Return a nice string representation of the exception, including the stack trace. */
+  def exceptionString(e: Exception): String = {
+    if (e == null) "" else exceptionString(getFormattedClassName(e), e.getMessage, e.getStackTrace)
+  }
+
+  /** Return a nice string representation of the exception, including the stack trace. */
+  def exceptionString(
+      className: String,
+      description: String,
+      stackTrace: Array[StackTraceElement]): String = {
+    val desc = if (description == null) "" else description
+    val st = if (stackTrace == null) "" else stackTrace.map("        " + _).mkString("\n")
+    s"$className: $desc\n$st"
+  }
+
 }