Explorar el Código

HUE-2987 [oozie] Fix sync workflow to update Coordinator definition

krish hace 10 años
padre
commit
12b7c1d

+ 4 - 0
apps/oozie/src/oozie/models2.py

@@ -1980,6 +1980,10 @@ class Coordinator(Job):
     tmpl = "editor2/gen/coordinator.xml.mako"
     return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
 
+  def clear_workflow_params(self):
+    # Repopulated in the config properties
+    self.data['variables'] = [dataset for dataset in self.data['variables'] if dataset['dataset_type'] != 'parameter']
+
   @property
   def properties(self):
     props = [{'name': dataset['workflow_variable'], 'value': dataset['dataset_variable']} for dataset in self.data['variables'] if dataset['dataset_type'] == 'parameter']

+ 22 - 11
apps/oozie/src/oozie/views/dashboard.py

@@ -648,31 +648,42 @@ def sync_coord_workflow(request, job_id):
   hue_coord = get_history().get_coordinator_from_config(job.conf_dict)
   hue_wf = (hue_coord and hue_coord.workflow) or get_history().get_workflow_from_config(job.conf_dict)
   wf_application_path = job.conf_dict.get('wf_application_path') and Hdfs.urlsplit(job.conf_dict['wf_application_path'])[2] or ''
+  coord_application_path = job.conf_dict.get('oozie.coord.application.path') and Hdfs.urlsplit(job.conf_dict['oozie.coord.application.path'])[2] or ''
+  properties = hue_coord and hue_coord.properties and dict([(param['name'], param['value']) for param in hue_coord.properties]) or None
 
   if request.method == 'POST':
     params_form = ParametersFormSet(request.POST)
     if params_form.is_valid():
       mapping = dict([(param['name'], param['value']) for param in params_form.cleaned_data])
 
-      submission = Submission(user=request.user, job=hue_wf, fs=request.fs, jt=request.jt, properties=mapping)
-      submission._sync_definition(wf_application_path, mapping)
+      # Update workflow params in coordinator
+      hue_coord.clear_workflow_params()
+      properties = dict([(param['name'], param['value']) for param in hue_coord.properties])
+
+      # Deploy WF XML
+      submission = Submission(user=request.user, job=hue_wf, fs=request.fs, jt=request.jt, properties=properties)
+      submission._create_file(wf_application_path, hue_wf.XML_FILE_NAME, hue_wf.to_xml(mapping=properties), do_as=True)
+
+      # Deploy Coordinator XML
+      job.conf_dict.update(mapping)
+      submission = Submission(user=request.user, job=hue_coord, fs=request.fs, jt=request.jt, properties=job.conf_dict, oozie_id=job.id)
+      submission._create_file(coord_application_path, hue_coord.XML_FILE_NAME, hue_coord.to_xml(mapping=job.conf_dict), do_as=True)
+      # Server picks up deployed Coordinator XML changes after running 'update' action
+      submission.update_coord()
 
       request.info(_('Successfully updated Workflow definition'))
       return redirect(reverse('oozie:list_oozie_coordinator', kwargs={'job_id': job_id}))
     else:
       request.error(_('Invalid submission form: %s' % params_form.errors))
   else:
-    parameters = hue_wf and hue_wf.find_all_parameters() or []
-    params_dict = dict([(param['name'], param['value']) for param in parameters])
-
-    submission = Submission(user=request.user, job=hue_wf, fs=request.fs, jt=request.jt, properties=None)
-    prev_properties = hue_wf and hue_wf.deployment_dir and \
-                      submission.get_external_parameters(request.fs.join(wf_application_path, hue_wf.XML_FILE_NAME)) or {}
+    new_params = hue_wf and hue_wf.find_all_parameters() or []
+    new_params = dict([(param['name'], param['value']) for param in new_params])
 
-    for key, value in params_dict.iteritems():
-      params_dict[key] = prev_properties[key] if key in prev_properties.keys() else params_dict[key]
+    # Set previous values
+    if properties:
+      new_params = dict([(key, properties[key]) if key in properties.keys() else (key, new_params[key]) for key, value in new_params.iteritems()])
 
-    initial_params = ParameterForm.get_initial_params(params_dict)
+    initial_params = ParameterForm.get_initial_params(new_params)
     params_form = ParametersFormSet(initial=initial_params)
 
   popup = render('editor2/submit_job_popup.mako', request, {

+ 1 - 1
desktop/libs/liboozie/src/liboozie/oozie_api.py

@@ -217,7 +217,7 @@ class OozieApi(object):
     job_control(jobid, action) -> None
     Raise RestException on error.
     """
-    if action not in ('start', 'suspend', 'resume', 'kill', 'rerun', 'coord-rerun', 'bundle-rerun', 'change', 'ignore'):
+    if action not in ('start', 'suspend', 'resume', 'kill', 'rerun', 'coord-rerun', 'bundle-rerun', 'change', 'ignore', 'update'):
       msg = 'Invalid oozie job action: %s' % (action,)
       LOG.error(msg)
       raise ValueError(msg)

+ 6 - 9
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -134,6 +134,12 @@ class Submission(object):
 
     return self.oozie_id
 
+  def update_coord(self):
+    self.api = get_oozie(self.user, api_version="v2")
+    self.api.job_control(self.oozie_id, action='update', properties=self.properties, parameters=None)
+    LOG.info("Update: %s" % (self,))
+
+    return self.oozie_id
 
   def rerun_bundle(self, deployment_dir, params):
     jt_address = cluster.get_cluster_addr_for_job_submission()
@@ -341,15 +347,6 @@ class Submission(object):
      self.fs.create(file_path, overwrite=True, permission=0644, data=smart_str(data))
    LOG.debug("Created/Updated %s" % (file_path,))
 
-  def _sync_definition(self, deployment_dir, mapping):
-    """ This is helper function for 'Sync Workflow' functionality in a Coordinator.
-      It copies updated workflow changes into HDFS """
-
-    self._create_file(deployment_dir, self.job.XML_FILE_NAME, self.job.to_xml(mapping=mapping), do_as=True)
-
-    data_properties = smart_str('\n'.join(['%s=%s' % (key, val) for key, val in mapping.iteritems()]))
-    self._create_file(deployment_dir, 'job.properties', data_properties, do_as=True)
-
 def create_directories(fs, directory_list=[]):
   # If needed, create the remote home, deployment and data directories
   directories = [REMOTE_DEPLOYMENT_DIR.get()] + directory_list