Просмотр исходного кода

HUE-7420 [jb] impala kill job + filter

jdesjean 8 лет назад
Родитель
Сommit
db396f7

+ 13 - 2
apps/impala/src/impala/server.py

@@ -207,7 +207,7 @@ class ImpalaDaemonApi(object):
     try:
       return json.loads(resp)
     except ValueError, e:
-      raise ImpalaDaemonApiException('ImpalaDaemonApi query_profile did not return valid JSON.')
+      raise ImpalaDaemonApiException('ImpalaDaemonApi query_profile did not return valid JSON: %s' % e)
 
   def get_query_memory(self, query_id):
     params = {
@@ -219,4 +219,15 @@ class ImpalaDaemonApi(object):
     try:
       return json.loads(resp)
     except ValueError, e:
-      raise ImpalaDaemonApiException('ImpalaDaemonApi query_memory did not return valid JSON.')
+      raise ImpalaDaemonApiException('ImpalaDaemonApi query_memory did not return valid JSON: %s' % e)
+
+  def kill(self, query_id):
+    params = {
+      'query_id': query_id,
+      'json': 'true'
+    }
+    resp = self._root.get('cancel_query', params=params)
+    try:
+      return json.loads(resp)
+    except ValueError, e:
+      raise ImpalaDaemonApiException('ImpalaDaemonApi kill did not return valid JSON: %s' % e)

+ 6 - 2
apps/jobbrowser/src/jobbrowser/api2.py

@@ -70,8 +70,12 @@ def job(request):
   interface = json.loads(request.POST.get('interface'))
   app_id = json.loads(request.POST.get('app_id'))
 
-  response['app'] = get_api(request.user, interface).app(app_id)
-  response['status'] = 0
+  response_app = get_api(request.user, interface).app(app_id)
+  if response_app.get('status') == -1 and response_app.get('message'):
+    response.update(response_app)
+  else:
+    response['app'] = response_app
+    response['status'] = 0
 
   return JsonResponse(response)
 

+ 102 - 28
apps/jobbrowser/src/jobbrowser/apis/query_api.py

@@ -18,6 +18,8 @@
 import itertools
 import logging
 import re
+import time
+from datetime import datetime
 
 from django.utils.translation import ugettext as _
 
@@ -47,27 +49,39 @@ class QueryApi(Api):
 
     jobs = api.get_queries(**kwargs)
 
+    filter_list = self._get_filter_list(filters)
+    jobs_iter = itertools.chain(jobs['in_flight_queries'], jobs['completed_queries'])
+    jobs_iter_filtered = self._n_filter(filter_list, jobs_iter)
+
     return {
       'apps': [{
-        'id': app['query_id'],
-        'name': app['stmt'][:100] + ('...' if len(app['stmt']) > 100 else ''),
-        'status': app['state'],
-        'apiStatus': self._api_status(app['state']),
-        'type': app['stmt_type'],
-        'user': app['effective_user'],
-        'queue': app['resource_pool'],
-        'progress': app['progress'],
-        'duration': 0, # app['duration'],
-        'submitted': app['start_time'],
+        'id': job['query_id'],
+        'name': job['stmt'][:100] + ('...' if len(job['stmt']) > 100 else ''),
+        'status': job['state'],
+        'apiStatus': self._api_status(job['state']),
+        'type': job['stmt_type'],
+        'user': job['effective_user'],
+        'queue': job['resource_pool'],
+        'progress': job['progress'],
+        'canWrite': job in jobs['in_flight_queries'],
+        'duration': self._time_in_ms_groups(re.search(r"\s*(([\d.]*)([a-z]*))(([\d.]*)([a-z]*))?(([\d.]*)([a-z]*))?", job['duration'], re.MULTILINE).groups()),
+        'submitted': job['start_time'],
         # Extra specific
-        'rows_fetched': app['rows_fetched'],
-        'waiting': app['waiting'],
-        'waiting_time': app['waiting_time']
-      } for app in itertools.chain(jobs['in_flight_queries'], jobs['completed_queries'])],
+        'rows_fetched': job['rows_fetched'],
+        'waiting': job['waiting'],
+        'waiting_time': job['waiting_time']
+      } for job in jobs_iter_filtered],
       'total': jobs['num_in_flight_queries'] + jobs['num_executing_queries'] + jobs['num_waiting_queries']
     }
 
