[FLINK-3131] [runtime-web] Add checkpoint statistics to web frontend

This closes #1453

flink-runtime-web/web-dashboard/app/partials/jobs/job.plan.jade

@@ -27,4 +27,7 @@
       li(ui-sref-active='active')
         a(ui-sref=".accumulators({nodeid: nodeid})") Accumulators
 
+      li(ui-sref-active='active')
+        a(ui-sref=".checkpoints({nodeid: nodeid})") Checkpoints
+
   .panel-body.clean(ui-view="node-details")

flink-runtime-web/web-dashboard/app/partials/jobs/job.plan.node-list.checkpoints.jade

@@ -0,0 +1,43 @@
+//
+  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.
+
+div(ng-if="!operatorCheckpointStats")
+  p
+    em No checkpoints
+
+div(ng-if="operatorCheckpointStats")
+  h2 Overview
+
+  div(ng-include=" 'partials/jobs/job.plan.node.checkpoints.job.html' ")
+
+  h2 Operators
+
+  table.table.table-body-hover.table-clickable.table-activable
+    thead
+      tr
+        th Name
+        th Status
+
+    tbody(ng-repeat="v in job.vertices" ng-class="{ active: v.id == nodeid }" ng-click="v.id == nodeid || changeNode(v.id)")
+      tr(ng-if="v.type == 'regular'")
+        td {{ v.name | humanizeText }}
+        td
+          bs-label(status="{{v.status}}") {{v.status}}
+
+      tr(ng-if="nodeid && v.id == nodeid")
+        td(colspan="10")
+          div(ng-include=" 'partials/jobs/job.plan.node.checkpoints.operator.html' ")

flink-runtime-web/web-dashboard/app/partials/jobs/job.plan.node.checkpoints.job.jade

@@ -0,0 +1,96 @@
+//
+  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.
+
+div(ng-if="!jobCheckpointStats")
+  p
+    em No checkpoints
+
+table(ng-if="jobCheckpointStats").table.table-hover.table-inner
+  thead
+    tr
+      td(colspan=3)
+        strong Overview
+  tbody
+    tr
+      td
+        strong Count
+      td(colspan=3)
+        span {{ jobCheckpointStats['count'] }}
+
+    tr
+      td
+        strong Duration
+      td
+        p
+          strong Minimum:
+          span  {{ jobCheckpointStats['duration']['min'] | humanizeDuration }}
+      td
+        p
+          strong Maximum:
+          span  {{ jobCheckpointStats['duration']['max'] | humanizeDuration }}
+
+      td
+        p
+          strong Average:
+          span  {{ jobCheckpointStats['duration']['avg'] | humanizeDuration }}
+
+    tr
+      td
+          strong State Size
+      td
+          p
+            strong Minimum:
+            span  {{ jobCheckpointStats['size']['min'] | humanizeBytes }}
+      td
+          p
+            strong Maximum:
+            span  {{ jobCheckpointStats['size']['max'] | humanizeBytes }}
+      td
+          p
+            strong Average:
+            span  {{ jobCheckpointStats['size']['avg'] | humanizeBytes }}
+
+div(ng-if="!showHistory && jobCheckpointStats && jobCheckpointStats['history'].length > 0")
+  a.btn.btn-default(ng-click="toggleHistory()")
+    | <strong>Show history</strong> ({{ jobCheckpointStats['history'].length }})
+    = ' '
+    i.fa.fa-chevron-down
+
+div(ng-if="showHistory && jobCheckpointStats && jobCheckpointStats['history'].length > 0")
+  a.btn.btn-default(ng-click="toggleHistory()")
+    | Hide history ({{ jobCheckpointStats['history'].length }})
+    = ' '
+    i.fa.fa-chevron-up
+
+  table.table.table-hover.table-inner
+    thead
+      tr
+        td
+          strong ID
+        td
+          strong Trigger Time
+        td
+          strong Duration
+        td
+          strong State Size
+
+    tbody(ng-if="jobCheckpointStats['history'] && jobCheckpointStats['history'].length > 0" ng-repeat="historic in jobCheckpointStats['history']")
+      tr
+        td {{ historic['id'] }}
+        td {{ historic['timestamp'] | amDateFormat:'H:mm:ss' }}
+        td {{ historic['duration'] | humanizeDuration }}
+        td {{ historic['size'] | humanizeBytes }}

