ariatosca-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mxm...@apache.org
Subject incubator-ariatosca git commit: Add transcription mechanism from task_graph into execution_graph [Forced Update!]
Date Tue, 18 Oct 2016 08:34:20 GMT
Repository: incubator-ariatosca
Updated Branches:
  refs/heads/operation_graph ee1a69170 -> 8d44901fc (forced update)


Add transcription mechanism from task_graph into execution_graph


Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/8d44901f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/8d44901f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/8d44901f

Branch: refs/heads/operation_graph
Commit: 8d44901fcb5f6a9405ac4167c6b2aea681c9aef0
Parents: b53aa1f
Author: mxmrlv <mxmrlv@gmail.com>
Authored: Thu Oct 13 19:52:35 2016 +0300
Committer: mxmrlv <mxmrlv@gmail.com>
Committed: Tue Oct 18 11:33:38 2016 +0300

----------------------------------------------------------------------
 aria/workflows/engine/engine.py                 | 98 ++++++++++++++------
 aria/workflows/engine/tasks.py                  | 72 ++++++++++++++
 tests/engine/__init__.py                        | 38 ++++++++
 .../test_task_graph_into_exececution_graph.py   | 85 +++++++++++++++++
 tests/requirements.txt                          |  1 +
 5 files changed, 267 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/8d44901f/aria/workflows/engine/engine.py
----------------------------------------------------------------------
diff --git a/aria/workflows/engine/engine.py b/aria/workflows/engine/engine.py
index 7b86fb5..4229fe9 100644
--- a/aria/workflows/engine/engine.py
+++ b/aria/workflows/engine/engine.py
@@ -28,6 +28,16 @@ from aria.events import (
     on_failure_task_signal,
 )
 from aria.logger import LoggerMixin
+from aria.contexts import OperationContext
+
+
+from .tasks import (
+    StartWorkflowTask,
+    EndWorkflowTask,
+    StartSubWorkflowTask,
+    EndSubWorkflowTask,
+    OperationTask,
+)
 
 
 class Engine(LoggerMixin):
@@ -38,10 +48,67 @@ class Engine(LoggerMixin):
         self._tasks_graph = tasks_graph
         self._execution_graph = DiGraph()
         self._executor = executor
-        self._build_execution_graph(self._workflow_context, self._tasks_graph)
-
-    def _build_execution_graph(self, workflow_context, graph):
-        pass
+        self._start_task_name = 'Starting.{0}'.format
+        self._end_task_name = 'Ending.{0}'.format
+        self._build_execution_graph(self._tasks_graph)
+
+    def _build_execution_graph(
+            self,
+            graph,
+            StartClass=StartWorkflowTask,
+            EndClass=EndWorkflowTask,
+            depends_on=()):
+
+        # Insert start marker
+        start_task = StartClass(name=self._start_task_name(graph.name), context=self._workflow_context)
+        self._add_task_and_dependencies(start_task, depends_on)
+
+        for operation_or_workflow, dependencies in graph.task_tree(reverse=True):
+            operation_dependencies = self._get_tasks_from_dependencies(dependencies, default=[start_task])
+
+            if self._is_operation(operation_or_workflow):
+                # Add the task an the dependencies
+                operation_task = OperationTask(
+                    name=operation_or_workflow.name,
+                    context=operation_or_workflow,
+                    **operation_or_workflow.engine_options
+                )
+                self._add_task_and_dependencies(operation_task, operation_dependencies)
+            else:
+                # Built the graph recursively while adding start and end markers
+                self._build_execution_graph(
+                    graph=operation_or_workflow,
+                    StartClass=StartSubWorkflowTask,
+                    EndClass=EndSubWorkflowTask,
+                    depends_on=operation_dependencies
+                )
+
+        # Insert end marker
+        workflow_dependencies = self._get_tasks_from_dependencies(graph.leaf_tasks, default=[start_task])
+        end_task = EndClass(name=self._end_task_name(graph.name), context=self._workflow_context)
+        self._add_task_and_dependencies(end_task, workflow_dependencies)
+
+    def _add_task_and_dependencies(self, operation_task, operation_dependencies=()):
+        self._execution_graph.add_node(operation_task.name, task=operation_task)
+        for dependency in operation_dependencies:
+            self._execution_graph.add_edge(dependency.name, operation_task.name)
+
+    def _get_tasks_from_dependencies(self, dependencies, default=()):
+        """
+        Returns task list from dependencies.
+        """
+        tasks = [self._execution_graph.node[
+                    dependency.name
+                    if self._is_operation(dependency) else
+                    # The task is an entire subgraph, so we need to check that the wrapper
task has been reached
+                    self._end_task_name(dependency.name)]['task']
+                 for dependency in dependencies]
+
+        return tasks or default
+
+    @staticmethod
+    def _is_operation(task):
+        return isinstance(task, OperationContext)
 
     def execute(self):
         execution_id = self._workflow_context.execution_id
