Эх сурвалжийг харах

[oozie] Add resubmit management command to workflow and coordinator dashboards

abec 13 жил өмнө
parent
commit
85d9c91

+ 0 - 1
apps/oozie/src/oozie/models.py

@@ -1046,7 +1046,6 @@ class History(models.Model):
 
     return history
 
-
 def find_parameters(instance, fields=None):
   """Find parameters in the given fields"""
   if fields is None:

+ 6 - 0
apps/oozie/src/oozie/templates/dashboard/list_oozie_coordinator.mako

@@ -76,11 +76,17 @@ ${ layout.menubar(section='dashboard') }
           <tr>
             <td>${ _('Manage') }</td>
             <td>
+            <form action="${ url('oozie:resubmit_coordinator', job_id=oozie_coordinator.id) }" method="post">
             % if oozie_coordinator.is_running():
               <button type="button" class="btn manage-oozie-job-btn" data-url="${ url('oozie:manage_oozie_jobs', job_id=oozie_coordinator.id, action='kill') }"  data-message="The coordinator was killed!">
                 ${ _('Kill') }
               </button>
+            % else:
+              <button type="submit" class="btn">
+                ${ _('Resubmit') }
+              </button>
             % endif
+            </form>
             </td>
           </tr>
         </tbody>

+ 7 - 1
apps/oozie/src/oozie/templates/dashboard/list_oozie_workflow.mako

@@ -112,11 +112,17 @@ ${ layout.menubar(section='dashboard') }
       ${ _('Manage') }
     </div>
     <div class="span3">
+      <form action="${ url('oozie:resubmit_workflow', job_id=oozie_workflow.id) }" method="post">
       % if oozie_workflow.is_running():
-        <button class="btn manage-oozie-job-btn" data-url="${ url('oozie:manage_oozie_jobs', job_id=oozie_workflow.id, action='kill') }" data-message="The workflow was killed!">
+        <button type="button" class="btn manage-oozie-job-btn" data-url="${ url('oozie:manage_oozie_jobs', job_id=oozie_workflow.id, action='kill') }" data-message="The workflow was killed!">
           ${ _('Kill') }
         </button>
+      % else:
+        <button type="submit" class="btn">
+          ${ _('Resubmit') }
+        </button>
       % endif
+      </form>
     </div>
   </div>
 

+ 70 - 6
apps/oozie/src/oozie/tests.py

@@ -31,25 +31,26 @@ from liboozie import oozie_api
 from liboozie.types import WorkflowList, Workflow as OozieWorkflow, Coordinator as OozieCoordinator,\
   CoordinatorList, WorkflowAction
 
-from oozie.models import Workflow, Node, Job, Coordinator, Fork
+from oozie.models import Workflow, Node, Job, Coordinator, Fork, History
 from oozie.conf import SHARE_JOBS, REMOTE_DEPLOYMENT_DIR
 
 LOG = logging.getLogger(__name__)
 
 
