Procházet zdrojové kódy

HUE-3797 [scheduler] Skeleton of Hive Scheduled queries browser

Next: scheduled executions
Romain před 5 roky
rodič
revize
29e079d7bb

+ 17 - 11
apps/jobbrowser/src/jobbrowser/apis/base_api.py

@@ -30,48 +30,54 @@ LOG = logging.getLogger(__name__)
 
 
 def get_api(user, interface, cluster=None):
-  from jobbrowser.apis.bundle_api import BundleApi
-  from jobbrowser.apis.data_eng_api import DataEngClusterApi, DataEngJobApi
-  from jobbrowser.apis.clusters import ClusterApi
-  from jobbrowser.apis.data_warehouse import DataWarehouseClusterApi
-  from jobbrowser.apis.history import HistoryApi
-  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.beeswax_query_api import BeeswaxQueryApi
-  from jobbrowser.apis.schedule_api import ScheduleApi
-  from jobbrowser.apis.workflow_api import WorkflowApi
 
   if interface == 'jobs':
+    from jobbrowser.apis.job_api import JobApi
     return JobApi(user)
   elif interface == 'queries-impala':
+    from jobbrowser.apis.query_api import QueryApi
     return QueryApi(user, cluster=cluster)
   elif interface == 'queries-hive':
+    from jobbrowser.apis.beeswax_query_api import BeeswaxQueryApi
     return BeeswaxQueryApi(user, cluster=cluster)
   elif interface == 'workflows':
+    from jobbrowser.apis.workflow_api import WorkflowApi
     return WorkflowApi(user)
   elif interface == 'schedules':
+    from jobbrowser.apis.schedule_api import ScheduleApi
     return ScheduleApi(user)
   elif interface == 'bundles':
+    from jobbrowser.apis.bundle_api import BundleApi
     return BundleApi(user)
   elif interface == 'celery-beat':
     from jobbrowser.apis.beat_api import BeatApi
     return BeatApi(user)
+  elif interface == 'schedule-hive':
+    from jobbrowser.apis.schedule_hive import HiveScheduleApi
+    return HiveScheduleApi(user)
   elif interface == 'history':
+    from jobbrowser.apis.history import HistoryApi
     return HistoryApi(user)
   elif interface == 'engines':
+    from jobbrowser.apis.clusters import ClusterApi
     return ClusterApi(user)
   elif interface == 'dataeng-clusters':
+    from jobbrowser.apis.data_eng_api import DataEngClusterApi
     return DataEngClusterApi(user)
   elif interface == 'dataware-clusters':
+    from jobbrowser.apis.data_warehouse import DataWarehouseClusterApi
     return DataWarehouseClusterApi(user)
   elif interface == 'dataware2-clusters':
+    from jobbrowser.apis.data_warehouse import DataWarehouseClusterApi
     return DataWarehouseClusterApi(user, version=2)
   elif interface == 'dataeng-jobs':
+    from jobbrowser.apis.data_eng_api import DataEngJobApi
     return DataEngJobApi(user)
   elif interface == 'livy-sessions':
+    from jobbrowser.apis.livy_api import LivySessionsApi
     return LivySessionsApi(user)
   elif interface == 'livy-job':
+    from jobbrowser.apis.livy_api import LivyJobApi
     return LivyJobApi(user)
   elif interface == 'slas':
     return Api(user)

+ 125 - 0
apps/jobbrowser/src/jobbrowser/apis/schedule_hive.py

