Browse Source

HUE-7316 [jb] Skeleton of Impala Query integration into Job Browser

Notes
- updateJobs won't work as currently expecting chronological ids
- Job page not done
Romain Rigaux 8 năm trước cách đây
mục cha
commit
b1dfff5

+ 1 - 1
apps/impala/src/impala/impala_flags.py

@@ -88,4 +88,4 @@ def _parse_impala_flags():
     _IMPALA_FLAGS = {}
   except Exception, ex:
     LOG.error('Failed to parse Impala config from "%s": %s' % (impala_flags_path, ex))
-    _IMPALA_FLAGS = {}
+    _IMPALA_FLAGS = {}

+ 26 - 3
apps/impala/src/impala/server.py

@@ -26,6 +26,7 @@ from beeswax.server.dbms import QueryServerException
 from beeswax.server.hive_server2_lib import HiveServerClient
 
 from ImpalaService import ImpalaHiveServer2Service
+from impala.impala_flags import get_webserver_certificate_file
 
 
 LOG = logging.getLogger(__name__)
@@ -47,6 +48,14 @@ def get_api(user, url):
   return API_CACHE
 
 
+def _get_impala_server_url(session):
+  impala_settings = session.get_formatted_properties()
+  http_addr = next((setting['value'] for setting in impala_settings if setting['key'].lower() == 'http_addr'), None)
+  # Remove scheme if found
+  http_addr = http_addr.replace('http://', '').replace('https://', '')
+  return ('https://' if get_webserver_certificate_file() else 'http://') + http_addr
+
+
 class ImpalaServerClientException(Exception):
   pass
 
@@ -163,15 +172,29 @@ class ImpalaDaemonApi(object):
       self._thread_local.user = user
 
 
+  def get_queries(self):
+    params = {
+      'json': 'true'
+    }
+
+    resp = self._root.get('queries', params=params)
+    try:
+      return json.loads(resp)
+    except ValueError, e:
+      raise ImpalaDaemonApiException('ImpalaDaemonApi did not return valid JSON: %s' % e)
+
+
+  def get_query(self, query_id): pass
+
+
   def get_query_profile(self, query_id):
     params = {
       'query_id': query_id,
       'json': 'true'
     }
-    profile = None
+
     resp = self._root.get('query_profile', params=params)
     try:
-      profile = json.loads(resp)
+      return json.loads(resp)
     except ValueError, e:
       raise ImpalaDaemonApiException('ImpalaDaemonApi query_profile did not return valid JSON.')
-    return profile

+ 3 - 0
apps/jobbrowser/src/jobbrowser/apis/base_api.py

@@ -33,11 +33,14 @@ def get_api(user, interface):
   from jobbrowser.apis.data_eng_api import DataEngClusterApi, DataEngJobApi
   from jobbrowser.apis.livy_api import LivySessionsApi, LivyJobApi
   from jobbrowser.apis.job_api import JobApi
+  from jobbrowser.apis.query_api import QueryApi
   from jobbrowser.apis.schedule_api import ScheduleApi
   from jobbrowser.apis.workflow_api import WorkflowApi
 
   if interface == 'jobs':
     return JobApi(user)
+  elif interface == 'queries':
+    return QueryApi(user)
   elif interface == 'workflows':
     return WorkflowApi(user)
   elif interface == 'schedules':

+ 2 - 14
apps/jobbrowser/src/jobbrowser/apis/job_api.py

@@ -43,15 +43,11 @@ class JobApi(Api):
 
   def __init__(self, user):
     self.user = user
-    self.yarn_api = YarnApi(user) # TODO: actually long term move job aggregations to the frontend instead probably
-    self.impala_api = ImpalaApi(user)
+    self.yarn_api = YarnApi(user)
     self.request = None
 
   def apps(self, filters):
-    jobs = self.yarn_api.apps(filters)
-    # += Impala
-    # += Sqoop2
-    return jobs
+    return self.yarn_api.apps(filters)
 
   def app(self, appid):
     return self._get_api(appid).app(appid)
