Browse Source

[oozie] Support for sub workflows

Romain Rigaux 11 years ago
parent
commit
4c03402

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

@@ -153,13 +153,17 @@ class Workflow():
 
 
     data = self.get_data()
     data = self.get_data()
     nodes = [Node(node) for node in data['workflow']['nodes']]
     nodes = [Node(node) for node in data['workflow']['nodes']]
-    node_mapping = dict([(node.id, node.name) for node in nodes])
+    node_mapping = dict([(node.id, node) for node in nodes])
+    
+    sub_wfs_ids = [node.data['properties']['workflow'] for node in nodes if node.data['type'] == 'subworkflow']
+    workflow_mapping = dict([(workflow.uuid, Workflow(document=workflow)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)])
 
 
     xml = re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
     xml = re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
               'workflow': data['workflow'],
               'workflow': data['workflow'],
               'nodes': nodes,
               'nodes': nodes,
               'mapping': mapping,
               'mapping': mapping,
-              'node_mapping': node_mapping
+              'node_mapping': node_mapping,
+              'workflow_mapping': workflow_mapping
           }))
           }))
     return force_unicode(xml)  
     return force_unicode(xml)  
 
 
@@ -205,16 +209,19 @@ class Node():
     
     
     self._augment_data()
     self._augment_data()
     
     
-  def to_xml(self, mapping=None, node_mapping=None):
+  def to_xml(self, mapping=None, node_mapping=None, workflow_mapping=None):
     if mapping is None:
     if mapping is None:
       mapping = {}
       mapping = {}
     if node_mapping is None:
     if node_mapping is None:
       node_mapping = {}
       node_mapping = {}
+    if workflow_mapping is None:
+      workflow_mapping = {}
 
 
     data = {
     data = {
       'node': self.data,
       'node': self.data,
       'mapping': mapping,
       'mapping': mapping,
-      'node_mapping': node_mapping
+      'node_mapping': node_mapping,
+      'workflow_mapping': workflow_mapping
     }
     }
 
 
     return django_mako.render_to_string(self.get_template_name(), data)
     return django_mako.render_to_string(self.get_template_name(), data)
@@ -463,6 +470,38 @@ class HiveAction():
     return [cls.FIELDS['script_path']]
     return [cls.FIELDS['script_path']]
 
 
 
 
