Repository: incubator-ariatosca Updated Branches: refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations 55a8830c9 -> db3f9aeaf (forced update)
ARIA-214 Dry execution changes the state of non implemented operations Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/db3f9aea Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/db3f9aea Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/db3f9aea Branch: refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations Commit: db3f9aeaf486bd54b862e00d484c378d19ff9c61 Parents: 1cb3086 Author: max-orlov <ma...@gigaspaces.com> Authored: Sun May 7 16:12:56 2017 +0300 Committer: max-orlov <ma...@gigaspaces.com> Committed: Sun May 7 17:35:00 2017 +0300 ---------------------------------------------------------------------- aria/orchestrator/workflows/core/task.py | 3 --- aria/orchestrator/workflows/core/translation.py | 6 +---- aria/orchestrator/workflows/executor/base.py | 15 ++++++------ aria/orchestrator/workflows/executor/celery.py | 2 ++ aria/orchestrator/workflows/executor/dry.py | 24 ++++++++++---------- aria/orchestrator/workflows/executor/process.py | 3 +++ aria/orchestrator/workflows/executor/thread.py | 2 ++ 7 files changed, 28 insertions(+), 27 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/core/task.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/core/task.py b/aria/orchestrator/workflows/core/task.py index 0e081c2..755a4b9 100644 --- a/aria/orchestrator/workflows/core/task.py +++ b/aria/orchestrator/workflows/core/task.py @@ -163,9 +163,6 @@ class OperationTask(BaseTask): self._task_id = task_model.id self._update_fields = None - def execute(self): - super(OperationTask, self).execute() - @contextmanager def _update(self): """ http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/core/translation.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/core/translation.py b/aria/orchestrator/workflows/core/translation.py index 0bbce90..fec108b 100644 --- a/aria/orchestrator/workflows/core/translation.py +++ b/aria/orchestrator/workflows/core/translation.py @@ -48,11 +48,7 @@ def build_execution_graph( execution_graph, dependencies, default=[start_task]) if isinstance(api_task, api.task.OperationTask): - if api_task.implementation: - operation_task = core_task.OperationTask(api_task, executor=default_executor) - else: - operation_task = core_task.OperationTask(api_task, - executor=base.EmptyOperationExecutor()) + operation_task = core_task.OperationTask(api_task, executor=default_executor) _add_task_and_dependencies(execution_graph, operation_task, operation_dependencies) elif isinstance(api_task, api.task.WorkflowTask): # Build the graph recursively while adding start and end markers http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/executor/base.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/executor/base.py b/aria/orchestrator/workflows/executor/base.py index a225837..a95e58f 100644 --- a/aria/orchestrator/workflows/executor/base.py +++ b/aria/orchestrator/workflows/executor/base.py @@ -31,7 +31,14 @@ class BaseExecutor(logger.LoggerMixin): Execute a task :param task: task to execute """ - raise NotImplementedError + if task.model_task.implementation: + self._execute_empty_task(task) + else: + raise NotImplementedError + + def _execute_empty_task(self, task): + self._task_started(task) + self._task_succeeded(task) def close(self): """ @@ -55,9 +62,3 @@ class BaseExecutor(logger.LoggerMixin): class StubTaskExecutor(BaseExecutor): def execute(self, task): task.status = task.SUCCESS - - -class EmptyOperationExecutor(BaseExecutor): - def execute(self, task): - events.start_task_signal.send(task) - events.on_success_task_signal.send(task) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/executor/celery.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/executor/celery.py b/aria/orchestrator/workflows/executor/celery.py index 7bd9b7c..0406d4e 100644 --- a/aria/orchestrator/workflows/executor/celery.py +++ b/aria/orchestrator/workflows/executor/celery.py @@ -43,6 +43,8 @@ class CeleryExecutor(BaseExecutor): self._started_queue.get(timeout=30) def execute(self, task): + if not task.implementation: + return self._execute_empty_task(task) self._tasks[task.id] = task inputs = dict(inp.unwrap() for inp in task.inputs.values()) inputs['ctx'] = task.context http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/executor/dry.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/executor/dry.py b/aria/orchestrator/workflows/executor/dry.py index eb70a41..5a4ca73 100644 --- a/aria/orchestrator/workflows/executor/dry.py +++ b/aria/orchestrator/workflows/executor/dry.py @@ -25,7 +25,6 @@ class DryExecutor(BaseExecutor): """ Executor which dry runs tasks - prints task information without causing any side effects """ - def execute(self, task): # updating the task manually instead of calling self._task_started(task), # to avoid any side effects raising that event might cause @@ -33,19 +32,20 @@ class DryExecutor(BaseExecutor): task.started_at = datetime.utcnow() task.status = task.STARTED - if hasattr(task.actor, 'source_node'): - name = '{source_node.name}->{target_node.name}'.format( - source_node=task.actor.source_node, target_node=task.actor.target_node) - else: - name = task.actor.name + if task.implementation: + if hasattr(task.actor, 'source_node'): + name = '{source_node.name}->{target_node.name}'.format( + source_node=task.actor.source_node, target_node=task.actor.target_node) + else: + name = task.actor.name - task.context.logger.info( - '<dry> {name} {task.interface_name}.{task.operation_name} started...' - .format(name=name, task=task)) + task.context.logger.info( + '<dry> {name} {task.interface_name}.{task.operation_name} started...' + .format(name=name, task=task)) - task.context.logger.info( - '<dry> {name} {task.interface_name}.{task.operation_name} successful' - .format(name=name, task=task)) + task.context.logger.info( + '<dry> {name} {task.interface_name}.{task.operation_name} successful' + .format(name=name, task=task)) # updating the task manually instead of calling self._task_succeeded(task), # to avoid any side effects raising that event might cause http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/executor/process.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py index 8481406..d8a23b9 100644 --- a/aria/orchestrator/workflows/executor/process.py +++ b/aria/orchestrator/workflows/executor/process.py @@ -118,6 +118,9 @@ class ProcessExecutor(base.BaseExecutor): self._listener_thread.join(timeout=60) def execute(self, task): + if not task.implementation: + return self._execute_empty_task(task) + self._check_closed() self._tasks[task.id] = task http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/db3f9aea/aria/orchestrator/workflows/executor/thread.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/executor/thread.py b/aria/orchestrator/workflows/executor/thread.py index 836b2bf..8e16870 100644 --- a/aria/orchestrator/workflows/executor/thread.py +++ b/aria/orchestrator/workflows/executor/thread.py @@ -47,6 +47,8 @@ class ThreadExecutor(BaseExecutor): self._pool.append(thread) def execute(self, task): + if not task.implementation: + return self._execute_empty_task(task) self._queue.put(task) def close(self):