Show prettier name in UI.

Work around a SparkUI issue where the name to show has to be
provided in the constructor.

Also remove explicit flushes from logging code, since they're
not really useful now that the HS only reads data from finished
apps (and the API used does not exist in Hadoop trunk).

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -163,7 +163,7 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
           fs.exists(new Path(entry.getPath(), APPLICATION_COMPLETE))
         } else {
           try {
-            val matcher(version, codecName, inprogress) = entry.getPath().getName()
+            val matcher(_, _, version, codecName, inprogress) = entry.getPath().getName()
             inprogress == null
           } catch {
             case e: Exception => false
@@ -247,32 +247,37 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
       return (null, null)
     }
 
-    val (logFile, lastUpdated) = if (log.isFile()) {
-        (elogInfo.path, log.getModificationTime())
+    val (appName, logFile, lastUpdated) = if (log.isFile()) {
+        // Due to SPARK-2169, we need to provide the name of the application to be shown in
+        // the SparkUI constructor; setting it afterwards has no effect. So get the portion
+        // of the log file name that contains that information.
+        val EventLoggingListener.LOG_FILE_NAME_REGEX(appName, timestamp, _, _, _) =
+          log.getPath().getName()
+
+        (s"$appName-$timestamp", elogInfo.path, log.getModificationTime())
       } else {
         // For old-style log directories, need to find the actual log file.
         val status = fs.listStatus(elogInfo.path)
           .filter(e => e.getPath().getName().startsWith(LOG_PREFIX))(0)
-        (status.getPath(), status.getModificationTime())
+        (elogInfo.path.getName(), status.getPath(), status.getModificationTime())
       }
 
-    val appId = elogInfo.path.getName
     val replayBus = new ReplayListenerBus(logFile, fs, elogInfo.compressionCodec)
     val appListener = new ApplicationEventListener
     replayBus.addListener(appListener)
 
     val ui: SparkUI = if (renderUI) {
         val conf = this.conf.clone()
         val appSecManager = new SecurityManager(conf)
-        new SparkUI(conf, appSecManager, replayBus, appId, "/history/" + appId)
+        new SparkUI(conf, appSecManager, replayBus, appName, "/history/" + elogInfo.path.getName())
         // Do not call ui.bind() to avoid creating a new server for each application
       } else {
         null
       }
 
     replayBus.replay()
     val appInfo = ApplicationHistoryInfo(
-      appId,
+      elogInfo.path.getName(),
       appListener.appName,
       appListener.startTime,
       appListener.endTime,

core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala

@@ -32,12 +32,7 @@ import org.json4s.jackson.JsonMethods._
 import org.apache.spark.{Logging, SparkConf, SparkContext}
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.io.CompressionCodec
-<<<<<<< HEAD
-import org.apache.spark.SPARK_VERSION
-import org.apache.spark.util.{FileLogger, JsonProtocol, Utils}
-=======
 import org.apache.spark.util.{JsonProtocol, Utils}
->>>>>>> Make event logger use a single file.
 
 /**
  * A SparkListener that logs events to persistent storage.
@@ -65,17 +60,11 @@ private[spark] class EventLoggingListener(
   private val testing = sparkConf.getBoolean("spark.eventLog.testing", false)
   private val outputBufferSize = sparkConf.getInt("spark.eventLog.buffer.kb", 100) * 1024
   private val logBaseDir = sparkConf.get("spark.eventLog.dir", DEFAULT_LOG_DIR).stripSuffix("/")
-<<<<<<< HEAD
-  private val name = appName.replaceAll("[ :/]", "-").replaceAll("[${}'\"]", "_")
-    .toLowerCase + "-" + System.currentTimeMillis
-  val logDir = Utils.resolveURI(logBaseDir) + "/" + name.stripSuffix("/")
-=======
   private val fileSystem = Utils.getHadoopFileSystem(new URI(logBaseDir))
   private lazy val compressionCodec = CompressionCodec.createCodec(sparkConf)
 
   // Only defined if the file system scheme is not local
   private var hadoopDataStream: Option[FSDataOutputStream] = None
->>>>>>> Make event logger use a single file.
 
   private var writer: Option[PrintWriter] = None
 
@@ -100,17 +89,6 @@ private[spark] class EventLoggingListener(
     }
 
   /**
-<<<<<<< HEAD
-   * Return only the unique application directory without the base directory.
-   */
-  def getApplicationLogDir(): String = {
-    name
-  }
-
-  /**
-   * Begin logging events.
-   * If compression is used, log a file that indicates which compression library is used.
-=======
    * Creates the log file in the configured log directory.
    *
    * The file name contains some metadata about its contents. It follows the following
@@ -128,16 +106,11 @@ private[spark] class EventLoggingListener(
    *   used to write the file
    * - ".inprogress" will be present while the log file is still being written to, and
    *   removed after the application is finished.
->>>>>>> Make event logger use a single file.
    */
   def start() {
     if (!fileSystem.isDirectory(new Path(logBaseDir))) {
       throw new IllegalArgumentException(s"Log directory $logBaseDir does not exist.");
     }
-<<<<<<< HEAD
-    logger.newFile(SPARK_VERSION_PREFIX + SPARK_VERSION)
-    logger.newFile(LOG_PREFIX + logger.fileIndex)
-=======
 
     val workingPath = logPath + IN_PROGRESS
     val uri = new URI(workingPath)
@@ -162,25 +135,17 @@ private[spark] class EventLoggingListener(
     writer = Some(new PrintWriter(cstream))
 
     logInfo("Logging events to %s".format(logPath))
->>>>>>> Make event logger use a single file.
   }
 
   /** Log the event as JSON. */
-  private def logEvent(event: SparkListenerEvent, flushLogger: Boolean = false) {
+  private def logEvent(event: SparkListenerEvent) {
     val eventJson = JsonProtocol.sparkEventToJson(event)
-
     writer.foreach(_.println(compact(render(eventJson))))
-    if (flushLogger) {
-      writer.foreach(_.flush())
-      hadoopDataStream.foreach(_.sync())
-    }
-
     if (testing) {
       loggedEvents += eventJson
     }
   }
 
-  // Events that do not trigger a flush
   override def onStageSubmitted(event: SparkListenerStageSubmitted) =
     logEvent(event)
   override def onTaskStart(event: SparkListenerTaskStart) =
@@ -191,24 +156,22 @@ private[spark] class EventLoggingListener(
     logEvent(event)
   override def onEnvironmentUpdate(event: SparkListenerEnvironmentUpdate) =
     logEvent(event)
-
-  // Events that trigger a flush
   override def onStageCompleted(event: SparkListenerStageCompleted) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onJobStart(event: SparkListenerJobStart) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onJobEnd(event: SparkListenerJobEnd) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onBlockManagerAdded(event: SparkListenerBlockManagerAdded) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onBlockManagerRemoved(event: SparkListenerBlockManagerRemoved) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onUnpersistRDD(event: SparkListenerUnpersistRDD) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onApplicationStart(event: SparkListenerApplicationStart) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   override def onApplicationEnd(event: SparkListenerApplicationEnd) =
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   // No-op because logging every update would be overkill
   override def onExecutorMetricsUpdate(event: SparkListenerExecutorMetricsUpdate) { }
 
@@ -234,7 +197,7 @@ private[spark] object EventLoggingListener extends Logging {
   val LOG_FILE_PERMISSIONS = FsPermission.createImmutable(Integer.parseInt("770", 8).toShort)
 
   // Regex for parsing log file names. See description of log file name format in start().
-  val LOG_FILE_NAME_REGEX = s".+-[0-9]+-([0-9](?:\\.[0-9])*)(?:-(.+?))?(\\$IN_PROGRESS)?".r
+  val LOG_FILE_NAME_REGEX = s"(.+)-([0-9]+)-([0-9](?:\\.[0-9])*)(?:-(.+?))?(\\$IN_PROGRESS)?".r
 
   // A cache for compression codecs to avoid creating the same codec many times
   private val codecMap = new mutable.HashMap[String, CompressionCodec]
@@ -248,7 +211,7 @@ private[spark] object EventLoggingListener extends Logging {
    */
   def parseLoggingInfo(log: Path): EventLoggingInfo = {
     try {
-      val LOG_FILE_NAME_REGEX(version, codecName, inprogress) = log.getName()
+      val LOG_FILE_NAME_REGEX(_, _, version, codecName, inprogress) = log.getName()
       val codec: Option[CompressionCodec] = if (codecName != null) {
           val conf = new SparkConf()
           conf.set("spark.io.compression.codec", codecName)
@@ -259,13 +222,8 @@ private[spark] object EventLoggingListener extends Logging {
       EventLoggingInfo(log, version, codec, inprogress == null)
     } catch {
       case e: Exception =>
-<<<<<<< HEAD
-        logError("Exception in parsing logging info from directory %s".format(logDir), e)
-        EventLoggingInfo.empty
-=======
         logError("Exception in parsing logging info from file %s".format(log), e)
         null
->>>>>>> Make event logger use a single file.
     }
   }
 

core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala

@@ -23,14 +23,14 @@ import com.google.common.io.Files
 import org.apache.hadoop.fs.Path
 import org.json4s.jackson.JsonMethods._
 import org.scalatest.{BeforeAndAfter, FunSuite}
-import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.Matchers
 
 import org.apache.spark.SparkConf
 import org.apache.spark.io._
 import org.apache.spark.scheduler._
 import org.apache.spark.util.{JsonProtocol, Utils}
 
-class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with ShouldMatchers {
+class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with Matchers {
 
   private var testDir: File = null
 
@@ -54,15 +54,15 @@ class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with ShouldMat
     val provider = new FsHistoryProvider(conf)
 
     // Write a new-style application log.
-    val logFile1 = new File(testDir, "app1-1-1.0")
+    val logFile1 = new File(testDir, "app1-1-2-1.0")
     writeFile(logFile1,
-      SparkListenerApplicationStart("app1", 1L, "test"),
+      SparkListenerApplicationStart("app1-1", 1L, "test"),
       SparkListenerApplicationEnd(2L)
       )
 
     // Write an unfinished app, new-style.
-    writeFile(new File(testDir, "app2-1-1.0.inprogress"),
-      SparkListenerApplicationStart("app2", 1L, "test")
+    writeFile(new File(testDir, "app2-2-1-1.0.inprogress"),
+      SparkListenerApplicationStart("app2-2", 1L, "test")
       )
 
     // Write an old-style application log.
@@ -93,7 +93,7 @@ class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with ShouldMat
 
     list(0) should be (ApplicationHistoryInfo(oldLog.getName(), "app3", 2L, 3L,
       oldLog.lastModified(), "test"))
-    list(1) should be (ApplicationHistoryInfo(logFile1.getName(), "app1", 1L, 2L,
+    list(1) should be (ApplicationHistoryInfo(logFile1.getName(), "app1-1", 1L, 2L,
       logFile1.lastModified(), "test"))
 
     // Make sure the UI can be rendered.