-  def time_in_ms(self, time, period):
+  def _time_in_ms_groups(self, groups):
+    time = 0
+    for x in range(0, len(groups), 3):
+      if groups[x+1]:
+        time += self._time_in_ms(groups[x+1], groups[x+2])
+    return time
+
+  def _time_in_ms(self, time, period):
     if period == 'ns':
       return float(time) / 1000
     elif period == 'ms':
@@ -78,6 +92,8 @@ class QueryApi(Api):
       return float(time) * 60000 #1000*60
     elif period == 'h':
       return float(time) * 3600000 #1000*60*60
+    elif period == 'd':
+      return float(time) * 86400000  # 1000*60*60*24
     else:
       return float(time)
 
@@ -85,15 +101,18 @@ class QueryApi(Api):
     api = get_impalad_api(user=self.user, url=self.server_url)
 
     query = api.get_query_profile(query_id=appid)
-    user = re.search(r"^\s*User:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
-    status = re.search(r"^\s*Query State:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
-    stmt = re.search(r"^\s*Sql Statement:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    if query.get('error'):
+      return {
+        'status': -1,
+        'message': query.get('error')
+      }
+
+    user = re.search(r"^\s*User:\s?([^\n\r]*)$", query['profile'], re.MULTILINE).group(1)
+    status = re.search(r"^\s*Query State:\s?([^\n\r]*)$$", query['profile'], re.MULTILINE).group(1)
+    stmt = re.search(r"^\s*Sql Statement:\s?([^\n\r]*)$$", query['profile'], re.MULTILINE).group(1)
     partitions = re.findall(r"partitions=\s*(\d)+\s*\/\s*(\d)+", query['profile'])
-    end_time = re.search(r"^\s*End Time:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
-    duration_1 = re.search(r"\s*Rows available:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
-    duration_2 = re.search(r"\s*Request finished:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
-    duration_3 = re.search(r"\s*Query Timeline:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
-    submitted = re.search(r"^\s*Start Time:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    end_time = re.search(r"^\s*End Time:\s?([^\n\r]*)$", query['profile'], re.MULTILINE).group(1)
+    submitted = re.search(r"^\s*Start Time:\s?([^\n\r]*)$", query['profile'], re.MULTILINE).group(1)
 
     progress = 0
     if end_time:
@@ -104,9 +123,10 @@ class QueryApi(Api):
       progress /= len(partitions)
       progress *= 100
 
-    duration = duration_1 or duration_2 or duration_3
-    if duration:
-      duration_ms = self.time_in_ms(duration.group(1), duration.group(2))
+    if end_time:
+      end_time_ms = int(time.mktime(datetime.strptime(end_time[:26], '%Y-%m-%d %H:%M:%S.%f').timetuple()))*1000
+      start_time_ms = int(time.mktime(datetime.strptime(submitted[:26], '%Y-%m-%d %H:%M:%S.%f').timetuple()))*1000
+      duration_ms = end_time_ms - start_time_ms
     else:
       duration_ms = 0
 
@@ -132,7 +152,20 @@ class QueryApi(Api):
 
 
   def action(self, appid, action):
-    return {}
+    message = {'message': '', 'status': 0}
+
+    if action.get('action') == 'kill':
+      api = get_impalad_api(user=self.user, url=self.server_url)
+
+      for _id in appid:
+        result = api.kill(_id)
+        if result.get('error'):
+          message['message'] = result.get('error')
+          message['status'] = -1
+        elif result.get('contents') and message.get('status') != -1:
+          message['message'] = result.get('contents')
+
+    return message;
 
 
   def logs(self, appid, app_type, log_name=None):
@@ -158,10 +191,51 @@ class QueryApi(Api):
     api = get_impalad_api(user=self.user, url=self.server_url)
     return api.get_query_profile(query_id=appid)
 
+  def _api_status_filter(self, status):
+    if status in ['RUNNING', 'CREATED']:
+      return 'RUNNING'
+    elif status in ['FINISHED']:
+      return 'COMPLETED'
+    else:
+      return 'FAILED'
+
   def _api_status(self, status):
     if status in ['RUNNING', 'CREATED']:
       return 'RUNNING'
     elif status in ['FINISHED']:
       return 'SUCCEEDED'
     else:
-      return 'FAILED' # INTERRUPTED , KILLED, TERMINATED and FAILED
+      return 'FAILED'
+
+  def _get_filter_list(self, filters):
+    filter_list = []
+    if filters.get("text"):
+      filter_names = {
+        'user':'effective_user',
+        'id':'query_id',
+        'name':'state',
+        'type':'stmt_type',
+        'status':'status'
+      }
+
+      def makeLambda(name, value):
+        return lambda app: app[name] == value
+
+      for key, name in filter_names.items():
+          text_filter = re.search(r"\s*("+key+")\s*:([^ ]+)", filters.get("text"))
+          if text_filter and text_filter.group(1) == key:
+            filter_list.append(makeLambda(name, text_filter.group(2).strip()))
+    if filters.get("time"):
+      time_filter = filters.get("time")
+      period_ms = self._time_in_ms(float(time_filter.get("time_value")), time_filter.get("time_unit")[0:1])
+      current_ms = time.time() * 1000.0
+      filter_list.append(lambda app: current_ms - (time.mktime(datetime.strptime(app['start_time'][:26], '%Y-%m-%d %H:%M:%S.%f').timetuple()) * 1000) < period_ms)
+    if filters.get("states"):
+      filter_list.append(lambda app: self._api_status_filter(app['state']).lower() in filters.get("states"))
+
+    return filter_list
+
+  def _n_filter(self, filters, tuples):
+    for f in filters:
+      tuples = filter(f, tuples)
+    return tuples

+ 4 - 22
apps/jobbrowser/src/jobbrowser/static/jobbrowser/js/impala_dagre.js

@@ -6,20 +6,15 @@ data_stream_target attribute.
 <script src="desktop/ext/js/dagre-d3-min.js"></script>
 Copied from https://github.com/apache/incubator-impala/blob/master/www/query_plan.tmpl
 */
-function impalaDagre(id, dataSource) {
+function impalaDagre(id) {
   var d3 = window.d3v3;
   var dagreD3 = window.dagreD3;
   var g = new dagreD3.graphlib.Graph().setGraph({rankDir: "BT"});
-  var svg = d3.select("#"+id);
+  var svg = d3.select("#"+id + " svg");
   var inner = svg.select("g");
   var _impalaDagree = {
-    start: function(refreshInterval) {
-      this._refreshInterval = refreshInterval;
-      // Force one refresh before starting the timer.
-      refresh();
-    },
-    stop: function() {
-      this._refreshInterval = null;
+    update: function(plan) {
+      renderGraph(plan);
     }
   };
 
@@ -147,18 +142,5 @@ function impalaDagre(id, dataSource) {
 
   }
 
-  // Called periodically, fetches the plan JSON from Impala and passes it to renderGraph()
-  // for display.
-  function refresh() {
-    if (!$("#"+id).parent().hasClass("active")) return;
-    dataSource().then(function(data) {
-      if (!$("#"+id).parent().hasClass("active")) return;
-      renderGraph(data);
-      if (_impalaDagree._refreshInterval) {
-        setTimeout(refresh, _impalaDagree._refreshInterval);
-      }
-    });
-  }
-
   return _impalaDagree;
 }

+ 29 - 49
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -859,7 +859,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
 <script type="text/html" id="queries-page${ SUFFIX }">
 
-  <div class="row-fluid">
+  <div class="row-fluid" data-jobType="queries">
     <div data-bind="css: {'span2': !$root.isMini(), 'span12': $root.isMini() }">
       <div class="sidebar-nav">
         <ul class="nav nav-list">
@@ -889,7 +889,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             </div>
           </li>
           <li class="nav-header">${ _('Duration') }</li>
-          <li><span data-bind="text: duration().toHHMMSS()"></span></li>
+          <li><span data-bind="text: duration() && duration().toHHMMSS()"></span></li>
           <li class="nav-header">${ _('Submitted') }</li>
           <li><span data-bind="moment: {data: submitted, format: 'LLL'}"></span></li>
         </ul>
@@ -899,7 +899,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
       <ul class="nav nav-pills margin-top-20">
         <li>
-          <a href="#queries-page-plan${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-plan${ SUFFIX }\']').tab('show'); }, event: {'shown': onQueriesPlanShown}">
+          <a href="#queries-page-plan${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-plan${ SUFFIX }\']').tab('show'); }, event: {'shown': function () { fetchProfile('plan'); } }">
             ${ _('Plan') }</a>
         </li>
         <li>
@@ -911,7 +911,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             ${ _('Text Plan') }</a>
         </li>
         <li>
-          <a href="#queries-page-summary${ SUFFIX }" data-bind="click: onQueriesSummaryClick">
+          <a href="#queries-page-summary${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-summary${ SUFFIX }\']').tab('show'); }">
             ${ _('Summary') }</a>
         </li>
         <li>
@@ -927,25 +927,25 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       <div class="clearfix"></div>
 
       <div class="tab-content">
-        <div class="tab-pane" id="queries-page-plan${ SUFFIX }">
+        <div class="tab-pane" id="queries-page-plan${ SUFFIX }" data-profile="plan" data-bind="impalaDagre: properties.plan() && properties.plan().plan_json">
           <svg style="border: 1px solid darkgray;width:100%;height:100%;" id="queries-page-plan-svg${ SUFFIX }">
             <g/>
           </svg>
         </div>
-        <div class="tab-pane" id="queries-page-stmt${ SUFFIX }">
-          <pre data-bind="text: properties.plan().stmt"/>
+        <div class="tab-pane" id="queries-page-stmt${ SUFFIX }" data-profile="plan">
+          <pre data-bind="text: properties.plan && properties.plan().stmt"/>
         </div>
-        <div class="tab-pane" id="queries-page-plan-text${ SUFFIX }">
-          <pre data-bind="text: properties.plan().plan"/>
+        <div class="tab-pane" id="queries-page-plan-text${ SUFFIX }" data-profile="plan">
+          <pre data-bind="text: properties.plan && properties.plan().plan"/>
         </div>
-        <div class="tab-pane" id="queries-page-summary${ SUFFIX }">
-          <pre data-bind="text: properties.plan().summary"/>
+        <div class="tab-pane" id="queries-page-summary${ SUFFIX }" data-profile="plan">
+          <pre data-bind="text: properties.plan && properties.plan().summary"/>
         </div>
-        <div class="tab-pane" id="queries-page-profile${ SUFFIX }">
-          <pre data-bind="text: properties.profile().profile"/>
+        <div class="tab-pane" id="queries-page-profile${ SUFFIX }" data-profile="profile">
+          <pre data-bind="text: properties.profile && properties.profile().profile"/>
         </div>
-        <div class="tab-pane" id="queries-page-memory${ SUFFIX }">
-          <pre data-bind="text: properties.memory().mem_usage"/>
+        <div class="tab-pane" id="queries-page-memory${ SUFFIX }" data-profile="mem_usage">
+          <pre data-bind="text: properties.memory && properties.memory().mem_usage"/>
         </div>
       </div>
     </div>
@@ -1710,34 +1710,6 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.logs = ko.observable('');
 
       self.properties = ko.mapping.fromJS(job.properties || {});
-      self.onQueriesPlanShown = function () {
-        if (this._impalaDagre) {
-          this._impalaDagre.stop();
-        }
-        var self = this;
-        var dataSource = function () {
-          return new Promise(function(resolve, reject) {
-            self.fetchProfile('plan', function (data) {
-              resolve(data.plan.plan_json);
-            });
-          });
-        }
-        this._impalaDagre = impalaDagre('queries-page-plan-svg${ SUFFIX }', dataSource);
-        this._impalaDagre.start(5000);
-      };
-      self.onQueriesSummaryClick = function(){
-        $('a[href=\'#queries-page-summary${ SUFFIX }\']').tab('show');
-        var self = this;
-        var refresh = function () {
-          if (!$('#queries-page-summary${ SUFFIX }').hasClass('active')) {
-            return;
-          }
-          self.fetchProfile('plan', function (data) {
-            setTimeout(refresh, 5000);
-          });
-        };
-        setTimeout(refresh, 5000);
-      };
       self.mainType = ko.observable(vm.interface());
 
       self.coordinatorActions = ko.pureComputed(function() {
@@ -1814,10 +1786,11 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.rerunModalContent = ko.observable('');
 
       self.hasKill = ko.pureComputed(function() {
-        return ['MAPREDUCE', 'SPARK', 'workflow', 'schedule', 'bundle'].indexOf(self.type()) != -1;
+        return ['MAPREDUCE', 'SPARK', 'workflow', 'schedule', 'bundle', 'QUERY'].indexOf(self.type()) != -1;
       });
       self.killEnabled = ko.pureComputed(function() {
-        return self.hasKill() && self.canWrite() && (self.apiStatus() == 'RUNNING' || self.apiStatus() == 'PAUSED');
+        //Impala can kill queries that are finished, but not yet terminated
+        return self.hasKill() && self.canWrite() && (vm.interface() === 'queries' || (self.apiStatus() == 'RUNNING' || self.apiStatus() == 'PAUSED'));
       });
 
       self.hasResume = ko.pureComputed(function() {
@@ -1940,7 +1913,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             crumbs.push({'id': vm.job().id(), 'name': vm.job().name(), 'type': vm.job().type()});
             vm.resetBreadcrumbs(crumbs);
             if (vm.job().type() === 'queries' && !$("#queries-page-plan${ SUFFIX }").parent().children().hasClass("active")) {
-              $("a[href=\'#queries-page-plan${ SUFFIX }\']").tab("show");
+              //show is still bound to old job, setTimeout allows knockout model change event done at begining of this method to sends it's notification
+              setTimeout(function () {
+                $("a[href=\'#queries-page-plan${ SUFFIX }\']").tab("show");
+              }, 0)
             }
             %if not is_mini:
             if (vm.job().type() === 'workflow' && !vm.job().workflowGraphLoaded) {
@@ -1966,7 +1942,6 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.updateJob = function () {
         vm.apiHelper.cancelActiveRequest(lastUpdateJobRequest);
         huePubSub.publish('graph.refresh.view');
-
         if (vm.job() == self && self.apiStatus() == 'RUNNING') {
           lastFetchJobRequest = self._fetchJob(function (data) {
             if (vm.job().type() == 'schedule') {
@@ -1975,7 +1950,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
               vm.job().fetchStatus();
               vm.job().fetchLogs();
             }
-            // vm.job().fetchProfile(); // Get name of active tab?
+            var profile = $("div[data-jobType] .tab-content .active").data("profile")
+            if (profile) {
+              vm.job().fetchProfile(profile);
+            }
           });
         }
       };
@@ -2027,6 +2005,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
             self.status(data.app.status);
             self.apiStatus(data.app.apiStatus);
             self.progress(data.app.progress);
+            self.canWrite(data.app.canWrite);
           } else {
             $(document).trigger("error", data.message);
           }
@@ -2147,7 +2126,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.selectedJobs = ko.observableArray();
 
       self.hasKill = ko.pureComputed(function() {
-        return ['jobs', 'workflows', 'schedules', 'bundles'].indexOf(vm.interface()) != -1 && !self.isCoordinator();
+        return ['jobs', 'workflows', 'schedules', 'bundles', 'queries'].indexOf(vm.interface()) != -1 && !self.isCoordinator();
       });
       self.killEnabled = ko.pureComputed(function() {
         return self.hasKill() && self.selectedJobs().length > 0 && $.grep(self.selectedJobs(), function(job) {
@@ -2322,6 +2301,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
                 if (self.apps()[i].status() != data.apps[j].status) {
                   self.apps()[i].status(data.apps[j].status);
                   self.apps()[i].apiStatus(data.apps[j].apiStatus);
+                  self.apps()[i].canWrite(data.apps[j].canWrite);
                 }
                 i++;
                 j++;

+ 12 - 0
desktop/core/src/desktop/static/desktop/js/ko.hue-bindings.js

@@ -6458,4 +6458,16 @@
     };
   })();
 
+  ko.bindingHandlers.impalaDagre = (function () {
+    return {
+      init: function (element, valueAccessor, allBindingsAccessor) {
+        var id = $(element).attr("id");
+        this._impalaDagre = impalaDagre(id);
+      },
+      update: function (element, valueAccessor) {
+        this._impalaDagre.update(ko.unwrap(valueAccessor()));
+      }
+    };
+  })();
+
 })();