@@ -0,0 +1,125 @@
+#!/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 logging
+
+from datetime import datetime
+
+from django.utils.translation import ugettext as _
+
+from dateutil import parser
+from desktop.lib.scheduler.lib.hive import HiveSchedulerApi
+
+from jobbrowser.apis.base_api import Api
+
+
+LOG = logging.getLogger(__name__)
+
+
+class HiveScheduleApi(Api):
+
+  def apps(self, filters):
+    api = HiveSchedulerApi(user=self.user)
+
+    tasks = api.list_tasks(self.user)
+
+    return {
+      'apps': [{
+          'id': 'schedule-hive-%(scheduled_query_id)s' % app,
+          'name': '%(schedule_name)s' % app,
+          'status': self._massage_status(app),
+          'apiStatus': self._api_status(self._massage_status(app)),
+          'type': 'schedule-hive',
+          'user': app['user'],
+          'progress': 50,
+          'queue': app['cluster_namespace'],
+          'canWrite': self.user.username == app['user'],
+          'duration': 1,
+          'submitted': app['enabled']
+        } for app in tasks
+      ],
+      'total': len(tasks)
+    }
+
+
+  def app(self, appid):
+    appid = appid.rsplit('-')[-1]
+    api = HiveSchedulerApi(user=self.user)
+
+    app = api.list_task(appid)
+
+    return {
+        'id': 'schedule-hive-%(scheduled_query_id)s' % app,
+        'name': '%(schedule_name)s' % app,
+        'status': self._massage_status(app),
+        'apiStatus': self._api_status(self._massage_status(app)),
+        'type': 'schedule-hive',
+        'user': app['user'],
+        'progress': 50,
+        'queue': app['cluster_namespace'],
+        'duration': 1,
+        'canWrite': self.user.username == app['user'],
+        'submitted': app['enabled'],
+        'properties': {
+        }
+    }
+
+
+  def action(self, app_ids, operation):
+    api = HiveSchedulerApi(user=self.user)
+
+    operations = []
+    actual_app_ids = [app_id.replace('schedule-hive-', '') for app_id in app_ids]
+
+    for app_id in actual_app_ids:
+      try:
+        api.action(app_id, operation['action'])
+        operations.append(app_id)
+      except Exception:
+        LOG.exception('Could not stop job %s' % app_id)
+
+    return {
+        'kills': operations,
+        'status': len(app_ids) - len(operations),
+        'message': _('%s signal sent to %s') % (operation['action'], operations)
+    }
+
+
+  def logs(self, appid, app_type, log_name=None, is_embeddable=False):
+    return {'logs': ''}
+
+
+  def profile(self, appid, app_type, app_property, app_filters):
+    appid = appid.rsplit('-')[-1]
+
+    if app_property == 'properties':
+      api = get_api(self.user)
+
+      return api.get_statements(appid)
+    else:
+      return {}
+
+
+  def _api_status(self, status):
+    if status == 'RUNNING':
+      return 'RUNNING'
+    else:
+      return 'PAUSED'
+
+
+  def _massage_status(self, task):
+    return 'RUNNING' if task['enabled'] else 'PAUSED'

+ 126 - 6
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -19,6 +19,7 @@ from django.utils.translation import ugettext as _
 from desktop.conf import CUSTOM, IS_K8S_ONLY
 from desktop.views import commonheader, commonfooter, _ko
 from metadata.conf import PROMETHEUS
+from notebook.conf import ENABLE_QUERY_SCHEDULING
 
 from jobbrowser.conf import DISABLE_KILLING_JOBS, MAX_JOB_FETCH, ENABLE_QUERY_BROWSER, ENABLE_HIVE_QUERY_BROWSER, ENABLE_HISTORY_V2
 %>
@@ -419,6 +420,10 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
                   <div class="jb-panel" data-bind="template: { name: 'celery-beat-page${ SUFFIX }' }"></div>
                 <!-- /ko -->
 
+                <!-- ko if: mainType() == 'schedule-hive' -->
+                  <div class="jb-panel" data-bind="template: { name: 'schedule-hive-page${ SUFFIX }' }"></div>
+                <!-- /ko -->
+
                 <!-- ko if: mainType() == 'workflows' -->
                   <!-- ko if: type() == 'workflow' -->
                     <div class="jb-panel" data-bind="template: { name: 'workflow-page${ SUFFIX }' }"></div>
