Преглед на файлове

HUE-3797 [scheduler] Refactor the scheduling API to be more abstracted

Romain преди 6 години
родител
ревизия
a8deaff734

+ 22 - 3
desktop/core/src/desktop/celery.py

@@ -1,8 +1,15 @@
 from __future__ import absolute_import, unicode_literals
+
 import os
+
 from celery import Celery
+from celery.schedules import crontab
+
+from desktop.settings import TIME_ZONE
+from desktop.conf import TASK_SERVER
 
-# set the default Django settings module for the 'celery' program.
+
+# Set the default Django settings module for the 'celery' program.
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'desktop.settings')
 
 app = Celery('desktop')
@@ -12,6 +19,7 @@ app = Celery('desktop')
 # - namespace='CELERY' means all celery-related configuration keys
 #   should have a `CELERY_` prefix.
 app.config_from_object('django.conf:settings', namespace='CELERY')
+app.conf.timezone = TIME_ZONE
 
 # Load task modules from all registered Django app configs.
 app.autodiscover_tasks()
@@ -19,5 +27,16 @@ app.autodiscover_tasks()
 
 @app.task(bind=True)
 def debug_task(self):
-    print('Request: {0!r}'.format(self.request))
-    return 'Hello'
+  print('Request: {0!r}'.format(self.request))
+  return 'Hello'
+
+#
+if TASK_SERVER.BEAT_ENABLED.get():
+  app.conf.beat_schedule = {
+    'add-every-monday-morning': {
+      'task': 'desktop.celery.debug_task',
+      'schedule': crontab(minute='*'),
+      # 'schedule': crontab(hour=7, minute=30, day_of_week=1),
+      #'args': (16, 16),
+    },
+  }

+ 15 - 0
desktop/core/src/desktop/lib/scheduler/__init__.py

@@ -0,0 +1,15 @@
+# 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.

+ 34 - 0
desktop/core/src/desktop/lib/scheduler/api.py

@@ -0,0 +1,34 @@
+#!/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 desktop.auth.backend import is_admin
+from desktop.lib.django_util import JsonResponse
+
+
+LOG = logging.getLogger(__name__)
+
+
+def get_schedule(request):
+  return JsonResponse({
+  })
+
+
+def submit_schedule(request):
+  return JsonResponse({
+  })

+ 15 - 0
desktop/core/src/desktop/lib/scheduler/lib/__init__.py

@@ -0,0 +1,15 @@
+# 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.

+ 24 - 0
desktop/core/src/desktop/lib/scheduler/lib/base.py

@@ -0,0 +1,24 @@
+
+
+from desktop.lib.scheduler.lib.beat import CeleryBeatApi
+from desktop.lib.scheduler.lib.oozie import OozieApi
+
+
+def get_api(request, interface):
+
+  if interface == 'beat':
+    return CeleryBeatApi(user=request.user)
+  elif interface == 'oozie':
+    return OozieApi(user=request.user)
+  else:
+    raise PopupException(_('Scheduler connector interface not recognized: %s') % interface)
+
+
+class Api():
+
+  def get_schedule():
+    return JsonResponse({
+    })
+
+  def submit_schedule():
+    return

+ 21 - 0
desktop/core/src/desktop/lib/scheduler/lib/beat.py

@@ -0,0 +1,21 @@
+
+from desktop.lib.scheduler.lib.base import Api
+
+
+class CeleryBeatApi(Api):
+
+  def __init__(self, user=None):
+    pass
+
+
+from celery.schedules import crontab
+from desktop.celery import app
+
+app.conf.beat_schedule = {
+  'add-every-monday-morning': {
+    'task': 'desktop.celery.debug_task',
+    'schedule': crontab(minute='*/15'),
+    # 'schedule': crontab(hour=7, minute=30, day_of_week=1),
+    #'args': (16, 16),
+  },
+}

+ 8 - 0
desktop/core/src/desktop/lib/scheduler/lib/oozie.py

@@ -0,0 +1,8 @@
+
+from desktop.lib.scheduler.lib.base import Api
+
+
+class OozieApi(Api):
+
+  def __init__(self, user=None):
+    pass