-
 class MockOozieApi:
   JSON_WORKFLOW_LIST = [{u'status': u'RUNNING', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:35:48 GMT', u'appName': u'WordCount1', u'lastModTime': u'Mon, 30 Jul 2012 22:37:00 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': None, u'consoleUrl': u'http://runreal:11000/oozie?job=0000012-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:35:48 GMT', u'toString': u'Workflow id[0000012-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:37:00 GMT', u'id': u'0000012-120725142744176-oozie-oozi-W', u'group': None, u'user': u'romain'},
                         {u'status': u'KILLED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:31:08 GMT', u'appName': u'WordCount2', u'lastModTime': u'Mon, 30 Jul 2012 22:32:20 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': None, u'consoleUrl': u'http://runreal:11000/oozie?job=0000011-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:31:08 GMT', u'toString': u'Workflow id[0000011-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:32:20 GMT', u'id': u'0000011-120725142744176-oozie-oozi-W', u'group': None, u'user': u'romain'},
                         {u'status': u'SUCCEEDED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:20:48 GMT', u'appName': u'WordCount3', u'lastModTime': u'Mon, 30 Jul 2012 22:22:00 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': None, u'consoleUrl': u'http://runreal:11000/oozie?job=0000009-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:20:48 GMT', u'toString': u'Workflow id[0000009-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:22:00 GMT', u'id': u'0000009-120725142744176-oozie-oozi-W', u'group': None, u'user': u'romain'},
                         {u'status': u'SUCCEEDED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:16:58 GMT', u'appName': u'WordCount4', u'lastModTime': u'Mon, 30 Jul 2012 22:18:10 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': None, u'consoleUrl': u'http://runreal:11000/oozie?job=0000008-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:16:58 GMT', u'toString': u'Workflow id[0000008-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:18:10 GMT', u'id': u'0000008-120725142744176-oozie-oozi-W', u'group': None, u'user': u'romain'}]
   WORKFLOW_IDS = [wf['id'] for wf in JSON_WORKFLOW_LIST]
+  WORKFLOW_DICT = dict([(wf['id'], wf) for wf in JSON_WORKFLOW_LIST])
 
   JSON_COORDINATOR_LIST = [{u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000041-120717205528122-oozie-oozi-C] status[RUNNING]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'RUNNING', u'conf': None, u'user': u'romain', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/romain/demo2', u'timeUnit': u'DAY', u'coordJobId': u'0000041-120717205528122-oozie-oozi-C', u'coordJobName': u'DailyWordCount1', u'nextMaterializedTime': u'Wed, 04 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Wed, 04 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 00:00:00 GMT'},
                            {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000011-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'romain', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000011-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount2', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'},
                            {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000010-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'romain', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000010-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount3', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'},
                            {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000009-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'romain', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000009-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount4', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'}]
   COORDINATOR_IDS = [coord['coordJobId'] for coord in JSON_COORDINATOR_LIST]
+  COORDINATOR_DICT = dict([(coord['coordJobId'], coord) for coord in JSON_COORDINATOR_LIST])
 
   WORKFLOW_ACTION = {u'status': u'OK', u'retries': 0, u'transition': u'end', u'stats': None, u'startTime': u'Fri, 10 Aug 2012 05:24:21 GMT', u'toString': u'Action name[WordCount] status[OK]', u'cred': u'null', u'errorMessage': None, u'errorCode': None, u'consoleUrl': u'http://localhost:50030/jobdetails.jsp?jobid=job_201208072118_0044', u'externalId': u'job_201208072118_0044', u'externalStatus': u'SUCCEEDED', u'conf': u'<map-reduce xmlns="uri:oozie:workflow:0.2">\r\n  <job-tracker>localhost:8021</job-tracker>\r\n  <name-node>hdfs://localhost:8020</name-node>\r\n  <configuration>\r\n    <property>\r\n      <name>mapred.mapper.regex</name>\r\n      <value>dream</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.input.dir</name>\r\n      <value>/user/romain/words/20120702</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.output.dir</name>\r\n      <value>/user/romain/out/rrwords/20120702</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.mapper.class</name>\r\n      <value>org.apache.hadoop.mapred.lib.RegexMapper</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.combiner.class</name>\r\n      <value>org.apache.hadoop.mapred.lib.LongSumReducer</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.reducer.class</name>\r\n      <value>org.apache.hadoop.mapred.lib.LongSumReducer</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.output.key.class</name>\r\n      <value>org.apache.hadoop.io.Text</value>\r\n    </property>\r\n    <property>\r\n      <name>mapred.output.value.class</name>\r\n      <value>org.apache.hadoop.io.LongWritable</value>\r\n    </property>\r\n  </configuration>\r\n</map-reduce>', u'type': u'map-reduce', u'trackerUri': u'localhost:8021', u'externalChildIDs': None, u'endTime': u'Fri, 10 Aug 2012 05:24:38 GMT', u'data': None, u'id': u'0000021-120807211836060-oozie-oozi-W@WordCount', u'name': u'WordCount'}
 
@@ -60,7 +61,7 @@ class MockOozieApi:
     pass
 
   def submit_job(self, properties):
-    return 'ONE-OOZIE-ID'
+    return 'ONE-OOZIE-ID-W'
 
   def get_workflows(self, **kwargs):
     return WorkflowList(self, {'offset': 0, 'total': 4, 'workflows': MockOozieApi.JSON_WORKFLOW_LIST})
@@ -69,10 +70,16 @@ class MockOozieApi:
     return CoordinatorList(self, {'offset': 0, 'total': 5, 'coordinatorjobs': MockOozieApi.JSON_COORDINATOR_LIST})
 
   def get_job(self, job_id):
-    return OozieWorkflow(self, MockOozieApi.JSON_WORKFLOW_LIST[0])
+    if job_id in MockOozieApi.WORKFLOW_DICT:
+      return OozieWorkflow(self, MockOozieApi.WORKFLOW_DICT[job_id])
+    else:
+      return OozieWorkflow(self, {'id': job_id, 'actions': []})
 
   def get_coordinator(self, job_id):
-    return OozieCoordinator(self, MockOozieApi.JSON_COORDINATOR_LIST[0])
+    if job_id in MockOozieApi.COORDINATOR_DICT:
+      return OozieCoordinator(self, MockOozieApi.COORDINATOR_DICT[job_id])
+    else:
+      return OozieCoordinator(self, {'id': job_id, 'actions': []})
 
   def get_action(self, action_id):
     return WorkflowAction(MockOozieApi.WORKFLOW_ACTION)
@@ -87,11 +94,14 @@ class MockOozieApi:
     return '<xml></xml>'
 
 
+class MockOozieCoordinatorApi(MockOozieApi):
+  def submit_job(self, properties):
+    return 'ONE-OOZIE-ID-C'
+
 # Monkey patch Lib Oozie with Mock API
 oozie_api.OozieApi = MockOozieApi
 
 
-
 class TestEditor:
 
   def setUp(self):
@@ -315,6 +325,10 @@ class TestEditor:
 
 
   def test_workflow_permissions(self):
+    # Monkey patch Lib Oozie with Mock API
+    oozie_api.OozieApi = MockOozieApi
+    oozie_api._api_cache = None
+
     response = self.c.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
     assert_true('Editor' in response.content, response.content)
     assert_true('Workflow wf-name-1' in response.content, response.content)
@@ -406,6 +420,26 @@ class TestEditor:
       finish()
       finish_deployement()
 
+    # Resubmit
+    finish = SHARE_JOBS.set_for_testing(False)
+    finish_deployement = REMOTE_DEPLOYMENT_DIR.set_for_testing('/tmp')
+    try:
+      job_id = History.objects.get(job=self.wf).oozie_job_id
+      response = client_not_me.post(reverse('oozie:resubmit_workflow', args=[job_id]))
+      assert_true('Permission denied' in response.content, response.content)
+    finally:
+      finish()
+      finish_deployement()
+    finish = SHARE_JOBS.set_for_testing(True)
+    finish_deployement = REMOTE_DEPLOYMENT_DIR.set_for_testing('/tmp')
+    try:
+      job_id = History.objects.get(job=self.wf).oozie_job_id
+      response = client_not_me.post(reverse('oozie:resubmit_workflow', args=[job_id]))
+      assert_false('Permission denied' in response.content, response.content)
+    finally:
+      finish()
+      finish_deployement()
+
     # Delete
     finish = SHARE_JOBS.set_for_testing(False)
     try:
@@ -512,6 +546,9 @@ class TestEditor:
 
 
   def test_coordinator_permissions(self):
+    oozie_api.OozieApi = MockOozieCoordinatorApi
+    oozie_api._api_cache = None
+
     coord = create_coordinator(self.wf)
 
     response = self.c.get(reverse('oozie:edit_coordinator', args=[coord.id]))
@@ -615,6 +652,27 @@ class TestEditor:
       finish()
       finish_deployement()
 
+    # Resubmit
+    # Monkey patch Lib Oozie with Mock API
+    finish = SHARE_JOBS.set_for_testing(False)
+    finish_deployement = REMOTE_DEPLOYMENT_DIR.set_for_testing('/tmp')
+    try:
+      oozie_job_id = History.objects.get(job=coord).oozie_job_id
+      response = client_not_me.post(reverse('oozie:resubmit_coordinator', args=[oozie_job_id]))
+      assert_true('Permission denied' in response.content, response.content)
+    finally:
+      finish()
+      finish_deployement()
+    finish = SHARE_JOBS.set_for_testing(True)
+    finish_deployement = REMOTE_DEPLOYMENT_DIR.set_for_testing('/tmp')
+    try:
+      oozie_job_id = History.objects.get(job=coord).oozie_job_id
+      response = client_not_me.post(reverse('oozie:resubmit_coordinator', args=[oozie_job_id]))
+      assert_false('Permission denied' in response.content, response.content)
+    finally:
+      finish()
+      finish_deployement()
+
     # Delete
     finish = SHARE_JOBS.set_for_testing(False)
     try:
@@ -768,18 +826,24 @@ class TestDashboard:
     # Kill button in response
     response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]), {}, follow=True)
     assert_true(('%s/kill' % MockOozieApi.WORKFLOW_IDS[0]) in response.content, response.content)