@@ -1927,7 +1932,6 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       </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="#celery-beat-page-statements${ SUFFIX }" data-bind="click: function(){ fetchProfile('properties'); $('a[href=\'#celery-beat-page-statements${ SUFFIX }\']').tab('show'); }">
@@ -1968,6 +1972,84 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 </script>
 
 
+<script type="text/html" id="schedule-hive-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': apiStatus() === 'RUNNING', '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="#schedule-hive-page-statements${ SUFFIX }" data-bind="click: function(){ fetchProfile('properties'); $('a[href=\'#schedule-hive-statements${ SUFFIX }\']').tab('show'); }">
+            ${ _('Properties') }
+          </a>
+        </li>
+        <li class="pull-right" data-bind="template: { name: 'job-actions${ SUFFIX }' }"></li>
+      </ul>
+
+      <div class="clearfix"></div>
+
+      <div class="tab-content">
+        <div class="tab-pane active" id="schedule-hive-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="job-actions${ SUFFIX }">
   <div class="btn-group">
     <!-- ko if: $root.job() && $root.job().type() === 'schedule' -->
@@ -2824,7 +2906,20 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
       self.hasKill = ko.pureComputed(function() {
         return self.type() && (
-          ['MAPREDUCE', 'SPARK', 'workflow', 'schedule', 'bundle', 'QUERY', 'TEZ', 'YarnV2', 'DDL', 'celery-beat', 'history'].indexOf(self.type()) != -1 ||
+          [
+            'MAPREDUCE',
+            'SPARK',
+            'workflow',
+            'schedule',
+            'bundle',
+            'QUERY',
+            'TEZ',
+            'YarnV2',
+            'DDL',
+            'schedule-hive',
+            'celery-beat',
+            'history'
+          ].indexOf(self.type()) != -1 ||
           self.type().indexOf('Data Warehouse') != -1 ||
           self.type().indexOf('Altus') != -1
         );
@@ -2835,7 +2930,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       });
 
       self.hasResume = ko.pureComputed(function() {
-        return ['workflow', 'schedule', 'bundle', 'celery-beat', 'history'].indexOf(self.type()) != -1;
+        return ['workflow', 'schedule', 'bundle', 'schedule-hive', 'celery-beat', 'history'].indexOf(self.type()) != -1;
       });
       self.resumeEnabled = ko.pureComputed(function() {
         return self.hasResume() && self.canWrite() && self.apiStatus() == 'PAUSED';
@@ -2849,7 +2944,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       });
 
       self.hasPause = ko.pureComputed(function() {
-        return ['workflow', 'schedule', 'bundle', 'celery-beat', 'history'].indexOf(self.type()) != -1;
+        return ['workflow', 'schedule', 'bundle', 'celery-beat', 'schedule-hive', 'history'].indexOf(self.type()) != -1;
       });
       self.pauseEnabled = ko.pureComputed(function() {
         return self.hasPause() && self.canWrite() && self.apiStatus() == 'RUNNING';
@@ -2917,6 +3012,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         else if (/celery-beat-\w+/.test(self.id())) {
           interface = 'celery-beat';
         }
+        else if (/schedule-hive-\w+/.test(self.id())) {
+          interface = 'schedule-hive';
+        }
         else if (/altus:dataeng/.test(self.id()) && /:job:/.test(self.id())) {
           interface = 'dataeng-jobs';
         }
@@ -3403,6 +3501,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           'dataware-clusters',
           'dataware2-clusters',
           'celery-beat',
+          'schedule-hive',
           'history'
         ].indexOf(vm.interface()) != -1 && !self.isCoordinator();
       });