@@ -426,11 +422,3 @@ class YarnMapReduceTaskAttemptApi(Api):
 
 class YarnAtsApi(Api):
   pass
-
-
-class ImpalaApi(Api):
-  pass
-
-
-class Sqoop2Api(Api):
-  pass

+ 109 - 0
apps/jobbrowser/src/jobbrowser/apis/query_api.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. 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.
+
+import itertools
+import logging
+
+from django.utils.translation import ugettext as _
+
+from jobbrowser.apis.base_api import Api
+
+LOG = logging.getLogger(__name__)
+
+
+try:
+  from beeswax.models import Session
+  from impala.server import get_api as get_impalad_api, _get_impala_server_url
+except Exception, e:
+  LOG.exception('Some application are not enabled: %s' % e)
+
+
+class QueryApi(Api):
+
+  def __init__(self, user):
+    self.user = user
+    session = Session.objects.get_session(self.user, application='impala')
+    self.server_url = _get_impala_server_url(session)
+
+  def apps(self, filters):
+    kwargs = {}
+
+    api = get_impalad_api(user=self.user, url=self.server_url)
+
+    jobs = api.get_queries(**kwargs)
+
+    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'],
+        # 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'])],
+      'total': jobs['num_in_flight_queries'] + jobs['num_executing_queries'] + jobs['num_waiting_queries']
+    }
+
+  def app(self, appid):
+    api = get_impalad_api(user=self.user, url=self.server_url)
+
+    job = api.get_query(query_id=appid) # TODO
+
+    common = {
+        'id': job['jobId'],
+        'name': job['jobId'],
+        'status': job['status'],
+        'apiStatus': self._api_status(job['status']),
+        'progress': 50,
+        'duration': 10 * 3600,
+        'submitted': job['creationDate'],
+        'type': 'dataeng-job-%s' % job['jobType'],
+    }
+
+    common['properties'] = {
+      'properties': job
+    }
+
+    return common
+
+
+  def action(self, appid, action):
+    return {}
+
+
+  def logs(self, appid, app_type, log_name=None):
+    return {'logs': ''}
+
+
+  def profile(self, appid, app_type, app_property):
+    return {}
+
+  def _api_status(self, status):
+    if status in ['CREATING', 'CREATED', 'TERMINATING']:
+      return 'RUNNING'
+    elif status in ['COMPLETED']:
+      return 'SUCCEEDED'
+    else:
+      return 'FAILED' # INTERRUPTED , KILLED, TERMINATED and FAILED

+ 8 - 1
apps/jobbrowser/src/jobbrowser/conf.py