@@ -160,26 +227,3 @@ class Engine(LoggerMixin):
             start_task_signal.disconnect(self._task_started_receiver)
             on_success_task_signal.disconnect(self._task_succeeded_receiver)
             on_failure_task_signal.disconnect(self._task_failed_receiver)
-
-
-class Task(object):
-
-    def __init__(self, operation_context):
-        self.operation_context = operation_context
-        self._create_operation_in_storage()
-
-    def _create_operation_in_storage(self):
-        Operation = self.operation_context.storage.operation.model_cls
-        operation = Operation(
-            id=self.operation_context.id,
-            execution_id=self.operation_context.execution_id,
-            max_retries=self.operation_context.parameters.get('max_retries', 1),
-            status=Operation.PENDING,
-        )
-        self.operation_context.operation = operation
-
-    def __getattr__(self, attr):
-        try:
-            return getattr(self.operation_context, attr)
-        except AttributeError:
-            return super(Task, self).__getattribute__(attr)

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/8d44901f/aria/workflows/engine/tasks.py
----------------------------------------------------------------------
diff --git a/aria/workflows/engine/tasks.py b/aria/workflows/engine/tasks.py
new file mode 100644
index 0000000..1403ce5
--- /dev/null
+++ b/aria/workflows/engine/tasks.py
@@ -0,0 +1,72 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+class BaseTask(object):
+
+    def __init__(self, name, context):
+        self.name = name
+        self.context = context
+
+
+class MarkerTask(BaseTask):
+    pass
+
+
+class WorkflowMarkerTask(MarkerTask):
+    pass
+
+
+class StartWorkflowTask(MarkerTask):
+    pass
+
+
+class EndWorkflowTask(MarkerTask):
+    pass
+
+
+class SubWorkflowMarkerTask(MarkerTask):
+    pass
+
+
+class StartSubWorkflowTask(SubWorkflowMarkerTask):
+    pass
+
+
+class EndSubWorkflowTask(SubWorkflowMarkerTask):
+    pass
+
+
+class OperationTask(BaseTask):
+    def __init__(self, *args, **kwargs):
+        super(OperationTask, self).__init__(*args, **kwargs)
+        self._create_operation_in_storage()
+
+    def _create_operation_in_storage(self):
+        Operation = self.operation_context.storage.operation.model_cls
+        operation = Operation(
+            id=self.operation_context.id,
+            execution_id=self.operation_context.execution_id,
+            max_retries=self.operation_context.parameters.get('max_retries', 1),
+            status=Operation.PENDING,
+        )
+        self.operation_context.operation = operation
+
+    def __getattr__(self, attr):
+        try:
+            return getattr(self.operation_context, attr)
+        except AttributeError:
+            return super(OperationTask, self).__getattribute__(attr)
+

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/8d44901f/tests/engine/__init__.py
----------------------------------------------------------------------
diff --git a/tests/engine/__init__.py b/tests/engine/__init__.py
new file mode 100644
index 0000000..e7533fd
--- /dev/null
+++ b/tests/engine/__init__.py
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from uuid import uuid4
+
+from aria.workflows.engine.executor import Executor
+from aria.workflows.engine.tasks import OperationTask
+from aria.contexts import OperationContext
+
+
+class MockExecutor(Executor):
+    def execute(self, task):
+        print task
+
+
+class MockOperationContext(OperationContext):
+    def __init__(self, id=str(uuid4()), **kwargs):
+        self.id = self.name = id
+        self.engine_options = {}
+
+    def __repr__(self):
+        return self.id
+
+
+class MockOperationTask(OperationTask):
+    def _create_operation_in_storage(self):
+        pass

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/8d44901f/tests/engine/test_task_graph_into_exececution_graph.py
----------------------------------------------------------------------
diff --git a/tests/engine/test_task_graph_into_exececution_graph.py b/tests/engine/test_task_graph_into_exececution_graph.py
new file mode 100644
index 0000000..b087a0f
--- /dev/null
+++ b/tests/engine/test_task_graph_into_exececution_graph.py
@@ -0,0 +1,85 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+import mock
+from networkx import topological_sort
+
+from aria.workflows.engine.engine import Engine
+from aria.workflows.api.tasks_graph import TaskGraph
+from aria.workflows.engine.tasks import (StartWorkflowTask,
+                                         EndWorkflowTask,
+                                         StartSubWorkflowTask,
+                                         EndSubWorkflowTask,
+                                         OperationTask)
+
+from . import MockExecutor, MockOperationContext, MockOperationTask
+
+
+@mock.patch('aria.workflows.engine.engine.OperationTask', new=MockOperationTask)
+def test_task_graph_into_execution_graph():
+    task_graph_name = 'test_task_graph'
+    inner_task_graph_name = 'test_inner_task_graph'
+    simple_before_task_name = 'test_simple_before_task'
+    simple_after_task_name = 'test_simple_after_task'
+    inner_task_name = 'test_inner_task'
+
+    task_graph = TaskGraph(task_graph_name)
+    simple_before_task = MockOperationContext(simple_before_task_name)
+    simple_after_task = MockOperationContext(simple_after_task_name)
+
+    inner_task_graph = TaskGraph(inner_task_graph_name)
+    inner_task = MockOperationContext(inner_task_name)
+    inner_task_graph.add_task(inner_task)
+
+    task_graph.add_task(simple_before_task)
+    task_graph.add_task(simple_after_task)
+    task_graph.add_task(inner_task_graph)
+    task_graph.dependency(inner_task_graph, [simple_before_task])
+    task_graph.dependency(simple_after_task, [inner_task_graph])
+
+    engine = Engine(executor=MockExecutor(),
+                    workflow_context=mock.MagicMock(),
+                    tasks_graph=task_graph)
+
+    # Direct check
+    execution_graph = engine._execution_graph
+    execution_tasks = topological_sort(execution_graph)
+
+    assert len(execution_tasks) == 7
+
+    tasks_names = [
+        'Starting.{0}'.format(task_graph_name),
+        simple_before_task_name,
+        'Starting.{0}'.format(inner_task_graph_name),
+        inner_task_name,
+        'Ending.{0}'.format(inner_task_graph_name),
+        simple_after_task_name,
+        'Ending.{0}'.format(task_graph_name)
+    ]
+
+    assert tasks_names == execution_tasks
+
+    assert isinstance(_get_task_by_name(execution_tasks[0], execution_graph), StartWorkflowTask)
+    assert simple_before_task == _get_task_by_name(execution_tasks[1], execution_graph).context
+    assert isinstance(_get_task_by_name(execution_tasks[2], execution_graph), StartSubWorkflowTask)
+    assert inner_task == _get_task_by_name(execution_tasks[3], execution_graph).context
+    assert isinstance(_get_task_by_name(execution_tasks[4], execution_graph), EndSubWorkflowTask)
+    assert simple_after_task == _get_task_by_name(execution_tasks[5], execution_graph).context
+    assert isinstance(_get_task_by_name(execution_tasks[6], execution_graph), EndWorkflowTask)
+
+
+def _get_task_by_name(task_name, graph):
+    return graph.node[task_name]['task']

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/8d44901f/tests/requirements.txt
----------------------------------------------------------------------
diff --git a/tests/requirements.txt b/tests/requirements.txt
index 07d82a6..f3443d1 100644
--- a/tests/requirements.txt
+++ b/tests/requirements.txt
@@ -4,3 +4,4 @@ tox==1.6.1
 pylint==1.5.5
 pytest==3.0.2
 pytest-cov==2.3.1
+pytest-mock==1.2


Mime
View raw message