@@ -3413,7 +3512,15 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       });
 
       self.hasResume = ko.pureComputed(function() {
-        return ['workflows', 'schedules', 'bundles', 'dataware2-clusters', 'celery-beat', 'history'].indexOf(vm.interface()) != -1 && !self.isCoordinator();
+        return [
+          'workflows',
+          'schedules',
+          'bundles',
+          'dataware2-clusters',
+          'celery-beat',
+          'schedule-hive',
+          'history'
+        ].indexOf(vm.interface()) != -1 && !self.isCoordinator();
       });
       self.resumeEnabled = ko.pureComputed(function() {
         return self.hasResume() && self.selectedJobs().length > 0 && $.grep(self.selectedJobs(), function(job) {
@@ -3431,7 +3538,15 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       });
 
       self.hasPause = ko.pureComputed(function() {
-        return ['workflows', 'schedules', 'bundles', 'dataware2-clusters', 'celery-beat', 'history'].indexOf(vm.interface()) != -1 && !self.isCoordinator();
+        return [
+          'workflows',
+          'schedules',
+          'bundles',
+          'dataware2-clusters',
+          'celery-beat',
+          'schedule-hive',
+          'history'
+        ].indexOf(vm.interface()) != -1 && !self.isCoordinator();
       });
       self.pauseEnabled = ko.pureComputed(function() {
         return self.hasPause() && self.selectedJobs().length > 0 && $.grep(self.selectedJobs(), function(job) {
@@ -3777,6 +3892,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         var queryHiveInterfaceCondition = function () {
           return '${ ENABLE_HIVE_QUERY_BROWSER.get() }' == 'True' && self.appConfig() && self.appConfig()['editor'] && self.appConfig()['editor']['interpreter_names'].indexOf('hive') != -1 && (!self.cluster() || self.cluster()['type'].indexOf('altus') == -1);
         };
+        var scheduleHiveInterfaceCondition = function () {
+          return '${ ENABLE_QUERY_SCHEDULING.get() }' == 'True';
+        };
 
         var interfaces = [
           {'interface': 'jobs', 'label': '${ _ko('Jobs') }', 'condition': jobsInterfaceCondition},
@@ -3787,6 +3905,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           {'interface': 'engines', 'label': '${ _ko('') }', 'condition': enginesInterfaceCondition},
           {'interface': 'queries-impala', 'label': '${ _ko('Impala') }', 'condition': queryInterfaceCondition},
           {'interface': 'queries-hive', 'label': '${ _ko('Hive') }', 'condition': queryHiveInterfaceCondition},
+          {'interface': 'schedule-hive', 'label': '${ _ko('Hive Schedules') }', 'condition': scheduleHiveInterfaceCondition},
           {'interface': 'celery-beat', 'label': '${ _ko('Scheduled Tasks') }', 'condition': schedulerBeatInterfaceCondition},
           {'interface': 'history', 'label': '${ _ko('History') }', 'condition': historyInterfaceCondition},
           {'interface': 'workflows', 'label': '${ _ko('Workflows') }', 'condition': schedulerInterfaceCondition},
@@ -3979,6 +4098,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
           case 'queries-impala':
           case 'queries-hive':
           case 'celery-beat':
+          case 'schedule-hive':
           case 'history':
           case 'workflows':
           case 'schedules':

+ 8 - 2
desktop/core/src/desktop/lib/scheduler/api.py

@@ -49,10 +49,11 @@ def new_schedule(request):
 def get_schedule(request):
   return edit_coordinator(request)
 
+
 # To move to lib in case oozie is blacklisted
 #@check_document_access_permission()
 def submit_schedule(request, doc_id):
-  interface = request.GET.get('interface', request.POST.get('interface', 'beat'))
+  interface = request.GET.get('interface', request.POST.get('interface', 'hive'))
 
   if doc_id.isdigit():
     coordinator = Coordinator(document=Document2.objects.get(id=doc_id))
@@ -74,7 +75,12 @@ def submit_schedule(request, doc_id):
         message = force_unicode(str(e))
         return JsonResponse({'status': -1, 'message': message}, safe=False)
       if jsonify:
-        schedule_type = 'celery-beat' if interface == 'beat' else 'schedule'
+        if interface == 'hive':
+          schedule_type = 'schedule-hive'  # Current Job Browser "types"
+        elif interface == 'beat':
+          schedule_type = 'celery-beat'
+        else:
+          schedule_type = 'schedule'
         return JsonResponse({'status': 0, 'job_id': job_id, 'type': schedule_type}, safe=False)
       else:
         request.info(_('Schedule submitted.'))

+ 4 - 1
desktop/core/src/desktop/lib/scheduler/lib/api.py

@@ -19,7 +19,10 @@ from builtins import object
 
 
 def get_api(request, interface):
-  if interface == 'beat':
+  if interface == 'hive':
+    from desktop.lib.scheduler.lib.hive import HiveSchedulerApi
+    return HiveSchedulerApi(user=request.user)
+  elif interface == 'beat':
     from desktop.lib.scheduler.lib.beat import CeleryBeatApi
     return CeleryBeatApi(user=request.user)
   elif interface == 'oozie':

+ 158 - 0
desktop/core/src/desktop/lib/scheduler/lib/hive.py

@@ -0,0 +1,158 @@
+#!/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 json
+
+from notebook.models import make_notebook, MockRequest
+
+from desktop.lib.scheduler.lib.api import Api
+from desktop.models import Document2
+
+
+class HiveSchedulerApi(Api):
+  """
+  * create
+  create scheduled query sc1 cron '0 */10 * * * ? *' as select 1
+
+  * change schedule
+  alter scheduled query Q1 cron '2 2 * * *'
+
+  * change query
+  alter scheduled query Q1 defined as select 2
+
+  * disable
+  alter scheduled query Q1 set disabled
+
+  * enable
+  alter scheduled query Q1 set enabled
+
+  * list status
+  select * from information_schema.scheduled_queries;
+
+  * drop
+  drop scheduled query Q1
+  """
+
+  def submit_schedule(self, request, coordinator, mapping):
+    """
+    coordinator
+      Document2.objects.get(uuid=coordinator.get_data_for_json()['properties']['document'])
+
+    mapping
+      {u'oozie.use.system.libpath': u'True', 'dryrun': False, u'start_date': u'2019-08-10T17:02', u'end_date': u'2019-08-17T17:02'}
+    """
+
+    document = Document2.objects.get(uuid=coordinator.get_data_for_json()['properties']['document'])  # Assumes Hive SQL queries
+
+    # (schedule_name,cluster_namespace) is unique
+    #_get_snippet_name(notebook) --> name
+
+    properties = {
+      'name': 'query-%(uuid)s' % {
+        'uuid': document.uuid
+      },
+      'username': request.user.username
+    }
+
+    sql_query = """
+    CREATE SCHEDULED QUERY %(name)s
+    CRON '1 1 * * *' AS
+    SELECT 1
+    """ % properties
+
+    job = make_notebook(
+        name=properties['name'],
+        editor_type='hive',
+        statement=sql_query,
+        status='ready',
+        database='default',
+        is_task=False,
+    )
+    handle = job.execute_and_wait(request)
+
+    return handle['history_uuid']
+
+
+  def list_tasks(self, user):
+    sql_query = "SELECT * FROM information_schema.scheduled_queries"
+
+    job = make_notebook(
+        name='List Hive schedules',
+        editor_type='hive',
+        statement=sql_query,
+        status='ready',
+        database='default',
+        is_task=False,
+    )
+    request = MockRequest(user)
+
+    handle = job.execute_and_wait(request, include_results=True)
+
+    return [
+      self._get_task(row) for row in handle['result']['data']
+    ]
+
+  def list_task(self, task_id):
+    task_id = task_id.replace('schedule-hive-', '')
+
+    sql_query = """
+    SELECT * FROM information_schema.scheduled_queries
+    WHERE scheduled_query_id = %(scheduled_query_id)s
+    """ % {
+      'scheduled_query_id': task_id
+    }
+
+    job = make_notebook(
+        name='List Hive schedule id',
+        editor_type='hive',
+        statement=sql_query,
+        status='ready',
+        database='default',
+        is_task=False,
+    )
+    request = MockRequest(self.user)
+
+    handle = job.execute_and_wait(request, include_results=True)
+
+    return self._get_task(handle['result']['data'][0])
+
+
+  def action(self, schedule_id, action='suspend'):
+    task = PeriodicTask.objects.get(id=schedule_id, description=self.user.username)
+
+    if action == 'suspend':
+      task.enabled = False
+      task.save()
+    elif action == 'resume':
+      task.enabled = True
+      task.save()
+    elif action == 'kill':
+      task.delete()
+
+
+  def _get_task(self, row):
+    return {
+      'scheduled_query_id': row[0],
+      'schedule_name': row[1],
+      'enabled': row[2],
+      'cluster_namespace': row[3],
+      'schedule': row[4],
+      'user': row[5],
+      'query': row[6],
+      'next_execution': row[7],
+      'active_execution_id': row[8],
+    }

+ 16 - 10
desktop/libs/notebook/src/notebook/api.py

@@ -42,7 +42,7 @@ from metadata.conf import OPTIMIZER
 from notebook.connectors.base import Notebook, QueryExpired, SessionExpired, QueryError, _get_snippet_name, patch_snippet_for_connector
 from notebook.connectors.hiveserver2 import HS2Api
 from notebook.decorators import api_error_handler, check_document_access_permission, check_document_modify_permission
-from notebook.models import escape_rows, make_notebook, upgrade_session_properties, get_api
+from notebook.models import escape_rows, make_notebook, upgrade_session_properties, get_api, MockRequest
 
 if sys.version_info[0] > 2:
   from urllib.parse import unquote as urllib_unquote
@@ -296,8 +296,6 @@ def _check_status(request, notebook=None, snippet=None, operation_id=None):
 @check_document_access_permission
 @api_error_handler
 def fetch_result_data(request):
-  response = {'status': -1}
-
   operation_id = request.POST.get('operationId')
   notebook = json.loads(request.POST.get('notebook', '{}'))
   snippet = json.loads(request.POST.get('snippet', '{}'))
@@ -305,25 +303,33 @@ def fetch_result_data(request):
   rows = json.loads(request.POST.get('rows', '100'))
   start_over = json.loads(request.POST.get('startOver', 'false'))
 
-  snippet = _get_snippet(request.user, notebook, snippet, operation_id)
-
   with opentracing.tracer.start_span('notebook-fetch_result_data') as span:
-    response['result'] = get_api(request, snippet).fetch_result(notebook, snippet, rows, start_over)
-
     span.set_tag('user-id', request.user.username)
     span.set_tag(
       'query-id',
       snippet['result']['handle']['guid'] if snippet['result'].get('handle') and snippet['result']['handle'].get('guid') else None
     )
 
+    response = _fetch_result_data(request.user, notebook, snippet, operation_id, rows=rows, start_over=start_over)
+    response['status'] = 0
+
+    return JsonResponse(response)
+
+
+def _fetch_result_data(user, notebook=None, snippet=None, operation_id=None, rows=100, start_over=False):
+  snippet = _get_snippet(user, notebook, snippet, operation_id)
+  request = MockRequest(user)
+
+  response = {
+    'result': get_api(request, snippet).fetch_result(notebook, snippet, rows, start_over)
+  }
+
   # Materialize and HTML escape results
   if response['result'].get('data') and response['result'].get('type') == 'table' and not response['result'].get('isEscaped'):
     response['result']['data'] = escape_rows(response['result']['data'])
     response['result']['isEscaped'] = True
 
-  response['status'] = 0
-
-  return JsonResponse(response)
+  return response
 
 
 @require_POST

+ 41 - 31
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -289,50 +289,60 @@ class Notebook(object):
     return _execute_notebook(request, notebook_data, snippet)
 
 
-  def execute_and_wait(self, request, timeout_sec=30.0, sleep_interval=0.5):
-      """
-      Run query and check status until it finishes or timeouts.
+  def execute_and_wait(self, request, timeout_sec=30.0, sleep_interval=0.5, include_results=False):
+    """
+    Run query and check status until it finishes or timeouts.
 
-      Check status until it finishes or timeouts.
-      """
-      handle = self.execute(request, batch=False)
+    Check status until it finishes or timeouts.
+    """
+    handle = self.execute(request, batch=False)
 
-      if handle['status'] != 0:
-        raise QueryError(e, message='SQL statement failed.', handle=handle)
+    if handle['status'] != 0:
+      raise QueryError(e, message='SQL statement failed.', handle=handle)
 
-      operation_id = handle['history_uuid']
-      curr = time.time()
-      end = curr + timeout_sec
+    operation_id = handle['history_uuid']
+    curr = time.time()
+    end = curr + timeout_sec
 
-      status = self.check_status(request, operation_id=operation_id)
+    handle = self.check_status(request, operation_id=operation_id)
 
-      while curr <= end:
-        if status['status'] not in ('waiting', 'running'):
-          return handle
+    while curr <= end:
+      if handle['status'] == 0 and handle['query_status']['status'] not in ('waiting', 'running'):
+        if include_results and handle['query_status']['status'] == 'available':
+          handle.update(
+            self.fetch_result_data(request.user, operation_id=operation_id)
+          )
+          # TODO: close
+        return handle
 
-        status = self.check_status(request, operation_id=operation_id)
-        time.sleep(sleep_interval)
-        curr = time.time()
+      status = self.check_status(request, operation_id=operation_id)
+      time.sleep(sleep_interval)
+      curr = time.time()
 
-      # TODO
-      # msg = "The query timed out after %(timeout)d seconds, canceled query." % {'timeout': timeout_sec}
-      # LOG.warning(msg)
-      # try:
-      #   self.cancel_operation(handle)
-      #   # get_api(request, snippet).cancel(notebook, snippet)
-      # except Exception as e:
-      #   msg = "Failed to cancel query."
-      #   LOG.warning(msg)
-      #   self.close_operation(handle)
-      #   raise QueryServerException(e, message=msg)
+    # TODO
+    # msg = "The query timed out after %(timeout)d seconds, canceled query." % {'timeout': timeout_sec}
+    # LOG.warning(msg)
+    # try:
+    #   self.cancel_operation(handle)
+    #   # get_api(request, snippet).cancel(notebook, snippet)
+    # except Exception as e:
+    #   msg = "Failed to cancel query."
+    #   LOG.warning(msg)
+    #   self.close_operation(handle)
+    #   raise QueryServerException(e, message=msg)
 
-      raise OperationTimeout()
+    raise OperationTimeout()
 
   def check_status(self, request, operation_id):
-    from notebook.api import _check_status  # Cyclic dependency
+    from notebook.api import _check_status
 
     return _check_status(request, operation_id=operation_id)
 
+  def fetch_result_data(self, user, operation_id):
+    from notebook.api import _fetch_result_data
+
+    return _fetch_result_data(user, operation_id=operation_id, rows=100, start_over=False)
+
 
 def get_interpreter(connector_type, user=None):
   interpreter = [

+ 6 - 0
desktop/libs/notebook/src/notebook/models.py

@@ -623,3 +623,9 @@ class Analytics(object):
     # Could count number of "forks" (but would need to start tracking parent of Saved As query cf. saveAsNotebook)
 
     return stats
+
+
+class MockRequest():
+  def __init__(self, user, ):
+    self.user = user
+    self.POST = {}