+class SubWorkflowAction():
+  TYPE = 'subworkflow'
+  FIELDS = {
+     'workflow': { 
+          'name': 'workflow',
+          'label': _('Sub-workflow'),
+          'value': None,
+          'help_text': _('The sub-workflow application to include. You must own all the sub-workflows')
+     },
+     'propagate_configuration': { 
+          'name': 'propagate_configuration',
+          'label': _('Propagate configuration'),
+          'value': True,
+          'help_text': _('If the workflow job configuration should be propagated to the child workflow.')
+     },
+     'job_properties': { 
+          'name': 'job_properties',
+          'label': _('Hadoop job properties'),
+          'value': [],
+          'help_text': _('Can be used to specify the job properties that are required to run the child workflow job.')
+     }
+  }
+
+  @classmethod
+  def get_fields(cls):
+    return [(f['name'], f['value']) for f in cls.FIELDS.itervalues()]
+  
+  @classmethod
+  def get_mandatory_fields(cls):
+    return []
+
+
 class KillAction():
 class KillAction():
   TYPE = 'kill'
   TYPE = 'kill'
   FIELDS = {
   FIELDS = {
@@ -487,6 +526,7 @@ NODES = {
   'pig-widget': PigAction,
   'pig-widget': PigAction,
   'java-widget': JavaAction,
   'java-widget': JavaAction,
   'hive-widget': HiveAction,
   'hive-widget': HiveAction,
+  'subworkflow-widget': SubWorkflowAction,
   'kill-widget': KillAction
   'kill-widget': KillAction
 }
 }
 
 

+ 1 - 1
apps/oozie/src/oozie/templates/editor/gen2/workflow-fork.xml.mako

@@ -17,6 +17,6 @@
 
 
     <fork name="${ node['name'] }">
     <fork name="${ node['name'] }">
         % for child in node['children']:
         % for child in node['children']:
-        <path start="${ node_mapping[child['to']] }" />
+        <path start="${ node_mapping[child['to']].name }" />
         % endfor
         % endfor
     </fork>
     </fork>

+ 2 - 2
apps/oozie/src/oozie/templates/editor/gen2/workflow-hive.xml.mako

@@ -36,7 +36,7 @@
 
 
             ${ common.distributed_cache(node['properties']['files'], node['properties']['archives']) }
             ${ common.distributed_cache(node['properties']['files'], node['properties']['archives']) }
         </hive>
         </hive>
-        <ok to="${ node_mapping[node['children'][0]['to']] }"/>
-        <error to="${ node_mapping[node['children'][1]['error']] }"/>
+        <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+        <error to="${ node_mapping[node['children'][1]['error']].name }"/>
         ${ common.sla(node) }
         ${ common.sla(node) }
     </action>
     </action>

+ 2 - 2
apps/oozie/src/oozie/templates/editor/gen2/workflow-java.xml.mako

@@ -44,7 +44,7 @@
             <capture-output/>
             <capture-output/>
             % endif
             % endif
         </java>
         </java>
-        <ok to="${ node_mapping[node['children'][0]['to']] }"/>
-        <error to="${ node_mapping[node['children'][1]['error']] }"/>
+        <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+        <error to="${ node_mapping[node['children'][1]['error']].name }"/>
         ${ common.sla(node) }
         ${ common.sla(node) }
     </action>
     </action>

+ 1 - 1
apps/oozie/src/oozie/templates/editor/gen2/workflow-join.xml.mako

@@ -15,4 +15,4 @@
 ## See the License for the specific language governing permissions and
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## limitations under the License.
 
 
-    <join name="${ node['name'] }" to="${ node_mapping[node['children'][0]['to']] }"/>
+    <join name="${ node['name'] }" to="${ node_mapping[node['children'][0]['to']].name }"/>

+ 2 - 2
apps/oozie/src/oozie/templates/editor/gen2/workflow-pig.xml.mako

@@ -39,7 +39,7 @@
 
 
             ${ common.distributed_cache(node['properties']['files'], node['properties']['archives']) }
             ${ common.distributed_cache(node['properties']['files'], node['properties']['archives']) }
         </pig>
         </pig>
-        <ok to="${ node_mapping[node['children'][0]['to']] }"/>
-        <error to="${ node_mapping[node['children'][1]['error']] }"/>
+        <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+        <error to="${ node_mapping[node['children'][1]['error']].name }"/>
         ${ common.sla(node) }
         ${ common.sla(node) }
     </action>
     </action>

+ 1 - 1
apps/oozie/src/oozie/templates/editor/gen2/workflow-start.xml.mako

@@ -15,4 +15,4 @@
 ## See the License for the specific language governing permissions and
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## limitations under the License.
 
 
-    <start to="${ node_mapping[node['children'][0]['to']] }"/>
+    <start to="${ node_mapping[node['children'][0]['to']].name }"/>

+ 6 - 6
apps/oozie/src/oozie/templates/editor/gen2/workflow-subworkflow.xml.mako

@@ -17,17 +17,17 @@
 
 
 <%namespace name="common" file="workflow-common.xml.mako" />
 <%namespace name="common" file="workflow-common.xml.mako" />
 
 
-    <action name="${ node }"${ common.credentials(node.credentials) }>
+    <action name="${ node['name'] }"${ common.credentials(node['properties']['credentials']) }>
         <sub-workflow>
         <sub-workflow>
-            <app-path>${'${'}nameNode}${ node.sub_workflow.deployment_dir }</app-path>
+            <app-path>${'${'}nameNode}${ workflow_mapping[node['properties']['workflow']].deployment_dir }</app-path>
 
 
-            % if node.propagate_configuration:
+            % if node['properties']['propagate_configuration']:
               <propagate-configuration/>
               <propagate-configuration/>
             % endif
             % endif
 
 
-            ${ common.configuration(node.get_properties()) }
+            ${ common.configuration(node['properties']['job_properties']) }
         </sub-workflow>
         </sub-workflow>
-        <ok to="${ node.get_oozie_child('ok') }"/>
-        <error to="${ node.get_oozie_child('error') }"/>
+        <ok to="${ node_mapping[node['children'][0]['to']].name }"/>
+        <error to="${ node_mapping[node['children'][1]['error']].name }"/>
         ${ common.sla(node) }
         ${ common.sla(node) }
     </action>
     </action>

+ 1 - 1
apps/oozie/src/oozie/templates/editor/gen2/workflow.xml.mako

@@ -47,7 +47,7 @@
   </credentials>
   </credentials>
   % endif
   % endif
   % for node in nodes:
   % for node in nodes:
-      ${ node.to_xml(mapping, node_mapping) | n }
+      ${ node.to_xml(mapping, node_mapping, workflow_mapping) | n }
   % endfor
   % endfor
   ${ common.sla(workflow) }
   ${ common.sla(workflow) }
 </workflow-app>
 </workflow-app>