@@ -43,7 +43,7 @@ ENABLE_V2 = Config(
     key="enable_v2",
     help=_("Show the version 2 of app which unifies all the past browsers into one."),
     type=coerce_bool,
-    default=False
+    default=True
 )
 
 MAX_JOB_FETCH = Config(
@@ -52,3 +52,10 @@ MAX_JOB_FETCH = Config(
   type=int,
   help=_('Maximum number of jobs to fetch and display when pagination is not supported for the type.')
 )
+
+ENABLE_QUERY_BROWSER = Config(
+    key="enable_query_browser",
+    help=_("Show the query section for listing and showing more troubleshooting information."),
+    type=coerce_bool,
+    default=False
+)

+ 89 - 1
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -18,7 +18,7 @@
 
   from desktop import conf
   from desktop.views import commonheader, commonfooter, _ko
-  from jobbrowser.conf import DISABLE_KILLING_JOBS, MAX_JOB_FETCH
+  from jobbrowser.conf import DISABLE_KILLING_JOBS, MAX_JOB_FETCH, ENABLE_QUERY_BROWSER
 %>
 
 <%
@@ -227,6 +227,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
               <div class="jb-panel" data-bind="template: { name: 'job-page${ SUFFIX }' }"></div>
             <!-- /ko -->
 
+            <!-- ko if: mainType() == 'queries' -->
+              <div class="jb-panel" data-bind="template: { name: 'queries-page${ SUFFIX }' }"></div>
+            <!-- /ko -->
+
             <!-- ko if: mainType() == 'workflows' -->
               <!-- ko if: type() == 'workflow' -->
                 <div class="jb-panel" data-bind="template: { name: 'workflow-page${ SUFFIX }' }"></div>
@@ -848,6 +852,85 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 </script>
 
 
+<script type="text/html" id="queries-page${ SUFFIX }">
+
+  <div class="row-fluid">
+    <div data-bind="css: {'span2': !$root.isMini(), 'span12': $root.isMini() }">
+      <div class="sidebar-nav">
+        <ul class="nav nav-list">
+          <li class="nav-header">${ _('Id') }</li>
+          <li class="break-word"><span data-bind="text: id"></span></li>
+          <!-- ko if: doc_url -->
+          <li class="nav-header">${ _('Document') }</li>
+          <li>
+            <a data-bind="hueLink: doc_url" href="javascript: void(0);" title="${ _('Open in editor') }">
+              <span data-bind="text: name"></span>
+            </a>
+          </li>
+          <!-- /ko -->
+          <!-- ko ifnot: doc_url -->
+          <li class="nav-header">${ _('Name') }</li>
+          <li><span data-bind="text: name"></span></li>
+          <!-- /ko -->
+          <li class="nav-header">${ _('Status') }</li>
+          <li><span data-bind="text: status"></span></li>
+          <li class="nav-header">${ _('User') }</li>
+          <li><span data-bind="text: user"></span></li>
+          <li class="nav-header">${ _('Progress') }</li>
+          <li><span data-bind="text: progress"></span>%</li>
+          <li>
+            <div class="progress-job progress" style="background-color: #FFF; width: 100%" data-bind="css: {'progress-danger': apiStatus() === 'FAILED', 'progress-warning': isRunning(), 'progress-success': apiStatus() === 'SUCCEEDED' }">
+              <div class="bar" data-bind="style: {'width': progress() + '%'}"></div>
+            </div>
+          </li>
+          <li class="nav-header">${ _('Duration') }</li>
+          <li><span data-bind="text: duration().toHHMMSS()"></span></li>
+          <li class="nav-header">${ _('Submitted') }</li>
+          <li><span data-bind="moment: {data: submitted, format: 'LLL'}"></span></li>
+        </ul>
+      </div>
+    </div>
+    <div data-bind="css:{'span10': !$root.isMini(), 'span12 no-margin': $root.isMini() }">
+
+      <ul class="nav nav-pills margin-top-20">
+        <li>
+          <a href="#livy-session-page-statements${ SUFFIX }" data-bind="click: function(){ fetchProfile('properties'); $('a[href=\'#livy-session-page-statements${ SUFFIX }\']').tab('show'); }">
+            ${ _('Properties') }</a>
+        </li>
+      </ul>
+
+      <div class="clearfix"></div>
+
+      <div class="tab-content">
+        <div class="tab-pane active" id="livy-session-page-statements${ SUFFIX }">
+          <table id="actionsTable" class="datatables table table-condensed">
+            <thead>
+            <tr>
+              <th>${_('Id')}</th>
+              <th>${_('State')}</th>
+              <th>${_('Output')}</th>
+            </tr>
+            </thead>
+            <tbody data-bind="foreach: properties['statements']">
+              <tr data-bind="click: function() {  $root.job().id(id); $root.job().fetchJob(); }" class="pointer">
+                <td>
+                  <a data-bind="hueLink: '/jobbrowser/jobs/' + id(), clickBubble: false">
+                    <i class="fa fa-tasks"></i>
+                  </a>
+                </td>
+                <td data-bind="text: state"></td>
+                <td data-bind="text: output"></td>
+              </tr>
+            </tbody>
+          </table>
+        </div>
+
+      </div>
+    </div>
+  </div>
+</script>
+
+
 <script type="text/html" id="livy-session-page${ SUFFIX }">
 
   <div class="row-fluid">
@@ -2281,9 +2364,13 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         var livyInterfaceCondition = function () {
           return self.appConfig() && self.appConfig()['editor'] && self.appConfig()['editor']['interpreter_names'].indexOf('pyspark') != -1;
         }
+        var queryInterfaceCondition = function () {
+          return '${ ENABLE_QUERY_BROWSER.get() }' == 'True' && self.appConfig() && self.appConfig()['editor'] && self.appConfig()['editor']['interpreter_names'].indexOf('impala') != -1;
+        }
 
         var interfaces = [
           {'interface': 'jobs', 'label': '${ _ko('Jobs') }', 'condition': jobsInterfaceCondition},
+          {'interface': 'queries', 'label': '${ _ko('Queries') }', 'condition': queryInterfaceCondition},
           {'interface': 'dataeng-jobs', 'label': '${ _ko('Jobs') }', 'condition': dataEngInterfaceCondition},
           {'interface': 'workflows', 'label': '${ _ko('Workflows') }', 'condition': schedulerInterfaceCondition},
           {'interface': 'schedules', 'label': '${ _ko('Schedules') }', 'condition': schedulerInterfaceCondition},
@@ -2437,6 +2524,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           case 'slas':
           case 'oozie-info':
           case 'jobs':
+          case 'queries':
           case 'workflows':
           case 'schedules':
           case 'bundles':

+ 4 - 1
desktop/conf.dist/hue.ini

@@ -1313,7 +1313,10 @@
   ## max_job_fetch=500
 
   # Show the version 2 of app which unifies all the past browsers into one.
-  ## enable_v2=false
+  ## enable_v2=true
+
+  # Show the query section for listing and showing more troubleshooting information.
+  ## enable_query_browser=false
 
 
 ###########################################################################

+ 4 - 1
desktop/conf/pseudo-distributed.ini.tmpl

@@ -1312,7 +1312,10 @@
   ## max_job_fetch=500
 
   # Show the version 2 of app which unifies all the past browsers into one.
-  ## enable_v2=false
+  ## enable_v2=true
+
+  # Show the query section for listing and showing more troubleshooting information.
+  ## enable_query_browser=false
 
 
 ###########################################################################

+ 3 - 11
desktop/libs/notebook/src/notebook/connectors/hiveserver2.py

@@ -61,7 +61,7 @@ try:
   from impala import api   # Force checking if Impala is enabled
   from impala.conf import CONFIG_WHITELIST as impala_settings, SSL as impala_ssl_conf
   from impala.impala_flags import get_webserver_certificate_file
-  from impala.server import get_api as get_impalad_api, ImpalaDaemonApiException
+  from impala.server import get_api as get_impalad_api, ImpalaDaemonApiException, _get_impala_server_url
 except ImportError, e:
   LOG.warn("Impala app is not enabled")
   impala_settings = None
@@ -186,7 +186,7 @@ class HS2Api(Api):
     response['properties'] = properties
 
     if lang == 'impala':
-      http_addr = self._get_impala_server_url(session)
+      http_addr = _get_impala_server_url(session)
       response['http_addr'] = http_addr
 
     return response
@@ -797,7 +797,7 @@ DROP TABLE IF EXISTS `%(table)s`;
 
     query_id = self._get_impala_query_id(snippet)
     session = Session.objects.get_session(self.user, application='impala')
-    server_url = self._get_impala_server_url(session)
+    server_url = _get_impala_server_url(session)
     if query_id:
       LOG.info("Attempting to get Impala query profile at server_url %s for query ID: %s" % (server_url, query_id))
 
@@ -824,14 +824,6 @@ DROP TABLE IF EXISTS `%(table)s`;
     return guid
 
 
-  def _get_impala_server_url(self, session):
-    impala_settings = session.get_formatted_properties()
-    http_addr = next((setting['value'] for setting in impala_settings if setting['key'].lower() == 'http_addr'), None)
-    # Remove scheme if found
-    http_addr = http_addr.replace('http://', '').replace('https://', '')
-    return ('https://' if get_webserver_certificate_file() else 'http://') + http_addr
-
-
   def _get_impala_query_profile(self, server_url, query_id):
     api = get_impalad_api(user=self.user, url=server_url)