Address review feedback.

Main changes:
- Restore old command line handling.
- Fix pagination.
- Restore showing the log directory in the listing page.

Author: Marcelo Vanzin

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

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.history
+
+import org.apache.spark.ui.SparkUI
+
+private[spark] case class ApplicationHistoryInfo(
+    id: String,
+    name: String,
+    startTime: Long,
+    endTime: Long,
+    lastUpdated: Long,
+    sparkUser: String,
+    viewAcls: String,
+    ui: SparkUI) {
+}
+
+private[spark] abstract class ApplicationHistoryProvider {
+
+  /**
+   * This method should return a list of applications available for the history server to
+   * show.
+   *
+   * @return List of all know applications.
+   */
+  def getListing(): Seq[ApplicationHistoryInfo]
+
+  /**
+   * This method should return the application information, including a rendered SparkUI.
+   *
+   * @param appId The application ID.
+   * @return The app info, or null if not found.
+   */
+  def getAppInfo(appId: String): ApplicationHistoryInfo
+
+  /**
+   * Called when the server is shutting down.
+   */
+  def stop(): Unit = { }
+
+  /**
+   * Returns configuration data to be shown in the HS home page.
+   *
+   * @return A map with the configuration data. Data is show in the order returned by the map.
+   */
+  def getConfig(): Map[String, String] = Map()
+
+}

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