flink-runtime-web/web-dashboard/app/partials/jobs/job.plan.node.checkpoints.operator.jade

@@ -0,0 +1,64 @@
+//
+  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.
+
+div(ng-if="!operatorCheckpointStats")
+  p
+    em No checkpoints
+
+div(ng-if="operatorCheckpointStats")
+  table.table.table-hover.table-clickable.table-activable.table-inner
+    thead
+      tr
+        th ID
+        th Trigger Timestamp
+        th Duration
+        th State Size
+
+    tbody
+      tr
+        td(width="22%") {{ operatorCheckpointStats['id'] }}
+        td(width="22%") {{ operatorCheckpointStats['timestamp'] | amDateFormat:'H:mm:ss' }}
+        td(width="22%") {{ operatorCheckpointStats['duration'] | humanizeDuration }}
+        td(width="22%") {{ operatorCheckpointStats['size'] | humanizeBytes }}
+
+  div(ng-if="!nodeUnfolded && subtasksCheckpointStats && subtasksCheckpointStats.length > 0")
+    a.btn.btn-default(ng-click="toggleFold()")
+      | Show subtasks
+      = ' '
+      i.fa.fa-chevron-down
+
+    a.btn.btn-default.pull-right(ng-click="deactivateNode(); $event.stopPropagation()" title="Fold")
+      i.fa.fa-chevron-up
+
+  div(ng-if="nodeUnfolded && subtasksCheckpointStats && subtasksCheckpointStats.length > 0")
+    a.btn.btn-default(ng-click="toggleFold()")
+      | Hide subtasks
+      = ' '
+      i.fa.fa-chevron-up
+
+    table.table.table-hover.table-clickable.table-activable.table-inner
+      thead
+        tr
+          th Subtask
+          th Duration
+          th Type
+
+      tbody(ng-repeat="subtask in subtasksCheckpointStats")
+        tr
+          td {{ subtask['subtask'] + 1 }}
+          td {{ subtask['duration'] | humanizeDuration }}
+          td {{ subtask['size'] | humanizeBytes }}

flink-runtime-web/web-dashboard/app/scripts/index.coffee

@@ -30,7 +30,7 @@ angular.module('flinkApp', ['ui.router', 'angularMoment'])
 
 .value 'flinkConfig', {
   jobServer: ''
-#  jobServer: 'http://localhost:8081/'
+# jobServer: 'http://localhost:8081/'
   "refresh-interval": 10000
 }
 
@@ -106,6 +106,13 @@ angular.module('flinkApp', ['ui.router', 'angularMoment'])
         templateUrl: "partials/jobs/job.plan.node-list.accumulators.html"
         controller: 'JobPlanAccumulatorsController'
 
+  .state "single-job.plan.checkpoints",
+    url: "/checkpoints"
+    views:
+      'node-details':
+        templateUrl: "partials/jobs/job.plan.node-list.checkpoints.html"
+        controller: 'JobPlanCheckpointsController'
+
   .state "single-job.timeline",
     url: "/timeline"
     views:

flink-runtime-web/web-dashboard/app/scripts/modules/jobs/jobs.ctrl.coffee

@@ -49,6 +49,8 @@ angular.module('flinkApp')
   $scope.job = null
   $scope.plan = null
   $scope.vertices = null
+  $scope.jobCheckpointStats = null
+  $scope.showHistory = false
 
   JobsService.loadJob($stateParams.jobid).then (data) ->
     $scope.job = data
@@ -67,6 +69,7 @@ angular.module('flinkApp')
     $scope.job = null
     $scope.plan = null
     $scope.vertices = null
+    $scope.jobCheckpointStats = null
 
     $interval.cancel(refresher)
 
@@ -75,6 +78,9 @@ angular.module('flinkApp')
     JobsService.cancelJob($stateParams.jobid).then (data) ->
       {}
 