+ 4 - 2
apps/oozie/src/oozie/templates/editor/workflow_editor.mako

@@ -72,7 +72,7 @@ ${ commonheader(_("Workflow Editor"), "Oozie", user) | n,unicode }
 
 
     <div data-bind="css: { 'draggable-widget': true },
     <div data-bind="css: { 'draggable-widget': true },
                     draggable: {data: draggableSubworkflowAction(), isEnabled: true,
                     draggable: {data: draggableSubworkflowAction(), isEnabled: true,
-                    options: {'start': function(event, ui){}}}"
+                    options: {'start': function(event, ui){$root.setCurrentDraggedWidget(draggableSubworkflowAction());}}}"
          title="${_('Sub workflow')}" rel="tooltip" data-placement="top">
          title="${_('Sub workflow')}" rel="tooltip" data-placement="top">
          <a class="draggable-icon"><i class="fa fa-code-fork"></i></a>
          <a class="draggable-icon"><i class="fa fa-code-fork"></i></a>
     </div>
     </div>
@@ -707,7 +707,9 @@ ${ commonheader(_("Workflow Editor"), "Oozie", user) | n,unicode }
       </ul>
       </ul>
       <div class="tab-content">
       <div class="tab-content">
         <div class="tab-pane active" id="action">
         <div class="tab-pane active" id="action">
-          <input type="text" data-bind="value: properties.subworkflow" />
+          <span data-bind="text: $root.workflow_properties.workflow.label"></span>
+          <input type="text" data-bind="value: properties.workflow" />
+          <select data-bind="options: $root.addActionWorkflows, optionsText: 'name', value: properties.selectedSubWorkflow"></select>
         </div>
         </div>
         <div class="tab-pane" id="properties">
         <div class="tab-pane" id="properties">
           <span data-bind="template: { name: 'common-action-properties' }"></span>
           <span data-bind="template: { name: 'common-action-properties' }"></span>

+ 9 - 9
apps/oozie/src/oozie/views/editor2.py

@@ -113,14 +113,6 @@ def save_workflow(request):
 
 
   return HttpResponse(json.dumps(response), mimetype="application/json")
   return HttpResponse(json.dumps(response), mimetype="application/json")
 
 
-#  elif node['widgetType'] == 'subworkflow-widget':
-#    workflows = [{
-#        'name': workflow.name,
-#        'owner': workflow.owner.username,
-#        'value': workflow.uuid
-#      } for workflow in Document2.objects.filter(type='oozie-workflow2', owner=request.user)
-#    ]
-
 
 
 def new_node(request):
 def new_node(request):
   response = {'status': -1}
   response = {'status': -1}
@@ -130,6 +122,14 @@ def new_node(request):
 
 
   properties = NODES[node['widgetType']].get_mandatory_fields()
   properties = NODES[node['widgetType']].get_mandatory_fields()
   workflows = []
   workflows = []
+
+  if node['widgetType'] == 'subworkflow-widget':
+    workflows = [{
+        'name': workflow.name,
+        'owner': workflow.owner.username,
+        'value': workflow.uuid
+      } for workflow in Document2.objects.filter(type='oozie-workflow2', owner=request.user)
+    ]
     
     
   response['status'] = 0
   response['status'] = 0
   response['properties'] = properties 
   response['properties'] = properties 
@@ -151,7 +151,7 @@ def add_node(request):
 
 
   if subworkflow:
   if subworkflow:
     _properties.update({
     _properties.update({
-       'subworkflow': subworkflow['value']
+       'workflow': subworkflow['value']
     })
     })
   _properties.update({
   _properties.update({
       'sla': Workflow.SLA_DEFAULT,
       'sla': Workflow.SLA_DEFAULT,

+ 4 - 4
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -147,10 +147,10 @@ class Submission(object):
     if hasattr(self.job, 'actions'):
     if hasattr(self.job, 'actions'):
       for action in self.job.actions:
       for action in self.job.actions:
         # Make sure XML is there
         # Make sure XML is there
-        # Don't support shared sub-worfklow
-        if action.node_type == 'subworkflow':
-          node = action.get_full_node()
-          sub_deploy = Submission(self.user, node.sub_workflow, self.fs, self.jt, self.properties)
+        # Don't support shared sub-worfklow, ore more than one level sub-workflow
+        if action.data['type'] == 'subworkflow':
+          workflow = Workflow(document=Document2.objects.get(uuid=action.data['properties']['workflow']))
+          sub_deploy = Submission(self.user, workflow, self.fs, self.jt, self.properties)
           sub_deploy.deploy()
           sub_deploy.deploy()
 
 
     return deployment_dir
     return deployment_dir