Skip to content

Commit

Permalink
[KYUUBI #5894] Separate closed and online sessions[statements] in the…
Browse files Browse the repository at this point in the history 
… SparkUI's engine tab
  • Loading branch information
wangjunbo committed Jan 4, 2024
1 parent c6bba91 commit 83c7b15
Show file tree
Hide file tree
Showing 3 changed files with 333 additions and 129 deletions.
290 changes: 206 additions & 84 deletions externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/ui/EnginePage.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import java.util.Date
import javax.servlet.http.HttpServletRequest

import scala.collection.JavaConverters.mapAsScalaMapConverter
import scala.collection.mutable
import scala.xml.{Node, Unparsed}

import org.apache.commons.text.StringEscapeUtils
Expand All @@ -36,18 +37,42 @@ case class EnginePage(parent: EngineTab) extends WebUIPage("") {
private val store = parent.store

override def render(request: HttpServletRequest): Seq[Node] = {
val onlineSession = new mutable.ArrayBuffer[SessionEvent]()
val closedSession = new mutable.ArrayBuffer[SessionEvent]()

val runningSqlStat = new mutable.ArrayBuffer[SparkOperationEvent]()
val completedSqlStat = new mutable.ArrayBuffer[SparkOperationEvent]()
val failedSqlStat = new mutable.ArrayBuffer[SparkOperationEvent]()

store.getSessionList.foreach { s =>
if (s.endTime <= 0L) {
onlineSession += s
} else {
closedSession += s
}
}

store.getStatementList.foreach { op =>
if (op.completeTime <= 0L) {
runningSqlStat += op
} else if (op.exception.isDefined) {
failedSqlStat += op
} else {
completedSqlStat += op
}
}

val content =
generateBasicStats() ++
<br/> ++
stop(request) ++
<br/> ++
<h4>
{store.getSessionCount} session(s) are online,
running {store.getStatementCount}
operations
{onlineSession.size} session(s) are online,
running {runningSqlStat.size} operation(s)
</h4> ++
generateSessionStatsTable(request) ++
generateStatementStatsTable(request)
generateSessionStatsTable(request, onlineSession, closedSession) ++
generateStatementStatsTable(request, runningSqlStat, completedSqlStat, failedSqlStat)
UIUtils.headerSparkPage(request, parent.name, content, parent)
}

Expand Down Expand Up @@ -129,102 +154,199 @@ case class EnginePage(parent: EngineTab) extends WebUIPage("") {
}
}

/** Generate stats of statements for the engine */
private def generateStatementStatsTable(request: HttpServletRequest): Seq[Node] = {

val numStatement = store.getStatementList.size

val table =
if (numStatement > 0) {
/** Generate stats of running statements for the engine */
private def generateStatementStatsTable(
request: HttpServletRequest,
running: Seq[SparkOperationEvent],
completed: Seq[SparkOperationEvent],
failed: Seq[SparkOperationEvent]): Seq[Node] = {

val content = mutable.ListBuffer[Node]()
if (running.nonEmpty) {
val sqlTableTag = "running"
val table =
statementStatsTable(request, sqlTableTag, parent, running)
content ++=
<span id="running" class="collapse-aggregated-runningSqlstat collapse-table"
onClick="collapseTable('collapse-aggregated-runningSqlstat',
'aggregated-runningSqlstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Running Statement Statistics (
{running.size}
)</a>
</h4>
</span> ++
<div class="aggregated-runningSqlstat collapsible-table">
{table}
</div>
}

val sqlTableTag = "sqlstat"
if (completed.nonEmpty) {
val table = {
val sessionTableTag = "completed"
statementStatsTable(
request,
sessionTableTag,
parent,
completed)
}

val sqlTablePage =
Option(request.getParameter(s"$sqlTableTag.page")).map(_.toInt).getOrElse(1)
content ++=
<span id="completed" class="collapse-aggregated-completedSqlstat collapse-table"
onClick="collapseTable('collapse-aggregated-completedSqlstat',
'aggregated-completedSqlstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Completed Statement Statistics (
{completed.size}
)</a>
</h4>
</span> ++
<div class="aggregated-completedSqlstat collapsible-table">
{table}
</div>
}

try {
Some(new StatementStatsPagedTable(
request,
parent,
store.getStatementList,
"kyuubi",
UIUtils.prependBaseUri(request, parent.basePath),
sqlTableTag).table(sqlTablePage))
} catch {
case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
Some(<div class="alert alert-error">
<p>Error while rendering job table:</p>
<pre>
{Utils.stringifyException(e)}
</pre>
</div>)
}
} else {
None
if (failed.nonEmpty) {
val table = {
val sessionTableTag = "failed"
statementStatsTable(
request,
sessionTableTag,
parent,
failed)
}
val content =
<span id="sqlstat" class="collapse-aggregated-sqlstat collapse-table"
onClick="collapseTable('collapse-aggregated-sqlstat',
'aggregated-sqlstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Statement Statistics ({numStatement})</a>
</h4>
</span> ++
<div class="aggregated-sqlstat collapsible-table">
{table.getOrElse("No statistics have been generated yet.")}

content ++=
<span id="failed" class="collapse-aggregated-failedSqlstat collapse-table"
onClick="collapseTable('collapse-aggregated-failedSqlstat',
'aggregated-failedSqlstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Failed Statement Statistics (
{failed.size}
)</a>
</h4>
</span> ++
<div class="aggregated-failedSqlstat collapsible-table">
{table}
</div>
}
content
}