+  $scope.toggleHistory = ->
+    $scope.showHistory = !$scope.showHistory
+
 # --------------------------------------
 
 .controller 'JobPlanController', ($scope, $state, $stateParams, JobsService) ->
@@ -90,6 +96,7 @@ angular.module('flinkApp')
       $scope.vertex = null
       $scope.subtasks = null
       $scope.accumulators = null
+      $scope.operatorCheckpointStats = null
 
       $scope.$broadcast 'reload'
 
@@ -99,13 +106,15 @@ angular.module('flinkApp')
       $scope.vertex = null
       $scope.subtasks = null
       $scope.accumulators = null
+      $scope.operatorCheckpointStats = null
 
   $scope.deactivateNode = ->
     $scope.nodeid = null
     $scope.nodeUnfolded = false
     $scope.vertex = null
     $scope.subtasks = null
     $scope.accumulators = null
+    $scope.operatorCheckpointStats = null
 
   $scope.toggleFold = ->
     $scope.nodeUnfolded = !$scope.nodeUnfolded
@@ -144,6 +153,33 @@ angular.module('flinkApp')
 
 # --------------------------------------
 
+.controller 'JobPlanCheckpointsController', ($scope, JobsService) ->
+  console.log 'JobPlanCheckpointsController'
+
+  # Get the per job stats
+  JobsService.getJobCheckpointStats($scope.jobid).then (data) ->
+    $scope.jobCheckpointStats = data
+
+  # Get the per operator stats
+  if $scope.nodeid and (!$scope.vertex or !$scope.vertex.operatorCheckpointStats)
+    JobsService.getOperatorCheckpointStats($scope.nodeid).then (data) ->
+      console.log('resvoled 1')
+      $scope.operatorCheckpointStats = data.operatorStats
+      $scope.subtasksCheckpointStats = data.subtasksStats
+
+  $scope.$on 'reload', (event) ->
+    console.log 'JobPlanCheckpointsController'
+
+    JobsService.getJobCheckpointStats($scope.jobid).then (data) ->
+      $scope.jobCheckpointStats = data
+
+    if $scope.nodeid
+      JobsService.getOperatorCheckpointStats($scope.nodeid).then (data) ->
+        $scope.operatorCheckpointStats = data.operatorStats
+        $scope.subtasksCheckpointStats = data.subtasksStats
+
+# --------------------------------------
+
 .controller 'JobTimelineVertexController', ($scope, $state, $stateParams, JobsService) ->
   console.log 'JobTimelineVertexController'
 

flink-runtime-web/web-dashboard/app/scripts/modules/jobs/jobs.svc.coffee

@@ -198,6 +198,40 @@ angular.module('flinkApp')
 
     deferred.promise
 
+  # Job-level checkpoint stats
+  @getJobCheckpointStats = (jobid) ->
+    deferred = $q.defer()
+
+    $http.get flinkConfig.jobServer + "jobs/" + jobid + "/checkpoints"
+    .success (data, status, headers, config) =>
+      if (angular.equals({}, data))
+        deferred.resolve(deferred.resolve(null))
+      else
+        deferred.resolve(data)
+
+    deferred.promise
+
+  # Operator-level checkpoint stats
+  @getOperatorCheckpointStats = (vertexid) ->
+    deferred = $q.defer()
+
+    deferreds.job.promise.then (data) =>
+      $http.get flinkConfig.jobServer + "jobs/" + currentJob.jid + "/vertices/" + vertexid + "/checkpoints"
+      .success (data) ->
+        # If no data available, we are done.
+        if (angular.equals({}, data))
+          deferred.resolve({ operatorStats: null, subtasksStats: null })
+        else
+          operatorStats = { id: data['id'], timestamp: data['timestamp'], duration: data['duration'], size: data['size'] }
+
+          if (angular.equals({}, data['subtasks']))
+            deferred.resolve({ operatorStats: operatorStats, subtasksStats: null })
+          else
+            subtaskStats = data['subtasks']
+            deferred.resolve({ operatorStats: operatorStats, subtasksStats: subtaskStats })
+
+    deferred.promise
+
   @loadExceptions = ->
     deferred = $q.defer()