add testsuite for thriftserver ui and fix some style issue

Author: tianyi

sql/core/src/main/scala/org/apache/spark/sql/SQLConf.scala

@@ -51,6 +51,8 @@ private[spark] object SQLConf {
 
   // This is only used for the thriftserver
   val THRIFTSERVER_POOL = "spark.sql.thriftserver.scheduler.pool"
+  val THRIFTSERVER_UI_STATEMENT_LIMIT = "spark.sql.thriftserver.ui.retainedStatements"
+  val THRIFTSERVER_UI_SESSION_LIMIT = "spark.sql.thriftserver.ui.retainedSessions"
 
   // This is used to set the default data source
   val DEFAULT_DATA_SOURCE_NAME = "spark.sql.sources.default"

sql/hive-thriftserver/pom.xml

@@ -44,6 +44,7 @@
     <dependency>
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
+      <scope>runtime</scope>
     </dependency>
     <dependency>
       <groupId>${hive.group}</groupId>
@@ -57,6 +58,18 @@
       <groupId>${hive.group}</groupId>
       <artifactId>hive-beeline</artifactId>
     </dependency>
+    <!-- Added for selenium: -->
+    <dependency>
+      <groupId>org.seleniumhq.selenium</groupId>
+      <artifactId>selenium-java</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>io.netty</groupId>
+          <artifactId>netty</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
   </dependencies>
   <build>
     <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2.scala

@@ -22,6 +22,7 @@ import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
 import org.apache.hive.service.cli.thrift.{ThriftBinaryCLIService, ThriftHttpCLIService}
 import org.apache.hive.service.server.{HiveServer2, ServerOptionsProcessor}
+import org.apache.spark.sql.SQLConf
 
 import org.apache.spark.{SparkContext, SparkConf, Logging}
 import org.apache.spark.annotation.DeveloperApi
@@ -52,7 +53,7 @@ object HiveThriftServer2 extends Logging {
     val server = new HiveThriftServer2(sqlContext)
     server.init(sqlContext.hiveconf)
     server.start()
-    listener = new HiveThriftServer2Listener(server, sqlContext.sparkContext.conf)
+    listener = new HiveThriftServer2Listener(server, sqlContext.conf)
     sqlContext.sparkContext.addSparkListener(listener)
     uiTab = if (sqlContext.sparkContext.getConf.getBoolean("spark.ui.enabled", true)) {
       Some(new ThriftServerTab(sqlContext.sparkContext))
@@ -80,7 +81,7 @@ object HiveThriftServer2 extends Logging {
       server.init(SparkSQLEnv.hiveContext.hiveconf)
       server.start()
       logInfo("HiveThriftServer2 started")
-      listener = new HiveThriftServer2Listener(server, SparkSQLEnv.sparkContext.conf)
+      listener = new HiveThriftServer2Listener(server, SparkSQLEnv.hiveContext.conf)
       SparkSQLEnv.sparkContext.addSparkListener(listener)
       uiTab = if (SparkSQLEnv.sparkContext.getConf.getBoolean("spark.ui.enabled", true)) {
         Some(new ThriftServerTab(SparkSQLEnv.sparkContext))
@@ -100,10 +101,10 @@ object HiveThriftServer2 extends Logging {
       val ip: String,
       val userName: String) {
     var finishTimestamp: Long = 0L
-    var totalExecute: Int = 0
+    var totalExecution: Int = 0
     def totalTime: Long = {
       if (finishTimestamp == 0L) {
-        System.currentTimeMillis() - startTimestamp
+        System.currentTimeMillis - startTimestamp
       } else {
         finishTimestamp - startTimestamp
       }
@@ -139,43 +140,37 @@ object HiveThriftServer2 extends Logging {
   /**
    * A inner sparkListener called in sc.stop to clean up the HiveThriftServer2
    */
-  class HiveThriftServer2Listener(
+  private[thriftserver] class HiveThriftServer2Listener(
       val server: HiveServer2,
-      val conf: SparkConf) extends SparkListener {
+      val conf: SQLConf) extends SparkListener {
 
     override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
       server.stop()
     }
 
-    val sessionList = new mutable.HashMap[String, SessionInfo]
-    val executeList = new mutable.HashMap[String, ExecutionInfo]
+    val sessionList = new mutable.LinkedHashMap[String, SessionInfo]
+    val executionList = new mutable.LinkedHashMap[String, ExecutionInfo]
     val retainedStatements =
-      conf.getInt("spark.thriftserver.ui.retainedStatements", 200)
+      conf.getConf(SQLConf.THRIFTSERVER_UI_STATEMENT_LIMIT, "200").toInt
     val retainedSessions =
-      conf.getInt("spark.thriftserver.ui.retainedSessions", 200)
+      conf.getConf(SQLConf.THRIFTSERVER_UI_SESSION_LIMIT, "200").toInt
     var totalRunning = 0
 
     override def onJobStart(jobStart: SparkListenerJobStart): Unit = {
-      val jobGroup = for (
-        props <- Option(jobStart.properties);
-        statement <- Option(props.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
-      ) yield statement
-
-      jobGroup.map( groupId => {
-        val ret = executeList.find( _ match {
-          case (id: String, info: ExecutionInfo) => info.groupId == groupId
-        })
-        if (ret.isDefined) {
-          ret.get._2.jobId += jobStart.jobId.toString
-          ret.get._2.groupId = groupId
-        }
-      })
+      for {
+        props <- Option(jobStart.properties)
+        groupId <- Option(props.getProperty(SparkContext.SPARK_JOB_GROUP_ID))
+        (_, info) <- executionList if info.groupId == groupId
+      } {
+        info.jobId += jobStart.jobId.toString
+        info.groupId = groupId
+      }
     }
 
     def onSessionCreated(ip: String, sessionId: String, userName: String = "UNKNOWN"): Unit = {
       val info = new SessionInfo(sessionId, System.currentTimeMillis, ip, userName)
-      sessionList(sessionId) = info
-      trimSessionIfNecessary
+      sessionList.put(sessionId, info)
+      trimSessionIfNecessary()
     }
 
     def onSessionClosed(sessionId: String): Unit = {
@@ -190,44 +185,44 @@ object HiveThriftServer2 extends Logging {
         userName: String = "UNKNOWN"): Unit = {
       val info = new ExecutionInfo(statement, sessionId, System.currentTimeMillis, userName)
       info.state = ExecutionState.STARTED
-      executeList(id) = info
-      trimExecutionIfNecessary
-      sessionList(sessionId).totalExecute += 1
-      executeList(id).groupId = groupId
+      executionList.put(id, info)
+      trimExecutionIfNecessary()
+      sessionList(sessionId).totalExecution += 1
+      executionList(id).groupId = groupId
       totalRunning += 1
     }
 
-    def onStatementParse(id: String, executePlan: String): Unit = {
-      executeList(id).executePlan = executePlan
-      executeList(id).state = ExecutionState.COMPILED
+    def onStatementParsed(id: String, executionPlan: String): Unit = {
+      executionList(id).executePlan = executionPlan
+      executionList(id).state = ExecutionState.COMPILED
     }
 
     def onStatementError(id: String, errorMessage: String, errorTrace: String): Unit = {
-      executeList(id).finishTimestamp = System.currentTimeMillis
-      executeList(id).detail = errorMessage
-      executeList(id).state = ExecutionState.FAILED
+      executionList(id).finishTimestamp = System.currentTimeMillis
+      executionList(id).detail = errorMessage
+      executionList(id).state = ExecutionState.FAILED
       totalRunning -= 1
     }
 
     def onStatementFinish(id: String): Unit = {
-      executeList(id).finishTimestamp = System.currentTimeMillis
-      executeList(id).state = ExecutionState.FINISHED
+      executionList(id).finishTimestamp = System.currentTimeMillis
+      executionList(id).state = ExecutionState.FINISHED
       totalRunning -= 1
     }
 
-    private def trimExecutionIfNecessary = synchronized {
-      if (executeList.size > retainedStatements) {
+    private def trimExecutionIfNecessary() = synchronized {
+      if (executionList.size > retainedStatements) {
         val toRemove = math.max(retainedStatements / 10, 1)
-        executeList.toList.sortBy(_._2.startTimestamp).take(toRemove).foreach { s =>
-          executeList.remove(s._1)
+        executionList.take(toRemove).foreach { s =>
+          executionList.remove(s._1)
         }
       }
     }
 
-    private def trimSessionIfNecessary = synchronized {
+    private def trimSessionIfNecessary() = synchronized {
       if (sessionList.size > retainedSessions) {
         val toRemove = math.max(retainedSessions / 10, 1)
-        sessionList.toList.sortBy(_._2.startTimestamp).take(toRemove).foreach { s =>
+        sessionList.take(toRemove).foreach { s =>
           sessionList.remove(s._1)
         }
       }

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPage.scala

@@ -20,13 +20,14 @@ package org.apache.spark.sql.hive.thriftserver.ui
 import java.util.Calendar
 import javax.servlet.http.HttpServletRequest
 
+import scala.xml.Node
+
 import org.apache.commons.lang3.StringEscapeUtils
 import org.apache.spark.Logging
 import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{SessionInfo, ExecutionState, ExecutionInfo}
 import org.apache.spark.ui.UIUtils._
 import org.apache.spark.ui._
 
-import scala.xml.Node
 
 /** Page for Spark Web UI that shows statistics of a streaming job */
 private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("") with Logging {
@@ -41,8 +42,8 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
       generateBasicStats() ++
       <br/> ++
       <h4>
-        Total {listener.sessionList.size} session online,
-        Total {listener.totalRunning} sql running
+        {listener.sessionList.size} session(s) are online,
+        running {listener.totalRunning} SQL statement(s)
       </h4> ++
       generateSessionStatsTable() ++
       generateSQLStatsTable()
@@ -64,11 +65,11 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
 
   /** Generate stats of batch statements of the thrift server program */
   private def generateSQLStatsTable(): Seq[Node] = {
-    val numStatement = listener.executeList.size
+    val numStatement = listener.executionList.size
     val table = if (numStatement > 0) {
       val headerRow = Seq("User", "JobID", "GroupID", "Start Time", "Finish Time", "Duration",
         "Statement", "State", "Detail")
-      val dataRows = listener.executeList.values.toSeq.sortBy(_.startTimestamp).reverse
+      val dataRows = listener.executionList.values
 
       def generateDataRow(info: ExecutionInfo): Seq[Node] = {
         val jobLink = info.jobId.map { id: String =>
@@ -99,7 +100,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
     }
 
     val content =
-      <h5>SQL Statistics</h5> ++
+      <h5 id="sqlstat">SQL Statistics</h5> ++
         <div>
           <ul class="unstyled">
             {table.getOrElse("No statistics have been generated yet.")}
@@ -138,7 +139,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
     val numBatches = listener.sessionList.size
     val table = if (numBatches > 0) {
       val dataRows =
-        listener.sessionList.values.toSeq.sortBy(_.startTimestamp).reverse
+        listener.sessionList.values
       val headerRow = Seq("User", "IP", "Session ID", "Start Time", "Finish Time", "Duration",
         "Total Execute")
       def generateDataRow(session: SessionInfo): Seq[Node] = {
@@ -151,7 +152,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
           <td> {formatDate(session.startTimestamp)} </td>
           <td> {if(session.finishTimestamp > 0) formatDate(session.finishTimestamp)} </td>
           <td> {formatDurationOption(Some(session.totalTime))} </td>
-          <td> {session.totalExecute.toString} </td>
+          <td> {session.totalExecution.toString} </td>
         </tr>
       }
       Some(UIUtils.listingTable(headerRow, generateDataRow, dataRows, true, None, Seq(null), false))
@@ -160,7 +161,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
     }
 
     val content =
-      <h5>Session Statistics</h5> ++
+      <h5 id="sessionstat">Session Statistics</h5> ++
       <div>
         <ul class="unstyled">
           {table.getOrElse("No statistics have been generated yet.")}

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerSessionPage.scala

@@ -20,14 +20,14 @@ package org.apache.spark.sql.hive.thriftserver.ui
 import java.util.Calendar
 import javax.servlet.http.HttpServletRequest
 
+import scala.xml.Node
+
 import org.apache.commons.lang3.StringEscapeUtils
 import org.apache.spark.Logging
 import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{ExecutionInfo, ExecutionState}
 import org.apache.spark.ui.UIUtils._
 import org.apache.spark.ui._
 
-import scala.xml.Node
-
 /** Page for Spark Web UI that shows statistics of a streaming job */
 private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
   extends WebUIPage("session") with Logging {
@@ -52,7 +52,7 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
         User {sessionStat._2.userName},
         IP {sessionStat._2.ip},
         Session created at {formatDate(sessionStat._2.startTimestamp)},
-        Total run {sessionStat._2.totalExecute} SQL
+        Total run {sessionStat._2.totalExecution} SQL
       </h4> ++
       generateSQLStatsTable(sessionStat._2.sessionId)
     UIUtils.headerSparkPage("ThriftServer", content, parent, Some(5000))
@@ -73,7 +73,7 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
 
   /** Generate stats of batch statements of the thrift server program */
   private def generateSQLStatsTable(sessionID: String): Seq[Node] = {
-    val executionList = listener.executeList
+    val executionList = listener.executionList
       .filter(_._2.sessionId == sessionID)
     val numStatement = executionList.size
     val table = if (numStatement > 0) {
@@ -157,7 +157,7 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
           formatDate(session.startTimestamp),
           formatDate(session.finishTimestamp),
           formatDurationOption(Some(session.totalTime)),
-          session.totalExecute.toString
+          session.totalExecution.toString
         )
       ).toSeq
       val headerRow = Seq("User", "IP", "Session ID", "Start Time", "Finish Time", "Duration",