@@ -32,7 +32,8 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
   with Logging {
 
   // Interval between each check for event log updates
-  private val UPDATE_INTERVAL_MS = conf.getInt("spark.history.fs.updateInterval", 10) * 1000
+  private val UPDATE_INTERVAL_MS = conf.getInt("spark.history.fs.updateInterval",
+    conf.getInt("spark.history.updateInterval", 10)) * 1000
 
   private val logDir = conf.get("spark.history.fs.logDirectory")
   private val fs = Utils.getHadoopFileSystem(logDir)
@@ -54,7 +55,7 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
   private val logCheckingThread = new Thread("LogCheckingThread") {
     override def run() = Utils.logUncaughtExceptions {
       while (true) {
-        val now = getMonotonicTime()
+        val now = getMonotonicTimeMs()
         if (now - lastLogCheckTimeMs > UPDATE_INTERVAL_MS) {
           Thread.sleep(UPDATE_INTERVAL_MS)
         } else {
@@ -97,13 +98,16 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
     }
   }
 
+  override def getConfig(): Map[String, String] =
+    Map(("Event Log Location" -> logDir))
+
   /**
    * Builds the application list based on the current contents of the log directory.
    * Tries to reuse as much of the data already in memory as possible, by not reading
    * applications that haven't been updated since last time the logs were checked.
    */
-  def checkForLogs() = {
-    lastLogCheckTimeMs = getMonotonicTime()
+  private def checkForLogs() = {
+    lastLogCheckTimeMs = getMonotonicTimeMs()
     logDebug("Checking for logs. Time is now %d.".format(lastLogCheckTimeMs))
     try {
       val logStatus = fs.listStatus(new Path(logDir))
@@ -142,6 +146,14 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
   /**
    * Parse the application's logs to find out the information we need to build the
    * listing page.
+   *
+   * When creating the listing of available apps, there is no need to load the whole UI for the
+   * application. The UI is requested by the HistoryServer (by calling getAppInfo()) when the user
+   * clicks on a specific application.
+   *
+   * @param logDir Directory with application's log files.
+   * @param renderUI Whether to create the SparkUI for the application. If false, the "ui"
+   *                 attribute of the returned object will be null.
    */
   private def loadAppInfo(logDir: FileStatus, renderUI: Boolean): ApplicationHistoryInfo = {
     val elogInfo = EventLoggingListener.parseLoggingInfo(logDir.getPath(), fs)
@@ -188,6 +200,6 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
   }
 
   /** Returns the system's mononotically increasing time. */
-  private def getMonotonicTime() = System.nanoTime() / (1000 * 1000)
+  private def getMonotonicTimeMs() = System.nanoTime() / (1000 * 1000)
 
 }

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

@@ -25,24 +25,28 @@ import org.apache.spark.ui.{WebUIPage, UIUtils}
 
 private[spark] class HistoryPage(parent: HistoryServer) extends WebUIPage("") {
 
-  val pageSize = 20
+  private val pageSize = 20
 
   def render(request: HttpServletRequest): Seq[Node] = {
     val requestedPage = Option(request.getParameter("page")).getOrElse("1").toInt
     val requestedFirst = (requestedPage - 1) * pageSize
 
     val allApps = parent.getApplicationList()
     val actualFirst = if (requestedFirst < allApps.size) requestedFirst else 0
-    val apps = allApps.slice(actualFirst, Math.min(pageSize, allApps.size))
+    val apps = allApps.slice(actualFirst, Math.min(actualFirst + pageSize, allApps.size))
 
     val actualPage = (actualFirst / pageSize) + 1
     val last = Math.min(actualFirst + pageSize, allApps.size) - 1
     val pageCount = allApps.size / pageSize + (if (allApps.size % pageSize > 0) 1 else 0)
 
     val appTable = UIUtils.listingTable(appHeader, appRow, apps)
+    val providerConfig = parent.getProviderConfig()
     val content =
       <div class="row-fluid">
         <div class="span12">
+          <ul class="unstyled">
+            { providerConfig.map(e => <li><strong>{e._1}:</strong> {e._2}</li>) }
+          </ul>
           {
             if (allApps.size > 0) {
               <h4>

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

@@ -54,7 +54,6 @@ class HistoryServer(
   private val uiAclsEnabled = conf.getBoolean("spark.history.ui.acls.enable", false)
 
   private val localHost = Utils.localHostName()
-  private val publicHost = Option(System.getenv("SPARK_PUBLIC_DNS")).getOrElse(localHost)
 
   private val appLoader = new CacheLoader[String, SparkUI] {
     override def load(key: String): SparkUI = {
@@ -80,13 +79,14 @@ class HistoryServer(
 
   private val loaderServlet = new HttpServlet {
     protected override def doGet(req: HttpServletRequest, res: HttpServletResponse): Unit = {
-      val parts = req.getPathInfo().split("/")
+      val parts = Option(req.getPathInfo()).getOrElse("").split("/")
       if (parts.length < 2) {
-        res.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+        res.sendError(HttpServletResponse.SC_BAD_REQUEST,
+          s"Unexpected path info in request (URI = ${req.getRequestURI()}")
         return
       }
 
-      var appId = parts(1)
+      val appId = parts(1)
 
       // Note we don't use the UI retrieved from the cache; the cache loader above will register
       // the app's UI, and all we need to do is redirect the user to the same URI that was
@@ -157,6 +157,13 @@ class HistoryServer(
    */
   def getApplicationList() = provider.getListing()
 
+  /**
+   * Returns the provider configuration to show in the listing page.
+   *
+   * @return A map with the provider's configuration.
+   */
+  def getProviderConfig() = provider.getConfig()
+
 }
 
 /**
@@ -174,7 +181,7 @@ object HistoryServer {
 
   def main(argStrings: Array[String]) {
     initSecurity()
-    parse(argStrings.toList)
+    val args = new HistoryServerArguments(conf, argStrings)
     val securityManager = new SecurityManager(conf)
 
     val providerName = conf.getOption("spark.history.provider")
@@ -212,89 +219,4 @@ object HistoryServer {
     }
   }
 
-  private def parse(args: List[String]): Unit = {
-    args match {
-      case ("--dir" | "-d") :: value :: tail =>
-        set("fs.logDirectory",  value)
-        parse(tail)
-
-      case ("-D") :: opt :: value :: tail =>
-        set(opt, value)
-        parse(tail)
-
-      case ("--help" | "-h") :: tail =>
-        printUsageAndExit(0)
-
-      case Nil =>
-
-      case _ =>
-        printUsageAndExit(1)
-    }
-  }
-
-  private def set(name: String, value: String) = {
-    conf.set("spark.history." + name, value)
-  }
-
-  private def printUsageAndExit(exitCode: Int) {
-    System.err.println(
-      """
-      |Usage: HistoryServer [options]
-      |
-      |Options are set by passing "-D option value" command line arguments to the class.
-      |Command line options will override JVM system properties (which should be prepended
-      |with "spark.history.").
-      |
-      |History Server options are always available; additional options depend on the provider.
-      |
-      |History Server options:
-      |
-      |  ui.port           Port where server will listen for connections (default 18080)
-      |  ui.acls.enable    Whether to enable view acls for all applications (default false)
-      |  provider          Name of history provider class (defaults to file system-based provider)
-      |
-      |FsHistoryProvider options:
-      |
-      |  fs.logDirectory   Directory where app logs are stored (required)
-      |  fs.updateInterval How often to reload log data from storage (seconds, default 10)
-      |""".stripMargin)
-    System.exit(exitCode)
-  }
-
-}
-
-private[spark] abstract class ApplicationHistoryProvider {
-
-  /**
-   * This method should return a list of applications available for the history server to
-   * show.
-   *
-   * @return List of all know applications.
-   */
-  def getListing(): Seq[ApplicationHistoryInfo]
-
-  /**
-   * This method should return the application information, including a rendered SparkUI.
-   *
-   * @param appId The application ID.
-   * @return The app info, or null if not found.
-   */
-  def getAppInfo(appId: String): ApplicationHistoryInfo
-
-  /**
-   * Called when the server is shutting down.
-   */
-  def stop(): Unit = { }
-
-}
-
-private[spark] case class ApplicationHistoryInfo(
-    id: String,
-    name: String,
-    startTime: Long,
-    endTime: Long,
-    lastUpdated: Long,
-    sparkUser: String,
-    viewAcls: String,
-    ui: SparkUI) {
 }

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

@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.history
+
+import java.net.URI
+
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.SparkConf
+import org.apache.spark.util.Utils
+
+/**
+ * Command-line parser for the master.
+ */
+private[spark] class HistoryServerArguments(conf: SparkConf, args: Array[String]) {
+  var logDir = conf.get("spark.history.fs.logDirectory", null)
+
+  parse(args.toList)
+
+  private def parse(args: List[String]): Unit = {
+    args match {
+      case ("--dir" | "-d") :: value :: tail =>
+        logDir = value
+        parse(tail)
+
+      case ("--help" | "-h") :: tail =>
+        printUsageAndExit(0)
+
+      case Nil =>
+
+      case _ =>
+        printUsageAndExit(1)
+    }
+    validateLogDir()
+    conf.set("spark.history.fs.logDirectory", logDir)
+  }
+
+  private def validateLogDir() {
+    if (logDir == null) {
+      System.err.println("Logging directory must be specified.")
+      printUsageAndExit(1)
+    }
+    val fileSystem = Utils.getHadoopFileSystem(new URI(logDir))
+    val path = new Path(logDir)
+    if (!fileSystem.exists(path)) {
+      System.err.println("Logging directory specified does not exist: %s".format(logDir))
+      printUsageAndExit(1)
+    }
+    if (!fileSystem.getFileStatus(path).isDir) {
+      System.err.println("Logging directory specified is not a directory: %s".format(logDir))
+      printUsageAndExit(1)
+    }
+  }
+
+  private def printUsageAndExit(exitCode: Int) {
+    System.err.println(
+      """
+      |Usage: HistoryServer [-d logDir]
+      |
+      |The preferred way to pass options is to set the configuration below using
+      |SPARK_HISTORY_OPTS. The "-d" command line argument is avalable for backwards
+      |compatibility, and overrides "spark.history.fs.logDirectory".
+      |
+      |History Server options are always available; additional options depend on the provider.
+      |
+      |History Server options:
+      |
+      |  spark.history.ui.port      Port where server will listen for connections (default 18080)
+      |  spark.history.acls.enable  Whether to enable view acls for all applications (default false)
+      |  spark.history.provider     Name of history provider class (defaults to file system-based
+      |                             provider)
+      |
+      |FsHistoryProvider options:
+      |
+      |  spark.history.fs.logDirectory    Directory where app logs are stored (required)
+      |  spark.history.fs.updateInterval  How often to reload log data from storage (seconds,
+      |                                   default 10)
+      |""".stripMargin)
+    System.exit(exitCode)
+  }
+
+}