|
@@ -30,7 +30,9 @@ from jobsub.models import OozieDesign
|
|
|
|
|
|
|
|
from oozie.forms import WorkflowForm, ImportJobsubDesignForm, NodeForm, design_form_by_type
|
|
from oozie.forms import WorkflowForm, ImportJobsubDesignForm, NodeForm, design_form_by_type
|
|
|
from oozie.import_jobsub import convert_jobsub_design
|
|
from oozie.import_jobsub import convert_jobsub_design
|
|
|
-from oozie.models import Workflow, Node, Mapreduce, Java, Streaming, Link, NODE_TYPES, ACTION_TYPES, CONTROL_TYPES
|
|
|
|
|
|
|
+from oozie.models import Workflow, Node, Start, End, Kill, Mapreduce, Java, Streaming,\
|
|
|
|
|
+ Link, Decision, Fork, DecisionEnd, Join,\
|
|
|
|
|
+ NODE_TYPES, ACTION_TYPES, CONTROL_TYPES
|
|
|
from oozie.decorators import check_job_access_permission, check_job_edition_permission
|
|
from oozie.decorators import check_job_access_permission, check_job_edition_permission
|
|
|
from oozie.utils import model_to_dict
|
|
from oozie.utils import model_to_dict
|
|
|
|
|
|
|
@@ -58,12 +60,104 @@ def format_dict_field_values(dictionary):
|
|
|
return dictionary
|
|
return dictionary
|
|
|
|
|
|
|
|
|
|
|
|
|
-def workflow_validate_node_json(node_type, node_dict, errors, user, workflow):
|
|
|
|
|
|
|
+def get_or_create_node(workflow, node_data):
|
|
|
|
|
+ node = None
|
|
|
|
|
+ id = str(node_data['id'])
|
|
|
|
|
+ separator_index = id.find(':')
|
|
|
|
|
+
|
|
|
|
|
+ if separator_index == -1:
|
|
|
|
|
+ return Node.objects.get(id=id, workflow=workflow).get_full_node()
|
|
|
|
|
+
|
|
|
|
|
+ node_type = id[0:separator_index]
|
|
|
|
|
+ node_model = NODE_TYPES.get(node_type, None)
|
|
|
|
|
+ kwargs = {'workflow': workflow, 'node_type': node_data['node_type']}
|
|
|
|
|
+
|
|
|
|
|
+ if node_data['node_type'] == 'subworkflow':
|
|
|
|
|
+ kwargs['sub_workflow'] = Workflow.objects.get(id=int(node_data['sub_workflow']))
|
|
|
|
|
+
|
|
|
|
|
+ if node_model:
|
|
|
|
|
+ node = node_model(**kwargs)
|
|
|
|
|
+ else:
|
|
|
|
|
+ raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Could not find node of type'), data=node_data, error_code=500)
|
|
|
|
|
+ node.save()
|
|
|
|
|
+ return node
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _validate_node_links_json(node_type, node_links, errors):
|
|
|
"""
|
|
"""
|
|
|
- Validates a single node.
|
|
|
|
|
|
|
+ Validate a single node's links.
|
|
|
|
|
+ node_type is the node type of the action information passed.
|
|
|
|
|
+ node_links is list of dictionaries describing the links.
|
|
|
|
|
+ errors is a dictionary that will be populated with any found errors.
|
|
|
|
|
+ """
|
|
|
|
|
+ assert isinstance(errors, dict), "errors must be a dict."
|
|
|
|
|
+ if not isinstance(node_links, list):
|
|
|
|
|
+ errors['links'] = _("links must be a list")
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
|
|
+ # Check link counts are accurate.
|
|
|
|
|
+ if node_type == Start.node_type:
|
|
|
|
|
+ if len(node_links) != 2:
|
|
|
|
|
+ errors['links'] = _("Start should have two children: 'related' to end, 'to' to any node but an end.")
|
|
|
|
|
+ return False
|
|
|
|
|
+ elif node_type == End.node_type:
|
|
|
|
|
+ if len(node_links) != 0:
|
|
|
|
|
+ errors['links'] = _("End should have no children.")
|
|
|
|
|
+ return False
|
|
|
|
|
+ elif node_type == Kill.node_type:
|
|
|
|
|
+ if len(node_links) != 0:
|
|
|
|
|
+ errors['links'] = _("Kill should have no children.")
|
|
|
|
|
+ return False
|
|
|
|
|
+ elif node_type in (Join.node_type, DecisionEnd.node_type):
|
|
|
|
|
+ if len(node_links) != 1:
|
|
|
|
|
+ errors['links'] = _("Join and Decision End should have one child: 'to' to any node.")
|
|
|
|
|
+ return False
|
|
|
|
|
+ elif node_type in (Fork.node_type, Decision.node_type):
|
|
|
|
|
+ if len(node_links) < 2:
|
|
|
|
|
+ errors['links'] = _("Join and Decision should have at least two children: 'related' to their respective ends, 'start' to any node.")
|
|
|
|
|
+ return False
|
|
|
|
|
+ else:
|
|
|
|
|
+ if len(node_links) != 2:
|
|
|
|
|
+ errors['links'] = _("Actions should have two children: 'error' to kill, 'ok' to any node.")
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
|
|
+ # Check if link types are okay.
|
|
|
|
|
+ link_names_by_node_type = {
|
|
|
|
|
+ 'start': {'related': 1, 'to': 1},
|
|
|
|
|
+ 'end': {},
|
|
|
|
|
+ 'kill': {},
|
|
|
|
|
+ 'fork': {'related': 1, 'start': 2},
|
|
|
|
|
+ 'join': {'to': 1},
|
|
|
|
|
+ 'decision': {'related': 1, 'start': 2},
|
|
|
|
|
+ 'decisionend': {'to': 1},
|
|
|
|
|
+ None: {'ok': 1, 'error': 1},
|
|
|
|
|
+ }
|
|
|
|
|
+ link_types = link_names_by_node_type.get(node_type, link_names_by_node_type[None])
|
|
|
|
|
+ for link in node_links:
|
|
|
|
|
+ link_name = link.get('name', None)
|
|
|
|
|
+ if link_name in link_types:
|
|
|
|
|
+ link_types[link_name] -= 1
|
|
|
|
|
+
|
|
|
|
|
+ for link_type in link_types:
|
|
|
|
|
+ if link_types[link_type] > 0:
|
|
|
|
|
+ errors['links'] = _('%(node_type)s should have %(count)d more %(link_type)s link' % {
|
|
|
|
|
+ 'node_type': node_type,
|
|
|
|
|
+ 'count': link_types[link_type],
|
|
|
|
|
+ 'link_type': link_type
|
|
|
|
|
+ })
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
|
|
+ return True
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _validate_node_json(node_type, node_dict, errors, user, workflow):
|
|
|
|
|
+ """
|
|
|
|
|
+ Validates a single node excluding links.
|
|
|
node_type is the node type of the action information passed.
|
|
node_type is the node type of the action information passed.
|
|
|
node_dict is a dictionary describing the node.
|
|
node_dict is a dictionary describing the node.
|
|
|
errors is a dictionary that will be populated with any found errors.
|
|
errors is a dictionary that will be populated with any found errors.
|
|
|
|
|
+ user is a User object that is associated with the node_type. Only needed for Subworkflow node.
|
|
|
|
|
+ workflow is the Workflow object associated with the node. Only needed for Subworkflow node.
|
|
|
Returns Boolean.
|
|
Returns Boolean.
|
|
|
"""
|
|
"""
|
|
|
assert isinstance(errors, dict), "errors must be a dict."
|
|
assert isinstance(errors, dict), "errors must be a dict."
|
|
@@ -86,49 +180,37 @@ def workflow_validate_node_json(node_type, node_dict, errors, user, workflow):
|
|
|
return False
|
|
return False
|
|
|
|
|
|
|
|
|
|
|
|
|
-def get_or_create_node(workflow, node_data):
|
|
|
|
|
- node = None
|
|
|
|
|
- id = str(node_data['id'])
|
|
|
|
|
- separator_index = id.find(':')
|
|
|
|
|
-
|
|
|
|
|
- if separator_index == -1:
|
|
|
|
|
- return Node.objects.get(id=id, workflow=workflow).get_full_node()
|
|
|
|
|
-
|
|
|
|
|
- node_type = id[0:separator_index]
|
|
|
|
|
- node_model = NODE_TYPES.get(node_type, None)
|
|
|
|
|
- kwargs = {'workflow': workflow, 'node_type': node_data['node_type']}
|
|
|
|
|
-
|
|
|
|
|
- if node_data['node_type'] == 'subworkflow':
|
|
|
|
|
- kwargs['sub_workflow'] = Workflow.objects.get(id=int(node_data['sub_workflow']))
|
|
|
|
|
-
|
|
|
|
|
- if node_model:
|
|
|
|
|
- node = node_model(**kwargs)
|
|
|
|
|
- else:
|
|
|
|
|
- raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Could not find node of type'), data=node_data, error_code=500)
|
|
|
|
|
- node.save()
|
|
|
|
|
- return node
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
-def update_workflow(json_workflow):
|
|
|
|
|
- workflow = Workflow.objects.get(id=json_workflow['id'])
|
|
|
|
|
-
|
|
|
|
|
- for key in json_workflow:
|
|
|
|
|
- if key not in ('nodes', 'start', 'end', 'job_ptr', 'owner'):
|
|
|
|
|
- setattr(workflow, key, json_workflow[key])
|
|
|
|
|
|
|
+def _validate_nodes_json(json_nodes, errors, user, workflow):
|
|
|
|
|
+ """
|
|
|
|
|
+ Validates every node and link in the workflow.
|
|
|
|
|
+ node_type is the node type of the action information passed.
|
|
|
|
|
+ node_dict is a dictionary describing the node.
|
|
|
|
|
+ errors is a dictionary that will be populated with any found errors.
|
|
|
|
|
+ user is a User object that is associated with the node_type. Only needed for Subworkflow node.
|
|
|
|
|
+ workflow is the Workflow object associated with the node. Only needed for Subworkflow node.
|
|
|
|
|
+ Returns Boolean.
|
|
|
|
|
+ """
|
|
|
|
|
+ assert isinstance(errors, dict), "errors must be a dict."
|
|
|
|
|
+ result = True
|
|
|
|
|
|
|
|
- workflow.save()
|
|
|
|
|
|
|
+ for node in json_nodes:
|
|
|
|
|
+ _errors = {}
|
|
|
|
|
+ node_dict = format_dict_field_values(node)
|
|
|
|
|
+ if node['node_type'] in ACTION_TYPES:
|
|
|
|
|
+ node_result = _validate_node_json(node['node_type'], node_dict, _errors, user, workflow)
|
|
|
|
|
+ else:
|
|
|
|
|
+ node_result = True
|
|
|
|
|
+ link_result = _validate_node_links_json(node['node_type'], node_dict['child_links'], _errors)
|
|
|
|
|
+ result = result and node_result and link_result
|
|
|
|
|
+ if not node.has_key('name') and ( not node.has_key('node_type') or not node.has_key('id') ):
|
|
|
|
|
+ raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': 'Node is missing a name.'}, error_code=400)
|
|
|
|
|
+ errors[node.get('name', '%s-%s' % ( node.get('node_type'), node.get('id')))] = _errors
|
|
|
|
|
|
|
|
- return workflow
|
|
|
|
|
|
|
+ return result
|
|
|
|
|
|
|
|
|
|
|
|
|
-def update_workflow_nodes(workflow, json_nodes, id_map, user):
|
|
|
|
|
|
|
+def _update_workflow_nodes_json(workflow, json_nodes, id_map, user):
|
|
|
"""Ideally would get objects from form validation instead."""
|
|
"""Ideally would get objects from form validation instead."""
|
|
|
- for json_node in json_nodes:
|
|
|
|
|
- errors = {}
|
|
|
|
|
- if json_node['node_type'] in ACTION_TYPES and \
|
|
|
|
|
- not workflow_validate_node_json(json_node['node_type'], format_dict_field_values(json_node), errors, user, workflow):
|
|
|
|
|
- raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Invalid action'), data={'errors': errors}, error_code=400)
|
|
|
|
|
-
|
|
|
|
|
nodes = []
|
|
nodes = []
|
|
|
|
|
|
|
|
for json_node in json_nodes:
|
|
for json_node in json_nodes:
|
|
@@ -159,6 +241,35 @@ def update_workflow_nodes(workflow, json_nodes, id_map, user):
|
|
|
return nodes
|
|
return nodes
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+def _update_workflow_json(json_workflow):
|
|
|
|
|
+ workflow = Workflow.objects.get(id=json_workflow['id'])
|
|
|
|
|
+
|
|
|
|
|
+ for key in json_workflow:
|
|
|
|
|
+ if key not in ('nodes', 'start', 'end', 'job_ptr', 'owner'):
|
|
|
|
|
+ setattr(workflow, key, json_workflow[key])
|
|
|
|
|
+
|
|
|
|
|
+ workflow.save()
|
|
|
|
|
+
|
|
|
|
|
+ return workflow
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _workflow(request, workflow):
|
|
|
|
|
+ response = {'status': -1, 'data': 'None'}
|
|
|
|
|
+
|
|
|
|
|
+ workflow_dict = model_to_dict(workflow)
|
|
|
|
|
+ node_list = [node.get_full_node() for node in workflow.node_list]
|
|
|
|
|
+ nodes = [model_to_dict(node) for node in node_list]
|
|
|
|
|
+
|
|
|
|
|
+ for index in range(0, len(node_list)):
|
|
|
|
|
+ nodes[index]['child_links'] = [model_to_dict(link) for link in node_list[index].get_all_children_links()]
|
|
|
|
|
+
|
|
|
|
|
+ workflow_dict['nodes'] = nodes
|
|
|
|
|
+
|
|
|
|
|
+ response['status'] = 0
|
|
|
|
|
+ response['data'] = workflow_dict
|
|
|
|
|
+ return HttpResponse(json.dumps(response), mimetype="application/json")
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
|
@check_job_edition_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
@check_job_edition_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
|
def workflow_validate_node(request, workflow, node_type):
|
|
def workflow_validate_node(request, workflow, node_type):
|
|
@@ -166,7 +277,7 @@ def workflow_validate_node(request, workflow, node_type):
|
|
|
|
|
|
|
|
node_dict = format_dict_field_values(json.loads(str(request.POST.get('node'))))
|
|
node_dict = format_dict_field_values(json.loads(str(request.POST.get('node'))))
|
|
|
|
|
|
|
|
- if workflow_validate_node_json(node_type, node_dict, response['data'], request.user, workflow):
|
|
|
|
|
|
|
+ if _validate_node_json(node_type, node_dict, response['data'], request.user, workflow):
|
|
|
response['status'] = 0
|
|
response['status'] = 0
|
|
|
else:
|
|
else:
|
|
|
response['status'] = -1
|
|
response['status'] = -1
|
|
@@ -191,9 +302,13 @@ def workflow_save(request, workflow):
|
|
|
|
|
|
|
|
json_nodes = json_workflow['nodes']
|
|
json_nodes = json_workflow['nodes']
|
|
|
id_map = {}
|
|
id_map = {}
|
|
|
|
|
+ errors = {}
|
|
|
|
|
+
|
|
|
|
|
+ if not _validate_nodes_json(json_nodes, errors, request.user, workflow):
|
|
|
|
|
+ raise StructuredException(code="INVALID_REQUEST_ERROR", message=_('Error saving workflow'), data={'errors': errors}, error_code=400)
|
|
|
|
|
|
|
|
- workflow = update_workflow(json_workflow)
|
|
|
|
|
- nodes = update_workflow_nodes(workflow, json_nodes, id_map, request.user)
|
|
|
|
|
|
|
+ workflow = _update_workflow_json(json_workflow)
|
|
|
|
|
+ nodes = _update_workflow_nodes_json(workflow, json_nodes, id_map, request.user)
|
|
|
|
|
|
|
|
# Update links
|
|
# Update links
|
|
|
index = 0
|
|
index = 0
|
|
@@ -224,23 +339,6 @@ def workflow_save(request, workflow):
|
|
|
return _workflow(request, workflow=workflow)
|
|
return _workflow(request, workflow=workflow)
|
|
|
|
|
|
|
|
|
|
|
|
|
-def _workflow(request, workflow):
|
|
|
|
|
- response = {'status': -1, 'data': 'None'}
|
|
|
|
|
-
|
|
|
|
|
- workflow_dict = model_to_dict(workflow)
|
|
|
|
|
- node_list = [node.get_full_node() for node in workflow.node_list]
|
|
|
|
|
- nodes = [model_to_dict(node) for node in node_list]
|
|
|
|
|
-
|
|
|
|
|
- for index in range(0, len(node_list)):
|
|
|
|
|
- nodes[index]['child_links'] = [model_to_dict(link) for link in node_list[index].get_all_children_links()]
|
|
|
|
|
-
|
|
|
|
|
- workflow_dict['nodes'] = nodes
|
|
|
|
|
-
|
|
|
|
|
- response['status'] = 0
|
|
|
|
|
- response['data'] = workflow_dict
|
|
|
|
|
- return HttpResponse(json.dumps(response), mimetype="application/json")
|
|
|
|
|
-
|
|
|
|
|
-
|
|
|
|
|
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
@check_job_access_permission(exception_class=(lambda x: StructuredException(code="UNAUTHORIZED_REQUEST_ERROR", message=x, data=None, error_code=401)))
|
|
|
def workflow(request, workflow):
|
|
def workflow(request, workflow):
|
|
|
if request.method != 'GET':
|
|
if request.method != 'GET':
|