+ 16 - 0
desktop/core/src/desktop/lib/scheduler/models.py

@@ -0,0 +1,16 @@
+#!/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.

+ 29 - 0
desktop/core/src/desktop/lib/scheduler/urls.py

@@ -0,0 +1,29 @@
+#!/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.
+
+from django.conf.urls import url
+
+from desktop.lib.scheduler import api
+
+
+urlpatterns = [
+  url(r'^api/schedule/submit?$', api.submit_schedule(), name='analytics.api.admin_stats'),
+  url(r'^api/schedule/?$', api.get_schedule, name='analytics.api.admin_stats'),
+]
+
+# /oozie/editor/coordinator
+# /oozie/editor/coordinator/submit/

+ 3 - 0
desktop/libs/notebook/src/notebook/connectors/base.py

@@ -580,10 +580,12 @@ class ExecutionWrapper():
           self.callback.on_execute(handle)
         self.should_close = True
         self._until_available()
+
     if self.snippet['result']['handle'].get('sync', False):
       result = self.snippet['result']['handle']['result']
     else:
       result = self.api.fetch_result(self.notebook, self.snippet, rows, start_over)
+
     return ResultWrapper(result.get('meta'), result.get('data'), result.get('has_more'))
 
   def _until_available(self):
@@ -593,6 +595,7 @@ class ExecutionWrapper():
     sleep_seconds = 1
     check_status_count = 0
     get_log_is_full_log = self.api.get_log_is_full_log(self.notebook, self.snippet)
+
     while True:
       response = self.api.check_status(self.notebook, self.snippet)
       if self.callback and hasattr(self.callback, 'on_status'):

+ 16 - 2
desktop/libs/notebook/src/notebook/tasks.py

@@ -31,6 +31,7 @@ from django.contrib.auth.models import User
 from django.db import transaction
 from django.http import FileResponse, HttpRequest
 
+from beeswax import data_export
 from desktop.auth.backend import rewrite_user
 from desktop.celery import app
 from desktop.conf import TASK_SERVER
@@ -41,6 +42,7 @@ from desktop.settings import CACHES_CELERY_KEY
 from notebook.connectors.base import get_api, QueryExpired, ExecutionWrapper
 from notebook.sql_utils import get_current_statement
 
+
 LOG_TASK = get_task_logger(__name__)
 LOG = logging.getLogger(__name__)
 STATE_MAP = {
@@ -60,6 +62,7 @@ STATE_MAP = {
 storage_info = json.loads(TASK_SERVER.RESULT_STORAGE.get())
 storage = get_storage_class(storage_info.get('backend'))(**storage_info.get('properties', {}))
 
+
 class ExecutionWrapperCallback(object):
   def __init__(self, uuid, meta, f_log):
     self.meta = meta
@@ -88,7 +91,6 @@ class ExecutionWrapperCallback(object):
 #TODO: UI should be able to close a query that is available, but not expired
 @app.task()
 def download_to_file(notebook, snippet, file_format='csv', max_rows=-1, **kwargs):
-  from beeswax import data_export
   download_to_file.update_state(task_id=notebook['uuid'], state='STARTED', meta={})
   request = _get_request(**kwargs)
   api = get_api(request, snippet)
@@ -119,6 +121,18 @@ def close_statement_async(notebook, snippet, **kwargs):
   request = _get_request(**kwargs)
   get_api(request, snippet).close_statement(notebook, snippet)
 
+
+@app.task(ignore_result=True)
+def batch_execute_query(doc_id, user):
+  # get SQL
+  # Add INSERT INTO table
+  # Add variables?
+  # execute query
+  # return when done. send email notification. get taskid.
+
+  # see in Flower API for listing runs?
+
+
 #TODO: Convert csv to excel if needed
 def download(*args, **kwargs):
   notebook = args[0]
@@ -368,4 +382,4 @@ def _get_request(postdict=None, user_id=None):
   user = User.objects.get(id=user_id)
   user = rewrite_user(user)
   request.user = user
-  return request
+  return request