+    assert_false('Resubmit' in response.content, response.content)
 
+    # Resubmit button in response
     response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[1]]), {}, follow=True)
     assert_false(('%s/kill' % MockOozieApi.WORKFLOW_IDS[1]) in response.content, response.content)
+    assert_true('Resubmit' in response.content, response.content)
 
 
   def test_manage_coordinator_dashboard(self):
     # Kill button in response
     response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]), {}, follow=True)
     assert_true(('%s/kill' % MockOozieApi.COORDINATOR_IDS[0]) in response.content, response.content)
+    assert_false('Resubmit' in response.content, response.content)
 
+    # Resubmit button in response
     response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[1]]), {}, follow=True)
     assert_false(('%s/kill' % MockOozieApi.COORDINATOR_IDS[1]) in response.content, response.content)
+    assert_true('Resubmit' in response.content, response.content)
 
 
   def test_list_workflows(self):

+ 2 - 0
apps/oozie/src/oozie/urls.py

@@ -30,6 +30,7 @@ urlpatterns = patterns(
   url(r'^delete_workflow/(?P<workflow>\d+)$', 'delete_workflow', name='delete_workflow'),
   url(r'^clone_workflow/(?P<workflow>\d+)$', 'clone_workflow', name='clone_workflow'),
   url(r'^submit_workflow/(?P<workflow>\d+)$', 'submit_workflow', name='submit_workflow'),
+  url(r'^resubmit_workflow/(?P<job_id>[-\w]+)$', 'resubmit_workflow', name='resubmit_workflow'),
 
   url(r'^new_action/(?P<workflow>\d+)/(?P<node_type>\w+)/(?P<parent_action_id>\d+)$', 'new_action', name='new_action'),
   url(r'^edit_action/(?P<action>\d+)$', 'edit_action', name='edit_action'),
@@ -47,6 +48,7 @@ urlpatterns = patterns(
   url(r'^create_coordinator_dataset/(?P<coordinator>[-\w]+)$', 'create_coordinator_dataset', name='create_coordinator_dataset'),
   url(r'^create_coordinator_data/(?P<coordinator>[-\w]+)/(?P<data_type>(input|output))$', 'create_coordinator_data', name='create_coordinator_data'),
   url(r'^submit_coordinator/(?P<coordinator>\d+)$', 'submit_coordinator', name='submit_coordinator'),
+  url(r'^resubmit_coordinator/(?P<job_id>[-\w]+)$', 'resubmit_coordinator', name='resubmit_coordinator'),
 
   url(r'^workflow_parameters/(?P<workflow>\d+)$', 'get_workflow_parameters', name='workflow_parameters'),
   url(r'^list_history$', 'list_history', name='list_history'),

+ 68 - 15
apps/oozie/src/oozie/views/editor.py

@@ -290,18 +290,43 @@ def submit_workflow(request, workflow):
 
   try:
     mapping = dict(request.POST.iteritems())
-    submission = Submission(request.user, workflow, request.fs, mapping)
-    job_id = submission.run()
+    job_id = _submit_workflow(request, workflow, mapping)
   except RestException, ex:
     raise PopupException(_("Error submitting workflow %s") % (workflow,),
                          detail=ex._headers.get('oozie-error-message', ex))
 
-  History.objects.create_from_submission(submission)
   request.info(_('Workflow submitted'))
 
   return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_id}))
 
 