/** Generate stats of sessions for the engine */
private def generateSessionStatsTable(request: HttpServletRequest): Seq[Node] = {
val numSessions = store.getSessionList.size
val table =
if (numSessions > 0) {
private def statementStatsTable(
request: HttpServletRequest,
sqlTableTag: String,
parent: EngineTab,
data: Seq[SparkOperationEvent]): Seq[Node] = {

val sqlTablePage =
Option(request.getParameter(s"$sqlTableTag.page")).map(_.toInt).getOrElse(1)

val sessionTableTag = "sessionstat"
try {
new StatementStatsPagedTable(
request,
parent,
data,
"kyuubi",
UIUtils.prependBaseUri(request, parent.basePath),
sqlTableTag).table(sqlTablePage)
} catch {
case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
<div class="alert alert-error">
<p>Error while rendering job table:</p>
<pre>
{Utils.stringifyException(e)}
</pre>
</div>
}
}

val sessionTablePage =
Option(request.getParameter(s"$sessionTableTag.page")).map(_.toInt).getOrElse(1)
/** Generate stats of online sessions for the engine */
private def generateSessionStatsTable(
request: HttpServletRequest,
online: Seq[SessionEvent],
closed: Seq[SessionEvent]): Seq[Node] = {
val content = mutable.ListBuffer[Node]()
if (online.nonEmpty) {
val sessionTableTag = "online"
val table = sessionTable(
request,
sessionTableTag,
parent,
online)
content ++=
<span id="online" class="collapse-aggregated-onlineSessionstat collapse-table"
onClick="collapseTable('collapse-aggregated-onlineSessionstat',
'aggregated-onlineSessionstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Online Session Statistics (
{online.size}
)</a>
</h4>
</span> ++
<div class="aggregated-onlineSessionstat collapsible-table">
{table}
</div>
}

try {
Some(new SessionStatsPagedTable(
request,
parent,
store.getSessionList,
"kyuubi",
UIUtils.prependBaseUri(request, parent.basePath),
sessionTableTag).table(sessionTablePage))
} catch {
case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
Some(<div class="alert alert-error">
<p>Error while rendering job table:</p>
<pre>
{Utils.stringifyException(e)}
</pre>
</div>)
}
} else {
None
if (closed.nonEmpty) {
val table = {
val sessionTableTag = "closed"
sessionTable(
request,
sessionTableTag,
parent,
closed)
}

val content =
<span id="sessionstat" class="collapse-aggregated-sessionstat collapse-table"
onClick="collapseTable('collapse-aggregated-sessionstat',
'aggregated-sessionstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Session Statistics ({numSessions})</a>
</h4>
</span> ++
<div class="aggregated-sessionstat collapsible-table">
{table.getOrElse("No statistics have been generated yet.")}
content ++=
<span id="closed" class="collapse-aggregated-closedSessionstat collapse-table"
onClick="collapseTable('collapse-aggregated-closedSessionstat',
'aggregated-closedSessionstat')">
<h4>
<span class="collapse-table-arrow arrow-open"></span>
<a>Closed Session Statistics (
{closed.size}
)</a>
</h4>
</span> ++
<div class="aggregated-closedSessionstat collapsible-table">
{table}
</div>

}
content
}

private def sessionTable(
request: HttpServletRequest,
sessionTage: String,
parent: EngineTab,
data: Seq[SessionEvent]): Seq[Node] = {
val sessionPage =
Option(request.getParameter(s"$sessionTage.page")).map(_.toInt).getOrElse(1)
try {
new SessionStatsPagedTable(
request,
parent,
data,
"kyuubi",
UIUtils.prependBaseUri(request, parent.basePath),
sessionTage).table(sessionPage)
} catch {
case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
<div class="alert alert-error">
<p>Error while rendering job table:</p>
<pre>
{Utils.stringifyException(e)}
</pre>
</div>
}
}

private class SessionStatsPagedTable(
request: HttpServletRequest,
parent: EngineTab,
Expand Down
Loading

0 comments on commit 83c7b15

Please sign in to comment.