[CARMEL-3519] Viewpoint Support (#27)

Author: xingchaozh

assembly/pom.xml

@@ -79,6 +79,11 @@
       <artifactId>spark-repl_${scala.binary.version}</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-viewpoint_${scala.binary.version}</artifactId>
+      <version>${project.version}</version>
+    </dependency>
 
     <!--
       Because we don't shade dependencies anymore, we need to restore Guava to compile scope so

core/src/main/resources/org/apache/spark/ui/static/historypage-common.js

@@ -18,7 +18,7 @@
 $(document).ready(function() {
   if ($('#last-updated').length) {
     var lastUpdatedMillis = Number($('#last-updated').text());
-    $('#last-updated').text(formatTimeMillis(lastUpdatedMillis));
+    $('#last-updated').text(formatTimeMillisWithTimeZone(lastUpdatedMillis, 0));
   }
 
   $('#time-zone').text(getTimeZone());

core/src/main/resources/org/apache/spark/ui/static/historypage-template.html

@@ -46,7 +46,6 @@
           Started
         </span>
       </th>
-      {{#showCompletedColumns}}
       <th>
         <span data-toggle="tooltip" data-placement="top" title="The completed time of this application.">
           Completed
@@ -57,7 +56,6 @@
           Duration
         </span>
       </th>
-      {{/showCompletedColumns}}
       <th>
         <span data-toggle="tooltip" data-placement="top" title="The Spark user of this application">
           Spark User
@@ -85,10 +83,8 @@
       <td><a href="{{uiroot}}/history/{{id}}/{{attemptId}}/jobs/">{{attemptId}}</a></td>
       {{/hasMultipleAttempts}}
       <td>{{startTime}}</td>
-      {{#showCompletedColumns}}
       <td>{{endTime}}</td>
       <td><span title="{{durationMillisec}}">{{duration}}</span></td>
-      {{/showCompletedColumns}}
       <td>{{sparkUser}}</td>
       <td>{{lastUpdated}}</td>
       <td><a href="{{log}}" class="btn btn-info btn-mini">Download</a></td>

core/src/main/resources/org/apache/spark/ui/static/historypage.js

@@ -55,6 +55,21 @@ function getColumnIndex(columns, columnName) {
   return -1;
 }
 
+function onSearch() {
+  var searchStr = $('#search').val().toLowerCase().trim();
+  if (searchStr == "") {
+    return;
+  }
+  var reg= /^[0-9a-z]{8}-[0-9a-z]{4}-[0-9a-z]{4}-[0-9a-z]{4}-[0-9a-z]{12}$/;
+  if (searchStr.match(reg)) {
+    window.location.href="/?session="+searchStr;
+  } else if(searchStr.indexOf("hdmi-") >= 0) {
+    window.location.href="/?queue="+searchStr;
+  } else {
+    window.location.href="/?user="+searchStr;
+  }
+}
+
 jQuery.extend( jQuery.fn.dataTableExt.oSort, {
     "title-numeric-pre": function ( a ) {
         var x = a.match(/title="*(-?[0-9\.]+)/)[1];
@@ -105,22 +120,34 @@ $(document).ready(function() {
 
     var historySummary = $("#history-summary");
     var searchString = window.location.search;
-    var requestedIncomplete = getParameterByName("showIncomplete", searchString);
-    requestedIncomplete = (requestedIncomplete == "true" ? true : false);
+
+    var searchSessionId = getParameterByName("session", searchString);
+    var searchUserId = getParameterByName("user", searchString)
+    var longRunning = getParameterByName("longRunning", searchString)
+    longRunning = (longRunning == "true" ? true : false);
+
+    var url = "api/v1/applications?limit=" + appLimit
+    if(searchSessionId != "") {
+      url += "&session=" + searchSessionId
+    }
+
+    if(searchUserId != "") {
+      url += "&user=" + searchUserId
+    }
+
+    if(longRunning) {
+      url += "&longRunning=" + longRunning
+    }
 
     var appParams = {
-      limit: appLimit,
-      status: (requestedIncomplete ? "running" : "completed")
     };
 
-    $.getJSON(uiRoot + "/api/v1/applications", appParams, function(response,status,jqXHR) {
+    console.log(url)
+    $.getJSON(url, appParams, function(response,status,jqXHR) {
       var array = [];
       var hasMultipleAttempts = false;
       for (var i in response) {
         var app = response[i];
-        if (app["attempts"][0]["completed"] == requestedIncomplete) {
-          continue; // if we want to show for Incomplete, we skip the completed apps; otherwise skip incomplete ones.
-        }
         var version = "Unknown"
         if (app["attempts"].length > 0) {
             version = app["attempts"][0]["appSparkVersion"]
@@ -131,18 +158,22 @@ $(document).ready(function() {
             hasMultipleAttempts = true;
         }
         var num = app["attempts"].length;
+        if(num >= 1) {
+          var latestAttempt = app["attempts"][num -1]
+          var latestAttemptId = (latestAttempt.hasOwnProperty("attemptId") ? latestAttempt["attemptId"] : "")
         for (var j in app["attempts"]) {
           var attempt = app["attempts"][j];
-          attempt["startTime"] = formatTimeMillis(attempt["startTimeEpoch"]);
-          attempt["endTime"] = formatTimeMillis(attempt["endTimeEpoch"]);
-          attempt["lastUpdated"] = formatTimeMillis(attempt["lastUpdatedEpoch"]);
+          attempt["startTime"] = formatTimeMillisWithTimeZone(attempt["startTimeEpoch"], 0);
+          attempt["endTime"] = formatTimeMillisWithTimeZone(attempt["endTimeEpoch"], 0);
+          attempt["lastUpdated"] = formatTimeMillisWithTimeZone(attempt["lastUpdatedEpoch"], 0);
           attempt["log"] = uiRoot + "/api/v1/applications/" + id + "/" +
             (attempt.hasOwnProperty("attemptId") ? attempt["attemptId"] + "/" : "") + "logs";
           attempt["durationMillisec"] = attempt["duration"];
           attempt["duration"] = formatDuration(attempt["duration"]);
-          var app_clone = {"id" : id, "name" : name, "version": version, "num" : num, "attempts" : [attempt]};
+          var app_clone = {"version": version, "id" : id, "name" : name, "num" : num, "attempts" : [attempt], "latestAttemptId": latestAttemptId};
           array.push(app_clone);
         }
+        }
       }
       if(array.length < 20) {
         $.fn.dataTable.defaults.paging = false;
@@ -152,7 +183,6 @@ $(document).ready(function() {
         "uiroot": uiRoot,
         "applications": array,
         "hasMultipleAttempts": hasMultipleAttempts,
-        "showCompletedColumns": !requestedIncomplete,
       };
 
       $.get(uiRoot + "/static/historypage-template.html", function(template) {
@@ -177,6 +207,7 @@ $(document).ready(function() {
             {name: 'eventLog'},
           ],
           "autoWidth": false,
+          "searching": false,
           "deferRender": true
         };
 
@@ -190,12 +221,7 @@ $(document).ready(function() {
           conf.columns = removeColumnByName(conf.columns, attemptIdColumnName);
         }
 
-        var defaultSortColumn = completedColumnName;
-        if (requestedIncomplete) {
-          defaultSortColumn = startedColumnName;
-          conf.columns = removeColumnByName(conf.columns, completedColumnName);
-          conf.columns = removeColumnByName(conf.columns, durationColumnName);
-        }
+        var defaultSortColumn = startedColumnName;
         conf.order = [[ getColumnIndex(conf.columns, defaultSortColumn), "desc" ]];
         conf.columnDefs = [
           {"searchable": false, "targets": [getColumnIndex(conf.columns, durationColumnName)]}
@@ -206,4 +232,17 @@ $(document).ready(function() {
         $('#history-summary [data-toggle="tooltip"]').tooltip();
       });
     });
+
+
+    if(searchSessionId != "") {
+      $('#search').val(searchSessionId);
+    }
+    if(searchUserId != "") {
+      $('#search').val(searchUserId);
+    }
+    $('#search').bind('keypress',function(event){
+      if(event.keyCode == "13") {
+        onSearch();
+      }
+    });
 });

core/src/main/resources/org/apache/spark/ui/static/utils.js

@@ -51,6 +51,21 @@ function padZeroes(num) {
   return ("0" + num).slice(-2);
 }
 
+function formatTimeMillisWithTimeZone(timeMillis, timeZone) {
+  if (timeMillis <= 0) {
+    return "-";
+  } else {
+    var offset_GMT = new Date().getTimezoneOffset();
+    var dt = new Date(timeMillis + offset_GMT * 60 * 1000 + timeZone * 60 * 60 * 1000);
+    return dt.getFullYear() + "-" +
+      padZeroes(dt.getMonth() + 1) + "-" +
+      padZeroes(dt.getDate()) + " " +
+      padZeroes(dt.getHours()) + ":" +
+      padZeroes(dt.getMinutes()) + ":" +
+      padZeroes(dt.getSeconds());
+  }
+}
+
 function formatTimeMillis(timeMillis) {
   if (timeMillis <= 0) {
     return "-";

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

@@ -1563,6 +1563,15 @@ class SparkContext(config: SparkConf) extends Logging {
     }
   }
 
+  /**
+   * :: DeveloperApi ::
+   * Register a listener to event log queue.
+   */
+  @DeveloperApi
+  def addSparkListenerToEventLogQueue(listener: SparkListenerInterface): Unit = {
+    listenerBus.addToEventLogQueue(listener)
+  }
+
   /**
    * :: DeveloperApi ::
    * Register a listener to receive up-calls from events that happen during execution.
@@ -2474,6 +2483,24 @@ class SparkContext(config: SparkConf) extends Logging {
       driverUpdates))
   }
 
+  def getSparkListenerApplicationStart(): Option[SparkListenerApplicationStart] = {
+    Some(SparkListenerApplicationStart(appName, Some(applicationId),
+      startTime, sparkUser, applicationAttemptId, schedulerBackend.getDriverLogUrls))
+  }
+
+  def getSparkListenerEnvironmentUpdate(): Option[SparkListenerEnvironmentUpdate] = {
+    if (taskScheduler != null) {
+      val schedulingMode = getSchedulingMode.toString
+      val addedJarPaths = addedJars.keys.toSeq
+      val addedFilePaths = addedFiles.keys.toSeq
+      val environmentDetails = SparkEnv.environmentDetails(conf, hadoopConfiguration,
+        schedulingMode, addedJarPaths, addedFilePaths)
+      Some(SparkListenerEnvironmentUpdate(environmentDetails))
+    } else {
+      None
+    }
+  }
+
   // In order to prevent multiple SparkContexts from being active at the same time, mark this
   // context as having finished construction.
   // NOTE: this must be placed at the end of the SparkContext constructor.

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -408,6 +408,8 @@ private[spark] object SparkHadoopUtil {
 
   val SPARK_YARN_CREDS_COUNTER_DELIM = "-"
 
+  val HIVE_SESSION_ID = "spark.hive.session.id"
+
   /**
    * Number of records to update input metrics when reading from HadoopRDDs.
    *

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

@@ -27,11 +27,20 @@ import org.apache.spark.ui.{UIUtils, WebUIPage}
 private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("") {
 
   def render(request: HttpServletRequest): Seq[Node] = {
-    val requestedIncomplete = Option(request.getParameter("showIncomplete"))
+    val sessionId =
+      Option(request.getParameter("session")).getOrElse("").trim
+
+    val userId =
+      Option(request.getParameter("user")).getOrElse("").trim
+
+    val longRunning = Option(request.getParameter("longRunning"))
       .getOrElse("false").toBoolean
 
-    val displayApplications = parent.getApplicationList()
-      .exists(isApplicationCompleted(_) != requestedIncomplete)
+    val displayApplications = parent.getApplicationList().
+      filter(p => sessionId == "" || p.id.endsWith(sessionId)).
+      filter(p => userId == "" || p.attempts.exists(_.sparkUser.equalsIgnoreCase(userId))).
+      length > 0
+
     val eventLogsUnderProcessCount = parent.getEventLogsUnderProcess()
     val lastUpdatedTime = parent.getLastUpdatedTime()
     val providerConfig = parent.getProviderConfig()
@@ -58,7 +67,27 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
             }
 
             {
-            <p>Client local time zone: <span id="time-zone"></span></p>
+            <p>Time zone: <span>UTC</span></p>
+            }
+
+            {
+            <div><table style="width:100%"><tbody><tr>
+              <td><b style="margin-left:-30px;margin-right:5px">
+                </b>
+                {
+                if (!longRunning) {
+                  <a href={
+                     UIUtils.prependBaseUri(request, "/?" + "longRunning=true")}>
+                    Long Running
+                  </a>
+                }
+                }
+              </td>
+              <td align="right">
+                <input type="text" placeholder="Input Session Id or User Id" id="search"
+                       style="min-width:280px;margin-right:1px;margin-top:10px"/>
+                <button class="btn btn-info" onclick="onSearch()">Search</button>
+              </td></tr></tbody></table></div>
             }
 
             {
@@ -68,31 +97,15 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
                 <div id="history-summary" class="row-fluid"></div> ++
                 <script src={UIUtils.prependBaseUri(request, "/static/historypage.js")}></script> ++
                 <script>setAppLimit({parent.maxApplications})</script>
-            } else if (requestedIncomplete) {
-              <h4>No incomplete applications found!</h4>
             } else if (eventLogsUnderProcessCount > 0) {
-              <h4>No completed applications found!</h4>
+              <h4>No applications found!</h4>
             } else {
-              <h4>No completed applications found!</h4> ++ parent.emptyListingHtml
+              <h4>No applications found!</h4> ++ parent.emptyListingHtml
             }
             }
-
-            <a href={makePageLink(request, !requestedIncomplete)}>
-              {
-              if (requestedIncomplete) {
-                "Back to completed applications"
-              } else {
-                "Show incomplete applications"
-              }
-              }
-            </a>
           </div>
       </div>
-    UIUtils.basicSparkPage(request, content, "History Server", true)
-  }
-
-  private def makePageLink(request: HttpServletRequest, showIncomplete: Boolean): String = {
-    UIUtils.prependBaseUri(request, "/?" + "showIncomplete=" + showIncomplete)
+    UIUtils.basicSparkPage(request, content, parent.serviceName, true)
   }
 
   private def isApplicationCompleted(appInfo: ApplicationInfo): Boolean = {