+def _submit_workflow(request, workflow, mapping):
+  submission = Submission(request.user, workflow, request.fs, mapping)
+  job_id = submission.run()
+  History.objects.create_from_submission(submission)
+  return job_id
+
+
+def resubmit_workflow(request, job_id):
+  if request.method != 'POST':
+    raise PopupException(_('A POST request is required.'))
+
+  history = History.objects.get(oozie_job_id=job_id)
+
+  can_access_job_or_exception(request, history.job.id)
+
+  try:
+    workflow = history.get_workflow().get_full_node()
+    properties = history.properties_dict
+    job_id = _submit_workflow(request, workflow, properties)
+  except RestException, ex:
+    raise PopupException(_("Error submitting workflow %s") % (workflow,),
+                         detail=ex._headers.get('oozie-error-message', ex))
+
+  request.info(_('Workflow submitted'))
+
+  return redirect(reverse('oozie:list_oozie_workflow', kwargs={'job_id': job_id}))
+
 @check_job_access_permission
 def get_workflow_parameters(request, workflow):
   """
@@ -626,23 +651,51 @@ def submit_coordinator(request, coordinator):
     raise PopupException(_('A POST request is required.'))
 
   try:
-    if not coordinator.workflow.is_deployed(request.fs):
-      submission = Submission(request.user, coordinator.workflow, request.fs, request.POST)
-      wf_dir = submission.deploy()
-      coordinator.workflow.deployment_dir = wf_dir
-      coordinator.workflow.save()
-
-    coordinator.deployment_dir = coordinator.workflow.deployment_dir
-    properties = {'wf_application_path': coordinator.workflow.deployment_dir}
-    properties.update(dict(request.POST.iteritems()))
-
-    submission = Submission(request.user, coordinator, request.fs, properties=properties)
-    job_id = submission.run()
+    job_id = _submit_coordinator(request, coordinator, request.POST)
   except RestException, ex:
     raise PopupException(_("Error submitting coordinator %s") % (coordinator,),
                          detail=ex._headers.get('oozie-error-message', ex))
 
+  request.info(_('Coordinator submitted'))
+
+  return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
+
+
+def _submit_coordinator(request, coordinator, mapping):
+  if not coordinator.workflow.is_deployed(request.fs):
+    submission = Submission(request.user, coordinator.workflow, request.fs, mapping)
+    wf_dir = submission.deploy()
+    coordinator.workflow.deployment_dir = wf_dir
+    coordinator.workflow.save()
+
+  coordinator.deployment_dir = coordinator.workflow.deployment_dir
+  properties = {'wf_application_path': coordinator.workflow.deployment_dir}
+  properties.update(dict(request.POST.iteritems()))
+
+  submission = Submission(request.user, coordinator, request.fs, properties=properties)
+  job_id = submission.run()
+
   History.objects.create_from_submission(submission)
+
+  return job_id
+
+
+def resubmit_coordinator(request, job_id):
+  if request.method != 'POST':
+    raise PopupException(_('A POST request is required.'))
+
+  history = History.objects.get(oozie_job_id=job_id)
+
+  can_access_job_or_exception(request, history.job.id)
+
+  try:
+    coordinator = history.get_coordinator().get_full_node()
+    properties = history.properties_dict
+    job_id = _submit_coordinator(request, coordinator, properties)
+  except RestException, ex:
+    raise PopupException(_("Error submitting coordinator %s") % (coordinator,),
+                         detail=ex._headers.get('oozie-error-message', ex))
+
   request.info(_('Coordinator submitted'))
 
   return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))

+ 9 - 8
desktop/libs/liboozie/src/liboozie/oozie_api.py

@@ -188,12 +188,13 @@ class OozieApi(object):
       raise ValueError(msg)
     params = self._get_params()
     params['action'] = action
-    return self._root.put('job/%s' % (jobid,), params)
 
+    return self._root.put('job/%s' % jobid, params)
 
-  def submit_workflow(self, application_path, properties=None):
+
+  def submit_workflow(self, application_path):
     """
-    submit_workflow(application_path, properties=None) -> jobid
+    submit_workflow(application_path) -> jobid
 
     Submit a job to Oozie. May raise PopupException.
     """
@@ -210,7 +211,7 @@ class OozieApi(object):
 
   def submit_job(self, properties=None):
     """
-    submit_workflow(properties=None) -> jobid
+    submit_workflow(properties=None, id=None) -> jobid
 
     Submit a job to Oozie. May raise PopupException.
     """
@@ -224,12 +225,12 @@ class OozieApi(object):
     properties = defaults
 
     params = self._get_params()
-    resp = self._root.post('jobs',
-                           params, data=config_gen(properties),
-                           contenttype=_XML_CONTENT_TYPE)
-
+    resp = self._root.post('jobs', params,
+                  data=config_gen(properties),
+                  contenttype=_XML_CONTENT_TYPE)
     return resp['id']
 
+
   def get_build_version(self):
     """
     get_build_version() -> Build version (dictionary)