|
|
@@ -36,12 +36,14 @@ from desktop.lib.json_utils import JSONEncoderForHTML
|
|
|
from desktop.models import Document2
|
|
|
|
|
|
from hadoop.fs.hadoopfs import Hdfs
|
|
|
+from hadoop.fs.exceptions import WebHdfsException
|
|
|
+
|
|
|
from liboozie.submission2 import Submission
|
|
|
from liboozie.submission2 import create_directories
|
|
|
|
|
|
from oozie.conf import REMOTE_SAMPLE_DIR
|
|
|
from oozie.utils import utc_datetime_format, UTC_TIME_FORMAT, convert_to_server_timezone
|
|
|
-from hadoop.fs.exceptions import WebHdfsException
|
|
|
+from oozie.importlib.workflows import generate_v2_graph_nodes, MalformedWfDefException, InvalidTagWithNamespaceException
|
|
|
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
@@ -346,6 +348,264 @@ class Workflow(Job):
|
|
|
def get_application_path_key(cls):
|
|
|
return 'oozie.wf.application.path'
|
|
|
|
|
|
+ @classmethod
|
|
|
+ def gen_workflow_data_from_xml(cls, user, oozie_workflow):
|
|
|
+ node_list = []
|
|
|
+ try:
|
|
|
+ node_list = generate_v2_graph_nodes(oozie_workflow.definition)
|
|
|
+ except MalformedWfDefException, e:
|
|
|
+ LOG.exception("Could not find any nodes in Workflow definition. Maybe it's malformed?")
|
|
|
+ except InvalidTagWithNamespaceException, e:
|
|
|
+ LOG.exception("Tag with namespace %(namespace)s is not valid. Please use one of the following namespaces: %(namespaces)s" % {
|
|
|
+ 'namespace': e.namespace,
|
|
|
+ 'namespaces': e.namespaces
|
|
|
+ })
|
|
|
+
|
|
|
+ adj_list = _create_graph_adjaceny_list(node_list)
|
|
|
+
|
|
|
+ node_hierarchy = ['start']
|
|
|
+ _get_hierarchy_from_adj_list(adj_list, adj_list['start']['ok_to'], node_hierarchy)
|
|
|
+
|
|
|
+ _add_uuids_to_adj_list(adj_list)
|
|
|
+
|
|
|
+ wf_rows = _create_workflow_layout(node_hierarchy, adj_list)
|
|
|
+ data = {'layout': [{}], 'workflow': {}}
|
|
|
+ if wf_rows:
|
|
|
+ data['layout'][0]['rows'] = wf_rows
|
|
|
+
|
|
|
+ wf_nodes = []
|
|
|
+ _dig_nodes(node_hierarchy, adj_list, user, wf_nodes)
|
|
|
+ data['workflow']['nodes'] = wf_nodes
|
|
|
+ data['workflow']['id'] = "123"
|
|
|
+ data['workflow']['properties'] = json.loads("""{
|
|
|
+ "job_xml": "",
|
|
|
+ "description": "",
|
|
|
+ "wf1_id": null,
|
|
|
+ "sla_enabled": false,
|
|
|
+ "deployment_dir": "/user/hue/oozie/workspaces/hue-oozie-1452553957.19",
|
|
|
+ "schema_version": "uri:oozie:workflow:0.5",
|
|
|
+ "sla": [
|
|
|
+ {
|
|
|
+ "key": "enabled",
|
|
|
+ "value": false
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "nominal-time",
|
|
|
+ "value": "${nominal_time}"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "should-start",
|
|
|
+ "value": ""
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "should-end",
|
|
|
+ "value": "${30 * MINUTES}"
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "max-duration",
|
|
|
+ "value": ""
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "alert-events",
|
|
|
+ "value": ""
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "alert-contact",
|
|
|
+ "value": ""
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "notification-msg",
|
|
|
+ "value": ""
|
|
|
+ },
|
|
|
+ {
|
|
|
+ "key": "upstream-apps",
|
|
|
+ "value": ""
|
|
|
+ }
|
|
|
+ ],
|
|
|
+ "show_arrows": true,
|
|
|
+ "parameters": [
|
|
|
+ {
|
|
|
+ "name": "oozie.use.system.libpath",
|
|
|
+ "value": true
|
|
|
+ }
|
|
|
+ ],
|
|
|
+ "properties": []
|
|
|
+ }""")
|
|
|
+
|
|
|
+ return data
|
|
|
+
|
|
|
+def _add_uuids_to_adj_list(adj_list):
|
|
|
+ uuids = {}
|
|
|
+ id = 1
|
|
|
+ for node in adj_list.keys():
|
|
|
+ adj_list[node]['id'] = id
|
|
|
+
|
|
|
+ if adj_list[node]['node_type'] == 'kill':
|
|
|
+ adj_list[node]['uuid'] = '17c9c895-5a16-7443-bb81-f34b30b21548'
|
|
|
+ elif adj_list[node]['node_type'] == 'start':
|
|
|
+ adj_list[node]['uuid'] = '3f107997-04cc-8733-60a9-a4bb62cebffc'
|
|
|
+ elif adj_list[node]['node_type'] == 'end':
|
|
|
+ adj_list[node]['uuid'] = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
|
|
|
+ else:
|
|
|
+ adj_list[node]['uuid'] = node[-4:] + str(uuid.uuid4())[4:]
|
|
|
+
|
|
|
+ uuids[id] = adj_list[node]['uuid']
|
|
|
+ id += 1
|
|
|
+ return adj_list
|
|
|
+
|
|
|
+def _dig_nodes(nodes, adj_list, user, wf_nodes):
|
|
|
+ for node in nodes:
|
|
|
+ if type(node) != list:
|
|
|
+ node = adj_list[node]
|
|
|
+ properties = {}
|
|
|
+ if '%s-widget' % node['node_type'] in NODES:
|
|
|
+ properties = dict(NODES['%s-widget' % node['node_type']].get_fields())
|
|
|
+
|
|
|
+ if node['node_type'] == 'pig':
|
|
|
+ properties['script_path'] = node.get('script_path')
|
|
|
+ elif node['node_type'] == 'spark':
|
|
|
+ properties['class'] = node.get('spark').get('class')
|
|
|
+ #TBD: jar_path
|
|
|
+ properties['jar_path'] = node.get('spark').get('jar')
|
|
|
+ elif node['node_type'] == 'hive' or node['node_type'] == 'hive2':
|
|
|
+ properties['script_path'] = node.get('hive2').get('script')
|
|
|
+ elif node['node_type'] == 'java':
|
|
|
+ properties['jar_path'] = node.get('jar_path')
|
|
|
+ elif node['node_type'] == 'sqoop':
|
|
|
+ properties['command'] = node.get('script_path')
|
|
|
+ elif node['node_type'] == 'mapreduce':
|
|
|
+ properties['jar_path'] = node.get('jar_path')
|
|
|
+ elif node['node_type'] == 'shell':
|
|
|
+ properties['shell_command'] = node.get('shell').get('command')
|
|
|
+ elif node['node_type'] == 'ssh':
|
|
|
+ properties['user'] = '%s@%s' % (node.get('ssh').get('user'), node.get('ssh').get('host'))
|
|
|
+ properties['ssh_command'] = node.get('ssh').get('command')
|
|
|
+ elif node['node_type'] == 'fs':
|
|
|
+ #TBD: all
|
|
|
+ properties['deletes'] = [{'value': f['name']} for f in json.loads(node.get('deletes'))]
|
|
|
+ properties['mkdirs'] = [{'value': f['name']} for f in json.loads(node.get('mkdirs'))]
|
|
|
+ properties['moves'] = json.loads(node.get('moves'))
|
|
|
+ properties['touchzs'] = [{'value': f['name']} for f in json.loads(node.get('touchzs'))]
|
|
|
+ elif node['node_type'] == 'email':
|
|
|
+ properties['to'] = node.get('email').get('to')
|
|
|
+ properties['subject'] = node.get('email').get('subject')
|
|
|
+ #TBD: body doesn't show up
|
|
|
+ properties['body'] = node.get('email').get('body')
|
|
|
+ elif node['node_type'] == 'streaming':
|
|
|
+ properties['mapper'] = node.get('streaming').get('mapper')
|
|
|
+ properties['reducer'] = node.get('streaming').get('reducer')
|
|
|
+ elif node['node_type'] == 'distcp':
|
|
|
+ #TBD: both
|
|
|
+ properties['source'] = node.get('distcp').get('source')
|
|
|
+ properties['destination'] = node.get('distcp').get('source')
|
|
|
+ elif node['node_type'] == 'sub-workflow':
|
|
|
+ properties['app-path'] = node.get('sub-workflow').get('app-path')
|
|
|
+ properties['workflow'] = node.get('uuid')
|
|
|
+ properties['job_properties'] = []
|
|
|
+ properties['sla'] = ''
|
|
|
+
|
|
|
+ children = []
|
|
|
+ if node['node_type'] == 'fork':
|
|
|
+ for key in node.keys():
|
|
|
+ if key.startswith('path'):
|
|
|
+ children.append({'to': adj_list[node[key]]['uuid'], 'condition': '${ 1 gt 0 }'})
|
|
|
+ else:
|
|
|
+ if node.get('ok_to'):
|
|
|
+ children.append({'to': adj_list[node['ok_to']]['uuid']})
|
|
|
+ if node.get('error_to'):
|
|
|
+ children.append({'error': adj_list[node['error_to']]['uuid']})
|
|
|
+
|
|
|
+ wf_nodes.append({
|
|
|
+ "id": node['uuid'],
|
|
|
+ "name": '%s-%s' % (node['node_type'].split('-')[0], node['uuid'][:4]),
|
|
|
+ "type": "%s-widget" % node['node_type'],
|
|
|
+ "properties": properties,
|
|
|
+ "children": children
|
|
|
+ })
|
|
|
+ else:
|
|
|
+ _dig_nodes(node, adj_list, user, wf_nodes)
|
|
|
+
|
|
|
+def _create_workflow_layout(nodes, adj_list, size=12):
|
|
|
+ wf_rows = []
|
|
|
+ for node in nodes:
|
|
|
+ if type(node) == list and len(node) == 1:
|
|
|
+ node = node[0]
|
|
|
+ if type(node) != list:
|
|
|
+ wf_rows.append({"widgets":[{"size":size, "name": adj_list[node]['node_type'], "id": adj_list[node]['uuid'], "widgetType": "%s-widget" % adj_list[node]['node_type'] if adj_list[node]['node_type'] != 'sub-workflow' else 'subworkflow-widget', "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
|
|
|
+ else:
|
|
|
+ if adj_list[node[0]]['node_type'] == 'fork':
|
|
|
+ wf_rows.append({"widgets":[{"size":size, "name": 'Fork', "id": adj_list[node[0]]['uuid'], "widgetType": "%s-widget" % adj_list[node[0]]['node_type'], "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
|
|
|
+
|
|
|
+ wf_rows.append({
|
|
|
+ "id": str(uuid.uuid4()),
|
|
|
+ "widgets":[
|
|
|
+
|
|
|
+ ],
|
|
|
+ "columns":[
|
|
|
+ {
|
|
|
+ "id": str(uuid.uuid4()),
|
|
|
+ "size": (size / len(node[1])),
|
|
|
+ "rows":
|
|
|
+ [{
|
|
|
+ "id": str(uuid.uuid4()),
|
|
|
+ "widgets": c['widgets'],
|
|
|
+ "columns":c.get('columns') or []
|
|
|
+ } for c in col],
|
|
|
+ "klass":"card card-home card-column span%s" % (size / len(node[1]))
|
|
|
+ }
|
|
|
+ for col in [_create_workflow_layout(item, adj_list, size) for item in node[1]]
|
|
|
+ ]
|
|
|
+ })
|
|
|
+
|
|
|
+ wf_rows.append({"widgets":[{"size":size, "name": 'Join', "id": adj_list[node[2]]['uuid'], "widgetType": "%s-widget" % adj_list[node[2]]['node_type'], "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
|
|
|
+ else:
|
|
|
+ wf_rows.append(_create_workflow_layout(node, adj_list, size))
|
|
|
+ return wf_rows
|
|
|
+
|
|
|
+
|
|
|
+def _get_hierarchy_from_adj_list(adj_list, curr_node, node_hierarchy):
|
|
|
+
|
|
|
+ if adj_list[curr_node]['node_type'] == 'join':
|
|
|
+ return curr_node
|
|
|
+
|
|
|
+ elif adj_list[curr_node]['node_type'] == 'end':
|
|
|
+ node_hierarchy.append(['Kill'])
|
|
|
+ node_hierarchy.append(['End'])
|
|
|
+ return node_hierarchy
|
|
|
+
|
|
|
+ elif adj_list[curr_node]['node_type'] == 'fork':
|
|
|
+ fork_nodes = []
|
|
|
+ fork_nodes.append(curr_node)
|
|
|
+
|
|
|
+ join_node = None
|
|
|
+ children = []
|
|
|
+ for key in adj_list[curr_node].keys():
|
|
|
+ if key.startswith('path'):
|
|
|
+ child = []
|
|
|
+ join_node = _get_hierarchy_from_adj_list(adj_list, adj_list[curr_node][key], child)
|
|
|
+ children.append(child)
|
|
|
+
|
|
|
+ fork_nodes.append(children)
|
|
|
+ fork_nodes.append(join_node)
|
|
|
+
|
|
|
+ node_hierarchy.append(fork_nodes)
|
|
|
+ return _get_hierarchy_from_adj_list(adj_list, adj_list[join_node]['ok_to'], node_hierarchy)
|
|
|
+
|
|
|
+ else:
|
|
|
+ node_hierarchy.append(curr_node)
|
|
|
+ return _get_hierarchy_from_adj_list(adj_list, adj_list[curr_node]['ok_to'], node_hierarchy)
|
|
|
+
|
|
|
+
|
|
|
+def _create_graph_adjaceny_list(nodes):
|
|
|
+ start_node = [node for node in nodes if node.get('node_type') == 'start'][0]
|
|
|
+ adj_list = {'start': start_node}
|
|
|
+
|
|
|
+ for node in nodes:
|
|
|
+ if node and node.get('node_type') != 'start':
|
|
|
+ adj_list[node['name']] = node
|
|
|
+
|
|
|
+ return adj_list
|
|
|
+
|
|
|
|
|
|
class Node():
|
|
|
def __init__(self, data):
|