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

HUE-3797 [scheduler] Display periodic task as Scheduled Tasks in job browser

Romain преди 6 години
родител
ревизия
075c96012b
променени са 3 файла, в които са добавени 47 реда и са изтрити 23 реда
  1. 3 3
      apps/oozie/src/oozie/models2.py
  2. 2 1
      desktop/core/src/desktop/lib/scheduler/api.py
  3. 42 19
      desktop/core/src/desktop/lib/scheduler/lib/beat.py

+ 3 - 3
apps/oozie/src/oozie/models2.py

@@ -964,7 +964,7 @@ class Node(object):
 #             'archives': [],
 #             'capture_output': True
 #             #       <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
-# 
+#
 #             #  Node(dict(AltusAction().get_fields()))
 #           }
 #         }
@@ -3514,12 +3514,12 @@ class Coordinator(Job):
     if self.data['properties']['document']:
       document = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['document'])
       wf_doc = WorkflowBuilder().create_workflow(document=document, user=self.document.owner, managed=True)
-      wf = Workflow(data=wf_doc.data,user=self.document.owner)
+      wf = Workflow(data=wf_doc.data, user=self.document.owner)
       wf_doc.delete()
       return wf
     else:
       wf_doc = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['workflow'])
-      return Workflow(document=wf_doc,user=self.document.owner)
+      return Workflow(document=wf_doc, user=self.document.owner)
 
   def get_absolute_url(self):
     return reverse('oozie:edit_coordinator') + '?coordinator=%s' % self.id

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

@@ -48,6 +48,7 @@ def get_schedule(request):
 #@check_document_access_permission()
 def submit_schedule(request, doc_id):
   interface = request.GET.get('interface', request.POST.get('interface', 'beat'))
+
   if doc_id.isdigit():
     coordinator = Coordinator(document=Document2.objects.get(id=doc_id))
   else:
@@ -75,7 +76,7 @@ def submit_schedule(request, doc_id):
     else:
       request.error(_('Invalid submission form: %s') % params_form.errors)
   else:
-    parameters = coordinator.find_all_parameters()
+    parameters = coordinator.find_all_parameters() if interface == 'oozie' else []
     initial_params = ParameterForm.get_initial_params(dict([(param['name'], param['value']) for param in parameters]))
     params_form = ParametersFormSet(initial=initial_params)
 

+ 42 - 19
desktop/core/src/desktop/lib/scheduler/lib/beat.py

@@ -38,6 +38,16 @@ class CeleryBeatApi(Api):
     # Assumes SQL queries currently
     document = Document2.objects.get(uuid=coordinator.get_data_for_json()['properties']['document'])
 
+    schedule_properties = {
+        'name': 'Scheduled document %(user)s %(uuid)s' % {
+        'user': request.user.username,
+        'uuid': document.uuid
+        },
+        'description': request.user.username, # Owner
+        'task': 'notebook.tasks.run_sync_query',
+        'defaults': {"args": json.dumps([document.uuid, request.user.username])},
+    }
+
     if is_cron:
       schedule, created = CrontabSchedule.objects.get_or_create(
           minute='*',
@@ -46,34 +56,47 @@ class CeleryBeatApi(Api):
           day_of_month='*',
           month_of_year='*'
       )
-
-      task = PeriodicTask.objects.update_or_create(
-        crontab=schedule,
-        name='Scheduled document %(user)s %(uuid)s' % {
-          'user': request.user.username,
-          'uuid': document.uuid
-        },
-        description=request.user.username, # Owner
-        task='notebook.tasks.run_sync_query',
-        defaults={"args": json.dumps([document.uuid, request.user.username])},
-      )
-      task.enabled=True
-      task.save()
+      schedule_properties['crontab'] = schedule
     else:
       schedule, created = IntervalSchedule.objects.get_or_create(
         every=15,
         period=IntervalSchedule.SECONDS,
       )
+      schedule_properties['interval'] = schedule
 
-      task, created = PeriodicTask.objects.update_or_create(
-        interval=schedule,
-        name='Scheduled query',
-        task='notebook.tasks.run_sync_query',
-      )
+    task = PeriodicTask.objects.update_or_create(**schedule_properties)
+    task.enabled=True
+    task.save()
+
+    return task
 
 
   def list_tasks(self, user):
-    PeriodicTask.objects.filter(description=user.username)
+    return [{
+        'id': task.id,
+        'name': task.name,
+        'description': task.description,
+        'task_name': task.name,
+        'task_id': task.id,
+        'args': task.args,
+        'kwargs': task.kwargs,
+        'queue': task.queue,
+        'exchange': task.exchange,
+        'routing_key': task.routing_key,
+        'priority': task.priority,
+        'expires': task.expires,
+        'one_off': task.one_off,
+        'start_time': task.start_time,
+        'enabled': task.enabled,
+        'last_run_at': task.last_run_at,
+        'total_run_count': task.total_run_count,
+        'date_changed': task.date_changed,
+        'interval_name': task.interval,
+        'crontab': task.crontab,
+        'solar': task.solar
+      }
+      for task in PeriodicTask.objects.filter(description=user.username)
+    ]
 
 
   def action(self, schedule_id, schedule_ids=None, action='pause'):